How to use the zmq.SUBSCRIBE function in zmq

def __call__(self):
        import zmq
        context = zmq.Context()
        sock = context.socket(zmq.SUB)
        sock.setsockopt(zmq.SUBSCRIBE, '')
        sock.connect('tcp://' + self.hostname +':'+str(self.port))
        # Get progress via socket
        percent = None
        while True:
            try:
                message= sock.recv()
                [percent_str, label] = message.split('|')
                percent = float(percent_str)
                self.label = label
                super(RemoteProgress, self).__call__(percent)
            except KeyboardInterrupt:
                if percent is not None:
                    self.message("Exited at %.3f%% completion" % percent)
                break
            except:
                self.message("Could not process socket message: %r"

outr, outw = os.pipe()
    errr, errw = os.pipe()
    t1 = Thread(target=forword, args=[outr, addr1, prefix])
    t1.daemon = True
    t1.start()
    t2 = Thread(target=forword, args=[errr, addr2, prefix])
    t2.daemon = True
    t2.start()
    wout = os.fdopen(outw, 'w', 0)
    werr = os.fdopen(errw, 'w', 0)

    if addr3:
        tid = int(task.task_id.value.split('-')[0])
        subscriber = ctx.socket(zmq.SUB)
        subscriber.connect(addr3)
        subscriber.setsockopt(zmq.SUBSCRIBE, '')
        poller = zmq.Poller()
        poller.register(subscriber, zmq.POLLIN)
        socks = dict(poller.poll(min(tid / 100.0 + 1, 5) * 60 * 1000))
        if socks and socks.get(subscriber) == zmq.POLLIN:
            hosts = pickle.loads(subscriber.recv(zmq.NOBLOCK))
            line = hosts.get(host)
            if line:
                command = line.split(' ')
            else:
                return reply_status(driver, task.task_id, 'TASK_FAILED')
        else:
            return reply_status(driver, task.task_id, 'TASK_FAILED')

    mem = 100
    for r in task.resources:
        if r.name == 'mem':

terminate_topic (str): control channel topic for stopping given service's listener loops and
                                   closing the listener sockets
        """
        self._domains.add(domain_name)
        for topic in sub_topics:
            self._register_sub_topic(service_name, topic)
        for topic in pub_topics:
            self._register_pub_topic(service_name, topic)

        # setup control channels
        self._start_topics.add(start_topic)
        self._end_topics.add(end_topic)
        self._terminate_topics.add(terminate_topic)

        self._control_channel_sub.setsockopt(zmq.SUBSCRIBE, bytes(f"ACK/{start_topic}", encoding="ascii"))
        self._control_channel_sub.setsockopt(zmq.SUBSCRIBE, bytes(f"ACK/{end_topic}", encoding="ascii"))
        self._control_channel_sub.setsockopt(zmq.SUBSCRIBE, bytes(f"ACK/{terminate_topic}", encoding="ascii"))

def subscribe(self, channel):
		"""
		Subscribes to all channels that start with the
		sub-string *channel*.

		:param channel: pattern of channels to subscribe
		:type channel: string
		"""
		self.sub.setsockopt_string(zmq.SUBSCRIBE, channel)

def main(settings):
    """
    Initialize the logger sink.
    """

    if os.path.isfile('logging.conf'):
        logging.config.fileConfig('logging.conf')

    ctx = zmq.Context()
    io_loop = IOLoop.instance()

    log_sub = ctx.socket(zmq.SUB)
    log_sub.setsockopt(zmq.SUBSCRIBE, "")
    log_sub.bind(settings.ZEROMQ_LOGGING)

    log_stream = ZMQStream(log_sub, io_loop)

    log_stream.on_recv(log_zmq_message)

    def handle_shutdown_signal(_sig, _frame):
        """
        Called from the os when a shutdown signal is fired.
        """
        log_stream.stop_on_recv()
        log_stream.flush()
        io_loop.stop()

    # handle kill signals
    signal.signal(signal.SIGINT, handle_shutdown_signal)

:param show_server_config: whether to show server configs when first connected
        :param identity: the UUID of this client
        :param check_version: check if server has the same version as client, raise AttributeError if not the same
        :param check_length: check if server `max_seq_len` is less than the sentence length before sent
        :param timeout: set the timeout (milliseconds) for receive operation on the client, -1 means no timeout and wait until result returns
        """

        self.context = zmq.Context()
        self.sender = self.context.socket(zmq.PUSH)
        self.sender.setsockopt(zmq.LINGER, 0)
        self.identity = identity or str(uuid.uuid4()).encode('ascii')
        self.sender.connect('tcp://%s:%d' % (ip, port))

        self.receiver = self.context.socket(zmq.SUB)
        self.receiver.setsockopt(zmq.LINGER, 0)
        self.receiver.setsockopt(zmq.SUBSCRIBE, self.identity)
        self.receiver.connect('tcp://%s:%d' % (ip, port_out))

        self.request_id = 0
        self.timeout = timeout
        self.pending_request = set()

        if output_fmt == 'ndarray':
            self.formatter = lambda x: x
        elif output_fmt == 'list':
            self.formatter = lambda x: x.tolist()
        else:
            raise AttributeError('"output_fmt" must be "ndarray" or "list"')

        self.output_fmt = output_fmt
        self.port = port
        self.port_out = port_out

def scl_get_socket(id, type_name):
   map = {"sub": zmq.SUB, "req": zmq.REQ, "push": zmq.PUSH,
          "pub": zmq.PUB, "rep": zmq.REP, "pull": zmq.PULL}
   socket_type = map[type_name]
   socket_path = pp_path + id
   socket = context.socket(socket_type)
   if socket_type in [zmq.SUB, zmq.REQ, zmq.PUSH]:
      socket.setsockopt(zmq.RCVHWM, 1)
      if socket_type == zmq.SUB:
         socket.setsockopt(zmq.SUBSCRIBE, "")
      socket.connect(socket_path)
   elif socket_type in [zmq.PUB, zmq.REP, zmq.PULL]:
      socket.setsockopt(zmq.SNDHWM, 1)
      socket.bind(socket_path)
   else:
      raise Exception("unknown socket type: %d" % socket_type)
   return SCL_Socket(socket)

import cmd
import zmq
import json
import threading,Queue

import multiprocessing as m
from copy import deepcopy

from db import store_change,start_db

ctx = zmq.Context()
out_sock = ctx.socket(zmq.PUB)
out_sock.bind('tcp://*:1338')

in_sock = ctx.socket(zmq.SUB)
in_sock.setsockopt(zmq.SUBSCRIBE,'')
in_sock.bind('tcp://*:1337')

poller = zmq.Poller()
poller.register(in_sock,zmq.POLLIN)
Q = Queue.Queue()

def sender():
    start_db()
    last_hashes=[]
    while True:
        msg = Q.get()
        if msg=='quit': break
        print msg

        if msg['hash'] in last_hashes: continue
        last_hashes.append(msg['hash'])

def start_session(self):
        session_id = str(uuid.uuid4())
        ports = self.kernel_manager.get_kernel_ports(self.kernel_id)
        iopub_stream = self.create_connected_stream(ports['iopub_port'], zmq.SUB)
        iopub_stream.socket.setsockopt(zmq.SUBSCRIBE, b'')
        shell_stream = self.create_connected_stream(ports['shell_port'], zmq.XREQ)
        self._sessions[session_id] = dict(
            iopub_stream = iopub_stream,
            shell_stream = shell_stream
        )
        return session_id