def controler_socket(self):
        msg = self.controler.recv_multipart()
        try:
            master_id = u(msg[0])
            action = u(msg[1])
            ping_interval = int(msg[2])

            if master_id != "master":
                self.logger.error("Invalid master id '%s'. Should be 'master'",
                                  master_id)
                return
            if action != "PONG":
                self.logger.error("Invalid answer '%s'. Should be 'PONG'",
                                  action)
                return
        except (IndexError, ValueError):
            self.logger.error("Invalid message '%s'", msg)
            return

        if ping_interval < TIMEOUT:
            self.logger.error("invalid ping interval (%d) too small", ping_interval)
            return

        self.logger.debug("master => PONG(%d)", ping_interval)
        self.ping_interval = ping_interval

    def controler_socket(self):
        try:
            # We need here to use the zmq.NOBLOCK flag, otherwise we could block
            # the whole main loop where this function is called.
            msg = self.controler.recv_multipart(zmq.NOBLOCK)
        except zmq.error.Again:
            return False
        # This is way to verbose for production and should only be activated
        # by (and for) developers
        # self.logger.debug("[CC] Receiving: %s", msg)

        # 1: the hostname (see ZMQ documentation)
        hostname = u(msg[0])
        # 2: the action
        action = u(msg[1])

        # Check that lava-logs only send PINGs
        if hostname == "lava-logs" and action != "PING":
            self.logger.error("%s => %s Invalid action from log daemon",
                              hostname, action)
            return True

        # Handle the actions
        if action == 'HELLO' or action == 'HELLO_RETRY':
            self._handle_hello(hostname, action, msg)
        elif action == 'PING':
            self._handle_ping(hostname, action, msg)
        elif action == 'END':
            self._handle_end(hostname, action, msg)
        elif action == 'START_OK':
            self._handle_start_ok(hostname, action, msg)
        else:
            self.logger.error("<%s> sent unknown action=%s, args=(%s)",
                              hostname, action, msg[1:])
        return True

def write_key(pub_key):
    banner = u("""#   ****  Generated on {0} by tsg  ****
#   ZeroMQ CURVE Public Certificate
#   Exchange securely, or use a secure mechanism to verify the contents
#   of this file after exchange. Store public certificates in your home
#   directory, in the .curve subdirectory.
""")

    base_dir = os.path.dirname(__file__)
    public_keys_dir = os.path.join(base_dir, 'public_keys')
    key = pub_key.decode('utf-8')
    with io.open(
        public_keys_dir+"/"+str(hashlib.sha1(pub_key).hexdigest())
            + ".key", 'w', encoding='utf8') as f:

        f.write(banner.format(datetime.datetime.now()))
        f.write(u('metadata\n'))
        f.write(u('curve\n'))
        f.write(u("    public-key = \"{0}\"\n").format(pub_key))

    #If it's the first client we start the secure server
    if not p.is_alive():
        p.start()
    #stop secure_srv process and restart it to be able to auth new clients
    q_mgmt.put(ThMgmt(0))
    print "reauth sended"

 def test_unicode(self):
     """Test the unicode representations of the Frames."""
     s = u('asdf')
     self.assertRaises(TypeError, zmq.Frame, s)
     for i in range(16):
         s = (2**i)*u('§')
         m = zmq.Frame(s.encode('utf8'))
         self.assertEqual(s, unicode(m.bytes,'utf8'))

    def _handle_pipe(self):
        """
        Handle a message from front-end API.
        """
        terminate = False

        # Get the whole message off the pipe in one go
        try:
            # Try/except needed for Windows "support"
            msg = self.pipe.recv_multipart()

            if msg is None:
                terminate = True
                return terminate
        except:
            terminate = True
            return terminate

        command = msg[0]
        self.log.debug("auth received API command %r", command)

        if command == b'ALLOW':
            addresses = [u(m, self.encoding) for m in msg[1:]]
            try:
                self.authenticator.allow(*addresses)
            except Exception as e:
                self.log.exception("Failed to allow %s", addresses)
            self.allow = True

        elif command == b'DENY':
            addresses = [u(m, self.encoding) for m in msg[1:]]
            try:
                self.authenticator.deny(*addresses)
            except Exception as e:
                self.log.exception("Failed to deny %s", addresses)

        elif command == b'PLAIN':
            domain = u(msg[1], self.encoding)
            json_passwords = msg[2]
            self.authenticator.configure_plain(domain, jsonapi.loads(json_passwords))

        elif command == b'CURVE':
            # For now we don't do anything with domains
            domain = u(msg[1], self.encoding)

            # If location is CURVE_ALLOW_ANY, allow all clients. Otherwise
            # treat location as a directory that holds the certificates.
            location = u(msg[2], self.encoding)
            self.authenticator.configure_curve(domain, location)
            self.curve = True

        elif command == b'TERMINATE':
            terminate = True

        else:
            self.log.error("Invalid auth command from API: %r", command)

        return terminate

 def test_unicode_message(self):
     logger, handler, sub = self.connect_handler()
     base_topic = b(self.topic + '.INFO')
     for msg, expected in [
         (u('hello'), [base_topic, b('hello\n')]),
         (u('héllo'), [base_topic, b('héllo\n')]),
         (u('tøpic::héllo'), [base_topic + b('.tøpic'), b('héllo\n')]),
     ]:
         logger.info(msg)
         received = sub.recv_multipart()
         self.assertEqual(received, expected)

    def messageReceived(self, msg):

        command = msg[0]

        if command == b'ALLOW':
            addresses = [u(m, self.encoding) for m in msg[1:]]
            try:
                self.authenticator.allow(*addresses)
            except Exception as e:
                log.err("Failed to allow %s", addresses)

        elif command == b'CURVE':
            domain = u(msg[1], self.encoding)
            location = u(msg[2], self.encoding)
            self.authenticator.configure_curve(domain, location)

    def run(self):
        self.setup()

        max_db_commit_retry = 3
        while True:
            msg = self.sub.recv_multipart()
            try:
                (topic, uuid, dt, username, data) = (u(m) for m in msg)
                dt = datetime.datetime.strptime(dt, "%Y-%m-%dT%H:%M:%S.%f")
            except (IndexError, ValueError):
                LOG.error("Invalid message: %s", msg)
                continue

            # Save into the database
            try:
                session = self.sessions()
                message = Message(topic=topic, uuid=uuid, datetime=dt, username=username, data=data)
                session.add(message)
            except SQLAlchemyError as err:
                LOG.erro("Unable to build the new message row: %s", err)
                continue

            # Retry the database commit
            for retry in range(1, max_db_commit_retry + 1):
                try:
                    session.commit()
                except SQLAlchemyError as err:
                    if retry == max_db_commit_retry:
                        LOG.error("Unable to commit to the database, dropping the message")
                        LOG.error("Database error: %s", err)

 def handle_recv(self, data):
     topic, msg = data
     topic_parts = u(topic).split(".")
     watcher = topic_parts[1]
     action = topic_parts[2]
     with open(self.config['file'], 'a+') as f:
         f.write('%s:%s' % (watcher, action))

    def handle_recv(self, data):
        """called each time circusd sends an event"""
        # maintains a periodic callback to compute mem and cpu consumption for
        # each pid.
        logger.debug('Received an event from circusd: %s' % str(data))
        topic, msg = data
        try:
            topic = u(topic)
            watcher = topic.split('.')[1:-1][0]
            action = topic.split('.')[-1]
            msg = json.loads(msg)

            if action in ('reap', 'kill'):
                # a process was reaped
                pid = msg['process_pid']
                self.remove_pid(watcher, pid)
            elif action == 'spawn':
                # a process was added
                pid = msg['process_pid']
                self._append_pid(watcher, pid)
            elif action == 'stop':
                # the whole watcher was stopped.
                self.stop_watcher(watcher)
            else:
                logger.debug('Unknown action: %r' % action)
                logger.debug(msg)
        except Exception:
            logger.exception('Failed to handle %r' % msg)

    def listen(self):
        listener_url = self.get_listener_url()

        self.log_debug("connecting to %s" % listener_url)

        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.SUB)
        self.socket.setsockopt_string(zmq.SUBSCRIBE, "")
        try:
            # requires PyZMQ to be built against ZeroMQ 4.2+
            self.socket.setsockopt(zmq.HEARTBEAT_IVL, 1000)  # 1 s
            self.socket.setsockopt(zmq.HEARTBEAT_TIMEOUT, 10000)  # 10 s
        except AttributeError:
            self.log_warn('PyZMQ has no support for heartbeat (requires ZeroMQ library 4.2+), connection may be unstable')
            pass

        self.socket.connect(listener_url)

        self.log_debug("connected to %s" % listener_url)

        while True:
            try:
                message = self.socket.recv_multipart()
                (topic, uuid, dt, username, data) = (u(m) for m in message[:])
                data = json.loads(data)
                self.receive_event(topic, data)
            except Exception as e:
                self.log_error(str(e) + "\n" + traceback.format_exc())

    def iter_messages(self):
        """ Yields tuples of (watcher, subtopic, stat)"""
        recv = self.pubsub_socket.recv_multipart
        with self:
            while True:
                try:
                    events = dict(self.poller.poll(self.timeout * 1000))
                except zmq.ZMQError as e:
                    if e.errno == errno.EINTR:
                        continue
                    raise

                if len(events) == 0:
                    continue

                try:
                    topic, stat = recv()
                except zmq.core.error.ZMQError as e:
                    if e.errno != errno.EINTR:
                        raise
                    else:
                        try:
                            sys.exc_clear()
                        except Exception:
                            pass
                        continue

                topic = u(topic).split(".")
                if len(topic) == 3:
                    __, watcher, subtopic = topic
                    yield watcher, subtopic, json.loads(stat)
                elif len(topic) == 2:
                    __, watcher = topic
                    yield watcher, None, json.loads(stat)

    def _handle_pipe(self):
        '''
        Handle a message from front-end API.
        '''
        terminate = False

        # Get the whole message off the pipe in one go
        msg = self.pipe.recv_multipart()

        if msg is None:
            terminate = True
            return terminate

        command = msg[0]
        logging.debug("auth received API command {0}".format(command))

        if command == b'ALLOW':
            address = u(msg[1], self.encoding)
            self.authenticator.allow(address)

        elif command == b'DENY':
            address = u(msg[1], self.encoding)
            self.authenticator.deny(address)

        elif command == b'PLAIN':
            domain = u(msg[1], self.encoding)
            json_passwords = msg[2]
            self.authenticator.configure_plain(domain, jsonapi.loads(json_passwords))

        elif command == b'CURVE':
            # For now we don't do anything with domains
            domain = u(msg[1], self.encoding)

            # If location is CURVE_ALLOW_ANY, allow all clients. Otherwise
            # treat location as a directory that holds the certificates.
            location = u(msg[2], self.encoding)
            self.authenticator.configure_curve(domain, location)

        elif command == b'TERMINATE':
            terminate = True

        else:
            logging.error("Invalid auth command from API: {0}".format(command))

        return terminate

def read_from_stream(stream, timeout=5):
    start = time.time()
    while time.time() - start < timeout:
        try:
            data = stream.get_nowait()
            raise tornado.gen.Return(u(data['data']))
        except Empty:
            yield tornado_sleep(0.1)
    raise TimeoutException('Timeout reading queue')

 def test():
     a, b = self.create_bound_pair(zmq.PUSH, zmq.PULL)
     f = b.recv_string()
     assert not f.done()
     msg = u('πøøπ')
     yield a.send_string(msg)
     recvd = yield f
     assert f.done()
     self.assertEqual(f.result(), msg)
     self.assertEqual(recvd, msg)

def run_ctl(args, queue=None, stdin=''):
    cmd = '%s -m circus.circusctl' % sys.executable
    proc = subprocess.Popen(cmd.split() + shlex.split(args),
                            stdin=subprocess.PIPE if stdin else None,
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE)
    stdout, stderr = proc.communicate(b(stdin) if stdin else None)
    stdout = u(stdout)
    stderr = u(stderr)
    if queue:
        queue.put(stderr)
        queue.put(stdout)
    try:
        import gevent
        if hasattr(gevent, 'shutdown'):
            gevent.shutdown()
    except ImportError:
        pass
    return stdout, stderr

    def handle_recv(self, data):
        topic, msg = data
        topic_parts = u(topic).split(".")
        if topic_parts[2] == "reap":
            timeline = self.timelines.get(topic_parts[1], [])
            timeline.append(time.time())
            self.timelines[topic_parts[1]] = timeline

            self.check(topic_parts[1])
        elif topic_parts[2] == "updated":
            self.update_conf(topic_parts[1])

def _write_key_file(key_filename, banner, public_key, secret_key=None, metadata=None, encoding='utf-8'):
    """ Create a certificate file """
    if isinstance(public_key, bytes):
        public_key = public_key.decode(encoding)
    if isinstance(secret_key, bytes):
        secret_key = secret_key.decode(encoding)
    with io.open(key_filename, 'w', encoding='utf8') as f:
        f.write(banner.format(datetime.datetime.now()))

        f.write(u('metadata\n'))
        if metadata:
            for k, v in metadata.items():
                if isinstance(v, bytes):
                    v = v.decode(encoding)
                f.write(u("    {0} = {1}\n").format(k, v))

        f.write(u('curve\n'))
        f.write(u("    public-key = \"{0}\"\n").format(public_key))

        if secret_key:
            f.write(u("    secret-key = \"{0}\"\n").format(secret_key))

    def run(self):
        self.setup()

        while True:
            msg = self.sock.recv_multipart()
            # TODO: use a pipeline to transform the messages
            try:
                (topic, uuid, dt, username, data) = msg[:]
            except IndexError:
                self.LOG.error("Droping invalid message")
                self.LOG.debug("=> %s", msg)
                continue
            self.LOG.debug("topic: %s, data: %s", u(topic), data)
            self.push.send_multipart(msg)

    def __call__(self, data):
        if self._should_rollover(data["data"]):
            self._do_rollover()

        # If we want to prefix the stream with the current datetime
        for line in u(data["data"]).split("\n"):
            if not line:
                continue
            if self.time_format is not None:
                self._file.write(
                    "{time} [{pid}] | ".format(time=self.now().strftime(self.time_format), pid=data["pid"])
                )
            self._file.write(line)
            self._file.write("\n")
        self._file.flush()