How to use the zmq.SUB function in To help you get started, we’ve selected a few zmq examples, based on popular ways it is used in public projects. Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

Enable here

# Set up a debug socket, if address is given.
            debug_pub = ctx.socket(zmq.PUB)
            debug_pub.bind("tcp://127.0.0.1:%s" % debug_port)
            zmq.proxy(xpub, xsub, debug_pub)
        def send_message(pub, msg):
            while True:
                pub.publish(msg)
                time.sleep(.1)
        t = Thread(target=run_proxy, args=(ctx,), daemon=True)
        t.start()
        time.sleep(1)
        test_sub = ctx.socket(zmq.SUB)
        test_sub.connect("tcp://127.0.0.1:%s" % sub_port)
        test_sub.setsockopt(zmq.SUBSCRIBE, b"")
        publisher = Publisher("tcp://127.0.0.1:%s" % port, 'TestPub', ctx=ctx)
        publisher.start()
        name, topic, data = 'TestNode', 'testing', ['Raw', 'this', 'is', 'data']
        msg = Envelope(topic, name, data)
        sender_t = Thread(target=send_message, args=(publisher, msg), daemon=True)
        sender_t.start()
        i = 0
        while i < 10:
            try:
                frames = test_sub.recv_multipart(zmq.NOBLOCK)
            except zmq.error.Again:
                frames = []

def monitor():
      socket = self.context.socket(zmq.SUB)
      socket.setsockopt(zmq.SUBSCRIBE, 'command')
      socket.connect(address)
      while self.running:
        try:
          command = socket.recv().split(' ', 1)[1]
          logging.info("Received command: %s" % command)
          if command == 'shutdown':
            self.running = False
            self.context.term()
            return
          elif command == 'status':
            self.log_status()
        except Exception as e:
          self.error_info(e)
    if address:

Create and bind all sockets
        :return:
            context = zmq.Context(1)
            # Socket to reply to job requests
            self.monitor = context.socket(zmq.REP)
            self.monitor.bind("tcp://%s:%s"%(self.ipaddress,self.localconf['replyport']))
            # Socket to reply to configuration requests
            self.confport = context.socket(zmq.REP)
            self.confport.bind("tcp://%s:%s"%(self.ipaddress,self.localconf['conf_reply']))
            # Socket to push jobs to streamer
            self.pusher = context.socket(zmq.PUSH)
            self.pusher.connect("tcp://%s:%s"%(self.ipaddress,self.localconf['pushport']))
            # Socket to subscribe to subscribe to  slavedrivers status messages
            self.sub_slaved_port = context.socket(zmq.SUB)
            self.sub_slaved_port.bind("tcp://%s:%s"%(self.ipaddress,self.localconf['sd_subport']))
            # Initialize poll set to listen on multiple channels at once
            self.poller = zmq.Poller()
            self.poller.register(self.monitor, zmq.POLLIN)
            self.poller.register(self.confport, zmq.POLLIN|zmq.POLLOUT)
            self.poller.register(self.sub_slaved_port, zmq.POLLIN)
        except KeyboardInterrupt:
            log.info('Bringing down Manager')
#        finally:

def initialize(self):
        self._context = zmq.Context()
        self._sock = self._context.socket(zmq.SUB)
        self._sock.set_hwm(BROADCAST_HWM)
        self._sock.connect(self._conn_info)
        self._sock.setsockopt(zmq.SUBSCRIBE, b'')

def __init__(self, name, ip, port = 5556, hwm=10, return_last=True):
        context = zmq.Context()
        self.socket = context.socket(zmq.SUB)
        self.socket.set_hwm(hwm)
        self.socket.connect("tcp://%s:%d" % (ip, port))
        self.socket.setsockopt_string(zmq.SUBSCRIBE, '')
        self.name = name
        self.return_last = return_last
        self.last = None

def main(topics, addrs):
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    for topic in topics:
        print "Subscribing to: %r"%topic
        socket.setsockopt(zmq.SUBSCRIBE, topic)
    if addrs:
        for addr in addrs:
            print "Connecting to: ", addr
            socket.connect(addr)
    else:
        socket.bind('tcp://*:%i'%logport)
    while True:
        # topic = socket.recv()
        # print topic
        # print 'tic'
        raw = socket.recv_multipart()
        if len(raw) != 2:

messages containing the string INSERT because it is expected to capture
        MySQL queries.
        Args:
            host (str): Publisher server IP.
            port (str): Listening port.
        if host is None and port is None:
            self.config = load_config('subscriber')
            # TODO (simon): do something if self.config is None
            host = self.config['server_ip']
            port = self.config['listening_port']
        self.server = "tcp://{:s}:{:s}".format(host, port)
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.SUB)
        self.socket.connect(self.server)
        # limit subscription to INSERT queries

def connect(self, key):
        # Log debug message.
        string = "{} establishes connections"
        message = string.format(self.name)
        self.log.debug(message)
        self.get_input(key).socket = self.context.socket(zmq.SUB)
        self.get_input(key).socket.setsockopt(zmq.RCVTIMEO, self._timeout)
        self.get_input(key).socket.connect(self.get_input(key).addr)
        self.get_input(key).socket.setsockopt_string(zmq.SUBSCRIBE, unicode(""))
        return

def __init__(self, room):
        import zmq
        self.zmq = zmq
        context = zmq.Context()
        self.room = room
        # 用来接受info更新
        self.suber = context.socket(zmq.SUB)
        self.suber.connect('ipc:///tmp/game_puber.ipc')
        self.suber.setsockopt(zmq.SUBSCRIBE, 'room:%d '%room)
        # 用来与服务器交互
        self.oper = context.socket(zmq.REQ)
        self.oper.connect('ipc:///tmp/game_oper.ipc')
        # poller
        self.poller = zmq.Poller()
        self.poller.register(self.suber, zmq.POLLIN)

def run(self):
    print "StreamSubscriber.run(): [pid: {}, OS pid: {}]".format(self.pid, os.getpid())
    # * Build ZMQ socket and connect to publisher
    self.context = zmq.Context()
    self.socket = self.context.socket(zmq.SUB)
    self.server_connect_addr = "{protocol}://{host}:{port}".format(
        protocol=server_protocol,
        host=server_host,
        port=server_port)
    self.socket.connect(self.server_connect_addr)
    self.socket.setsockopt(zmq.SUBSCRIBE, "")  # subscribe to all topics
    print "StreamSubscriber.run(): Subscribed to {}".format(self.server_connect_addr)
    # * Keep receiving and displaying images until stopped or null image
    print "StreamSubscriber.run(): Starting display loop [Esc or Q on image, or Ctrl+C on terminal to quit]..."
    self.isOkay = True
    self.image = None
    self.meta = None
    self.lastImageId = self.imageId = -1
    while self.isOkay:
      try: