def zmq_pub_mon(endpoints, counter):
 ctx = zmq.Context.instance()
 poller = zmq.Poller()
 for (name, uri) in endpoints.iteritems():
  logging.info('Registering %s (%s).', name, uri)
  sock = ctx.socket(zmq.SUB)
  sock.setsockopt(zmq.IDENTITY, name)
  sock.connect(uri)
  sock.setsockopt(zmq.SUBSCRIBE, '')
  poller.register(sock, zmq.POLLIN)
 while 1:
  try:
   for (socket, _) in poller.poll():
    socket.recv(zmq.NOBLOCK)
    name = socket.getsockopt(zmq.IDENTITY)
    counter[name] += 1
  except zmq.ZMQError as e:
   if (e.errno == errno.EINTR):
    continue
   raise

def zmq_pub_mon ( endpoints , counter ) : ctx = zmq . Context . instance ( ) poller = zmq . Poller ( ) for ( name , uri ) in endpoints . iteritems ( ) : logging . info ( 'Registering %s (%s).' , name , uri ) sock = ctx . socket ( zmq . SUB ) sock . setsockopt ( zmq . IDENTITY , name ) sock . connect ( uri ) sock . setsockopt ( zmq . SUBSCRIBE , '' ) poller . register ( sock , zmq . POLLIN ) while 1 : try : for ( socket , _ ) in poller . poll ( ) : socket . recv ( zmq . NOBLOCK ) name = socket . getsockopt ( zmq . IDENTITY ) counter [ name ] += 1 except zmq . ZMQError as e : if ( e . errno == errno . EINTR ) : continue raise
measure throughput of zeromq publishers .