def create_monitor(address, socketio):
    def create_poller():
        context = zmq.Context()
        socket = context.socket(zmq.PULL)
        socket.bind(address)

        sockets = {
            "master": {"socket": socket, "receive": socket.recv, "send": socket.send},
        }
        time_func = time.time

        return Poller(sockets, time_func)

    def emit(message):
        socketio.emit("status_update", message)

    scale_workers = lambda command: True
    run_forever = lambda: True

    return Monitor(create_poller, emit, scale_workers, run_forever) 

def start(self, start_dispatching=True):
        self.context = zmq.Context()

        for rfcc in self.rpc_fanout_clients_channels:
            rfcc.connect(self.context)

        for rpcc in self.rpc_server_channels.values():
            rpcc.connect(self.context)

        for sc in self.sub_channels:
            sc.connect()

        for mwsc in self.mw_sub_channels:
            mwsc.connect(self.context)

        for pc in self.pub_channels:
            pc.connect()

        for mwpc in self.mw_pub_channels:
            mwpc.connect(self.context)

        if start_dispatching:
            self.start_dispatching() 

def tearDown(self):
        contexts = set([self.context])
        while self.sockets:
            sock = self.sockets.pop()
            contexts.add(sock.context) # in case additional contexts are created
            sock.close(0)
        for ctx in contexts:
            t = Thread(target=ctx.term)
            t.daemon = True
            t.start()
            t.join(timeout=2)
            if t.is_alive():
                # reset Context.instance, so the failure to term doesn't corrupt subsequent tests
                zmq.sugar.context.Context._instance = None
                raise RuntimeError("context could not terminate, open sockets likely remain in test")
        super(BaseZMQTestCase, self).tearDown() 

def init(self):
        self.log.debug("init zeromq ..")
        if not zmq:
            self.log.critical("missing zeromq, please install pyzmq to use this plugin")
            raise RuntimeError("zeromq python module not found")

        self.ctx = zmq.Context()

        # sanity check config
        if 'bind' in self.config:
            if 'connect' in self.config:
                msg = "bind and connect are mutually exclusive"
                self.log.critical(msg)
                raise ValueError(msg)

        elif 'connect' not in self.config:
            msg = "missing bind or connect"
            self.log.critical(msg)
            raise ValueError(msg) 

def tearDown(self):
        contexts = set([self.context])
        while self.sockets:
            sock = self.sockets.pop()
            contexts.add(sock.context) # in case additional contexts are created
            sock.close(0)
        for ctx in contexts:
            t = Thread(target=ctx.term)
            t.daemon = True
            t.start()
            t.join(timeout=2)
            if t.is_alive():
                # reset Context.instance, so the failure to term doesn't corrupt subsequent tests
                zmq.sugar.context.Context._instance = None
                raise RuntimeError("context could not terminate, open sockets likely remain in test")
        super(BaseZMQTestCase, self).tearDown() 

def setUp(self):
        context = zmq.Context()
        worker_socket = context.socket(zmq.REP)
        worker_socket.bind("ipc:///tmp/worker")
        frontend_socket = context.socket(zmq.REP)
        frontend_socket.bind("ipc:///tmp/frontend")

        sockets = {
            "worker": {"socket": worker_socket, "receive": worker_socket.recv_json, "send": worker_socket.send_json},
            "frontend": {"socket": frontend_socket, "receive": frontend_socket.recv_json, "send": worker_socket.send_json},
        }
        time = TimeStub()

        self.poller = Poller(sockets, time) 

def send_message(self, address, message):
        context = zmq.Context()
        socket = context.socket(zmq.REQ)
        socket.connect(address)
        socket.send_json(message)

        import gevent
        gevent.sleep(0.05)

        return socket 

def create_frontend_worker(master_address):
    context = zmq.Context()
    master_socket = context.socket(zmq.REQ)
    master_socket.connect(master_address)
    worker_socket = context.socket(zmq.REQ)
    decoder = Decoder()
    id_generator = lambda: uuid.uuid4().int

    return FrontendWorker(master_socket, worker_socket, decoder, id_generator) 

def create_recordings_saver(address, model):
    def create_socket():
        context = zmq.Context()
        socket = context.socket(zmq.PULL)
        socket.bind(address)

        return socket

    run_forever = lambda: True

    return RecordingsSaver(create_socket, model, run_forever) 

def context(self):
        if self._context is None:
            if Thread.__module__.startswith('gevent'):
                # gevent has monkey-patched Thread, use green Context
                from zmq import green
                self._context = green.Context()
            else:
                self._context = zmq.Context()
        return self._context 

def start(self):
        """Start a new garbage collection thread.
        
        Creates a new zmq Context used for garbage collection.
        Under most circumstances, this will only be called once per process.
        """
        if self.thread is not None and self.pid != getpid():
            # It's re-starting, must free earlier thread's context
            # since a fork probably broke it
            self._stop()
        self.pid = getpid()
        self.refs = {}
        self.thread = GarbageCollectorThread(self)
        self.thread.start()
        self.thread.ready.wait() 

def Context(self):
        if self.green:
            return gzmq.Context
        else:
            return zmq.Context 

def setUp(self):
        super(BaseZMQTestCase, self).setUp()
        if self.green and not have_gevent:
                raise SkipTest("requires gevent")
        self.context = self.Context.instance()
        self.sockets = [] 

def __init__(self):
        self.context = zmq.Context()
        self.XSUB_URL = ait.config.get('server.xsub',
                                        ait.SERVER_DEFAULT_XSUB_URL)
        self.XPUB_URL = ait.config.get('server.xpub',
                                        ait.SERVER_DEFAULT_XPUB_URL)

        ## Name of the topic associated with external commands
        self.command_topic = ait.config.get('command.topic',
                                        ait.DEFAULT_CMD_TOPIC)

        gevent.Greenlet.__init__(self) 

def context(self):
        if self._context is None:
            if Thread.__module__.startswith('gevent'):
                # gevent has monkey-patched Thread, use green Context
                from zmq import green
                self._context = green.Context()
            else:
                self._context = zmq.Context()
        return self._context 

def start(self):
        """Start a new garbage collection thread.
        
        Creates a new zmq Context used for garbage collection.
        Under most circumstances, this will only be called once per process.
        """
        if self.thread is not None and self.pid != getpid():
            # It's re-starting, must free earlier thread's context
            # since a fork probably broke it
            self._stop()
        self.pid = getpid()
        self.refs = {}
        self.thread = GarbageCollectorThread(self)
        self.thread.start()
        self.thread.ready.wait() 

def Context(self):
        if self.green:
            return gzmq.Context
        else:
            return zmq.Context 

def setUp(self):
        super(BaseZMQTestCase, self).setUp()
        if self.green and not have_gevent:
                raise SkipTest("requires gevent")
        self.context = self.Context.instance()
        self.sockets = [] 

def context(self):
        if self._context is None:
            if Thread.__module__.startswith('gevent'):
                # gevent has monkey-patched Thread, use green Context
                from zmq import green
                self._context = green.Context()
            else:
                self._context = zmq.Context()
        return self._context 

def start(self):
        """Start a new garbage collection thread.
        
        Creates a new zmq Context used for garbage collection.
        Under most circumstances, this will only be called once per process.
        """
        if self.thread is not None and self.pid != getpid():
            # It's re-starting, must free earlier thread's context
            # since a fork probably broke it
            self._stop()
        self.pid = getpid()
        self.refs = {}
        self.thread = GarbageCollectorThread(self)
        self.thread.start()
        self.thread.ready.wait()