def main(pat):
    
    fname = find_connection_file(pat)
    with open(fname) as f:
        cfg = json.load(f)
    
    url = "%s://%s:%s" % (cfg.get('transport', 'tcp'), cfg['ip'], cfg['iopub_port'])
    
    session = Session(key=cfg['key'])
    
    ctx = zmq.Context.instance()
    sub = ctx.socket(zmq.SUB)
    sub.subscribe = b''
    sub.connect(url)
    # import IPython
    # IPython.embed()
    # return
    
    stream = ZMQStream(sub)
    
    stream.on_recv(lambda msg_list: log_msg(session, msg_list))
    
    pc = PeriodicCallback(print_time, 5 * 60 * 1000)
    pc.start()
    IOLoop.instance().start()

def device_manager(context):

    global manager
    global publisher

    worker_address = Config.get('services', 'worker_address')
    worker_port = Config.get('services', 'worker_port')
    worker_address = "tcp://" + worker_address + ":" + worker_port

    publisher_address = Config.get('device_service', 'publisher_address')
    publisher_port = Config.get('device_service', 'publisher_port')
    publisher_address = "tcp://" + publisher_address + ":" + publisher_port

    service = Config.get('device_service', 'service_name')

    publisher = DeviceServicePublisher(context, publisher_address)
    manager = DeviceServiceManager(context, worker_address, service)

    try:
        IOLoop.instance().start()
    except KeyboardInterrupt:
        IOLoop.instance().stop()
        publisher.shutdown()
        manager.shutdown()

    return

    def _setup_pipe_in(self):
        """setup listening pipe for subprocesses"""
        ctx = self.pub_socket.context

        # use UUID to authenticate pipe messages
        self._pipe_uuid = uuid.uuid4().bytes

        self._pipe_in = ctx.socket(zmq.PULL)
        self._pipe_in.linger = 0
        try:
            self._pipe_port = self._pipe_in.bind_to_random_port("tcp://127.0.0.1")
        except zmq.ZMQError as e:
            warn("Couldn't bind IOStream to 127.0.0.1: %s" % e +
                "\nsubprocess output will be unavailable."
            )
            self._pipe_flag = False
            self._pipe_in.close()
            del self._pipe_in
            return
        self._pipe_poller = zmq.Poller()
        self._pipe_poller.register(self._pipe_in, zmq.POLLIN)
        if IOLoop.initialized():
            # subprocess flush should trigger flush
            # if kernel is idle
            IOLoop.instance().add_handler(self._pipe_in,
                lambda s, event: self.flush(),
                IOLoop.READ,
            )

    def ready(self, ioloopstart=1):
        """ to call at the end of the __init__ of classes that inherits of XplPlugin
        """
        if self.dont_run_ready == True:
            return

        ### activate xpl hbeat
        if self.enable_hbeat_called == True:
            self.log.error(u"in ready() : enable_hbeat() function already called : the plugin may not be fully converted to the 0.4+ Domogik format")
        else:
            self.enable_hbeat()

        # send the status for the xpl hbeat
        self.myxpl.update_status(2)

        ### send plugin status : STATUS_ALIVE
        # TODO : why the dbmgr has no self._name defined ???????
        # temporary set as unknown to avoir blocking bugs
        if not hasattr(self, '_name'):
            self._name = "unknown"
        self._set_status(STATUS_ALIVE)

        ### Instantiate the MQ
        # nothing can be launched after this line (blocking call!!!!)
        self.log.info(u"Start IOLoop for MQ : nothing else can be executed in the __init__ after this! Make sure that the self.ready() call is the last line of your init!!!!")
        if ioloopstart == 1:
            IOLoop.instance().start()

    def on_message(self, msgid, content):
        """ Handle MQ messages
            @param msgid : message id
            @content : message content
        """
        if msgid == "plugin.status":
            # we may miss starting and stop-request events but we only want to do some checks on alive and stopped...
            # and sometimes it happens that we still receive a last 'alive' status before the 'stop' one
            if self.count == 0:
                print(u"Message skipped (we skip the first one) : msgid={0}, content={1}".format(msgid, content))
                self.count = 1
                return 

            print(u"Message received : msgid={0}, content={1}".format(msgid, content))
            if content['name'] == self.name and \
               content['type'] == self.type and \
               content['host'] == self.host:
                self.plugin_status = content['event']
                # plugin started
                if content['event'] == STATUS_ALIVE:
                    print(u"Plugin is started")
                    print(u"Stop listening to MQ as we get our result")
                    IOLoop.instance().stop() 
    
                # plugin stopped
                elif content['event'] == STATUS_STOPPED:
                    print(u"Plugin is stopped")
                    print(u"Stop listening to MQ as we get our result")
                    IOLoop.instance().stop() 

def main():
    """main method"""

    url_worker = "ipc://backend.ipc"
    url_client = "ipc://frontend.ipc"

    # Prepare our context and sockets
    context = zmq.Context()
    frontend = context.socket(zmq.ROUTER)
    frontend.bind(url_client)
    backend = context.socket(zmq.ROUTER)
    backend.bind(url_worker)

    # create workers and clients threads
    for i in range(NBR_WORKERS):
        thread = threading.Thread(target=worker_thread, args=(url_worker, i, ))
        thread.daemon = True
        thread.start()

    for i in range(NBR_CLIENTS):
        thread_c = threading.Thread(target=client_thread, args=(url_client, i, ))
        thread_c.daemon = True
        thread_c.start()

    # create queue with the sockets
    queue = LRUQueue(backend, frontend)

    # start reactor
    IOLoop.instance().start()

def main():
    http_fetcher = fetcher.HTTPFetcher()
    rb = robot.Robot(http_fetcher, [
        ('/.*', AHandler),
    ])
    worker.Worker(rb).start()
    IOLoop.instance().start()

 def on_message(self, msg):
     if _do_print:
         print 'client received:',
         pprint(msg)
     self.last_msg = msg
     IOLoop.instance().stop()
     return

    def send_to_butler(self, command_line, widget_proxy, live):
        you = u"You > {0}".format(command_line)
        self.parent.value.set_values(self.parent.value.get() + [you])
        self.parent.wMain.values = self.parent.value.get()
        self.parent.wMain.display()

        # handle special commands
        if command_line.lower() == "quit":
            IOLoop.instance().stop()
            sys.exit(0)  # why is this needed ?

        elif command_line.lower() == "reload":
            try:
                cli = MQSyncReq(zmq.Context())
                msg = MQMessage()
                msg.set_action('butler.reload.do')
                result = cli.request('butler', msg.get(), timeout=10).get()
                if result:
                    msg = "*** action reload : {0}".format(result)
                    self.parent.value.set_values(self.parent.value.get() + [msg])
            except:
                msg = u"*** action reload : error (is butler component ok ?)"
                self.parent.value.set_values(self.parent.value.get() + [msg])

        # handle butler
        else:
            self.parent.butler_cb(command_line, identity = "cli user", media = "chat", location = None, mood = None)

        self.parent.wMain.values = self.parent.value.get()
        self.parent.wMain.display()

def subscriber_manager(context):

    global subscriber
    global publisher
    global subscriber_topics

    session = Session()

    # Set up forwarder device
    logger.debug("Configuring forwarder device")
    forwarder_subscriber_address = session.query(RoutingDefiniton.route) \
                                          .filter_by(name="internal_pub") \
                                          .scalar()
    forwarder_publisher_address = session.query(RoutingDefiniton.route) \
                                         .filter_by(name="internal_sub") \
                                         .scalar()

    forwarder = ProcessDevice(zmq.FORWARDER, zmq.SUB, zmq.PUB)
    forwarder.connect_in(forwarder_subscriber_address)
    forwarder.setsockopt_in(zmq.SUBSCRIBE, "")
    logger.debug("forwarder connect in: {0}".format(forwarder_subscriber_address))

    forwarder.bind_out(forwarder_publisher_address)
    logger.debug("forwarder bind out: {0}".format(forwarder_publisher_address))

    # Set up subscriber listening to Farm-Monitor server
    subscriber_address = session.query(RoutingDefiniton.route)\
                                .filter_by(name="subscriber_address").scalar()
    subscriber_port = session.query(RoutingDefiniton.route)\
                             .filter_by(name="subscriber_port").scalar()
    subscriber_address = "tcp://" + subscriber_address + ":" + subscriber_port
    device_id = session.query(HardwareDefinition.serial_number).scalar()
    subscriber_topics.append(device_id)

    subscriber = Subscriber(context, subscriber_address, subscriber_topics)
    subscriber.subscriber.on_recv(callback=server_subscriber_recv)
    logger.debug("Subscriber listening to: {0} topics: {1}".format(subscriber_address,
                                                                   subscriber_topics))

    # Set up publisher to forward messages from Farm-Monitor to internal
    publisher_address = session.query(RoutingDefiniton.route)\
                               .filter_by(name="internal_pub").scalar()
    publisher = Publisher(context, publisher_address)
    logger.debug("Publisher configured to: {0}".format(publisher_address))

    session.close()
    try:
        logger.info("starting forwarder, Subscriber, and Publisher")
        # Start the forwarder
        forwarder.start()
        IOLoop.instance().start()

    except KeyboardInterrupt:
        logger.info("stopping forwarder, Subscriber, and Publisher")
        IOLoop.instance().stop()
        subscriber.shutdown()
        publisher.shutdown()

    return

 def test_03_timeout_01(self):
     """Test MDPclient request w/ timeout.
     """
     client = MyClient(self.context, self.endpoint, self.service)
     client.request(b'XXX', 20) # 20 millisecs timeout
     IOLoop.instance().start()
     client.shutdown()
     self.assertEquals(client.timed_out, True)
     return

class AsyncServerAdapter(object):
    producer_class = Producer

    def __init__(self, backend_rep_uri, frontend_rep_uri, frontend_pub_uri,
                 control_pipe=None):
        self.uris = OrderedDict([
            ('backend_rep', backend_rep_uri),
            ('consumer_push_be', unique_ipc_uri()),
            ('consumer_pull_be', unique_ipc_uri()),
            ('frontend_rep_uri', frontend_rep_uri),
            ('frontend_pub_uri', frontend_pub_uri)
        ])
        self.control_pipe = control_pipe
        self.done = False
        logging.getLogger(log_label(self)).info("uris: %s", self.uris)

    def watchdog(self):
        if self.control_pipe is None:
            return
        elif not self.done and self.control_pipe.poll():
            self.done = True
            self.finish()

    def run(self):
        consumer = Process(target=Consumer(self.uris['backend_rep'],
                                        self.uris['consumer_push_be'],
                                        self.uris['consumer_pull_be']).run
        )
        producer = Process(target=self.producer_class(
                self.uris['frontend_rep_uri'],
                self.uris['frontend_pub_uri'],
                self.uris['consumer_pull_be'],
                self.uris['consumer_push_be']).run
        )
        self.io_loop = IOLoop()
        periodic_callback = PeriodicCallback(self.watchdog, 500, self.io_loop)
        periodic_callback.start()
        try:
            consumer.start()
            producer.start()
            self.io_loop.start()
        except KeyboardInterrupt:
            pass
        producer.terminate()
        consumer.terminate()
        logging.getLogger(log_label(self)).info('PRODUCER and CONSUMER have '
                                                'been terminated')

    def __del__(self):
        uris = [self.uris[label] for label in ('consumer_push_be',
                                               'consumer_pull_be', )]
        cleanup_ipc_uris(uris)

    def finish(self):
        logging.getLogger(log_label(self)).debug('"finish" request received')
        self.io_loop.stop()

    def _schedule_flush(self):
        """schedule a flush in the main thread

        only works with a tornado/pyzmq eventloop running
        """
        if IOLoop.initialized():
            IOLoop.instance().add_callback(self.flush)
        else:
            # no async loop, at least force the timer
            self._start = 0

def main():
    from frontier import Frontier
    from options import parse_command_line

    parse_command_line()

    ft = Frontier([
        ('http://localhost/', 1),
    ])
    Master(ft).start()
    IOLoop.instance().start()

 def _on_msg(self, msg):
     self._msgs.append(msg)
     if _do_print:
         print 'broker received:',
         pprint(msg)
     if self.broker.do_reply:
         new_msg = msg[:4]
         new_msg.append(b'REPLY')
         self.broker.send_multipart(new_msg)
     else:
         IOLoop.instance().stop()
     return

def main():
    cfg = Loader('mq')
    my_conf = cfg.load()
    config = dict(my_conf[1])

    ctx = DaemonContext()
    ctx.open()

    context = zmq.Context()
    print(("tcp://{0}:{1}".format(config['ip'], config['req_rep_port'])))
    broker = MDPBroker(context, "tcp://{0}:{1}".format(config['ip'], config['req_rep_port']))
    IOLoop.instance().start()
    broker.shutdown()

def main():
    context = zmq.Context(1)
    
    #qq = q.TheQueue(context, CLIENT_URL, WORKER_URL)
    qq = q.TheQueue(context, CLIENT_URL, WORKER_URL)
    for i in range(2):
        t = threading.Thread(target=w.TheWorker,args=(context,))
        t.start()
        #t.join()
    try:
        IOLoop.instance().start()
    except KeyboardInterrupt:
        times_str('ctrlc')

def main():
    ft = frontier.Frontier([
        ('http://m.sohu.com/', 1),
    ])
    http_fetcher = fetcher.HTTPFetcher()
    rb = robot.Robot(http_fetcher, [
        ('/.*', AHandler),
    ])
    id = fork_processes(0)
    if id == master_id():
        master.Master(ft).start()
    else:
        worker.Worker(rb).start()
    IOLoop.instance().start()

 def test_01_simple_01(self):
     """Test MDPWorker simple req/reply.
     """
     self._start_broker()
     time.sleep(0.2)
     worker = MyWorker(self.context, self.endpoint, self.service)
     sender = DelayedCallback(self.send_req, 500)
     stopper = DelayedCallback(self.stop_test, 2500)
     sender.start()
     stopper.start()
     IOLoop.instance().start()
     worker.shutdown()
     self._stop_broker()
     return

 def test_04_receive_01(self):
     """Test MDPclient message receive.
     """
     self._start_broker(do_reply=True)
     client = MyClient(self.context, self.endpoint, self.service)
     client.request(b'XXX')
     IOLoop.instance().start()
     client.shutdown()
     self._stop_broker()
     self.assertEquals(True, hasattr(client, b'last_msg'))
     self.assertEquals(3, len(client.last_msg))
     self.assertEquals(b'REPLY', client.last_msg[-1])
     self.assertEquals(self.service, client.last_msg[-2])
     return