def send_event(topic, user, data):
    # Get back the thread local storage
    try:
        context = thread_local.context
        socket = thread_local.socket
    except AttributeError:
        # Create the context and socket
        thread_local.context = context = zmq.Context.instance()
        thread_local.socket = socket = context.socket(zmq.PUSH)
        socket.connect(settings.INTERNAL_EVENT_SOCKET)

    try:
        # The format is [topic, uuid, datetime, username, data as json]
        msg = [
            b(settings.EVENT_TOPIC + topic),
            b(str(uuid.uuid1())),
            b(datetime.datetime.utcnow().isoformat()),
            b(user),
            b(simplejson.dumps(data))
        ]
        # Send the message in the non-blockng mode.
        # If the consumer (lava-publisher) is not active, the message will be lost.
        socket.send_multipart(msg, zmq.DONTWAIT)
    except (TypeError, ValueError, zmq.ZMQError):
        # The event can't be send, just skip it
        print("Unable to send the zmq event %s" % (settings.EVENT_TOPIC + topic))

 def test_init_iface(self):
     logger = self.logger
     ctx = self.context
     handler = handlers.PUBHandler(self.iface)
     self.assertFalse(handler.ctx is ctx)
     self.sockets.append(handler.socket)
     # handler.ctx.term()
     handler = handlers.PUBHandler(self.iface, self.context)
     self.sockets.append(handler.socket)
     self.assertTrue(handler.ctx is ctx)
     handler.setLevel(logging.DEBUG)
     handler.root_topic = self.topic
     logger.addHandler(handler)
     sub = ctx.socket(zmq.SUB)
     self.sockets.append(sub)
     sub.setsockopt(zmq.SUBSCRIBE, b(self.topic))
     sub.connect(self.iface)
     import time; time.sleep(0.25)
     msg1 = 'message'
     logger.info(msg1)
     
     (topic, msg2) = sub.recv_multipart()
     self.assertEquals(topic, b'zmq.INFO')
     self.assertEquals(msg2, b(msg1)+b'\n')
     logger.removeHandler(handler)

def main():
    # Get the arguments
    if len(sys.argv) != 4 and len(sys.argv) != 6:
        print("%d arguments" % len(sys.argv))
        print("Usage: push.py url topic num_messages [master_cert slave_cert]")
        sys.exit(1)

    url = sys.argv[1]
    topic = sys.argv[2]
    num_messages = int(sys.argv[3])
    username = pwd.getpwuid(os.geteuid()).pw_name

    # Create the socket
    context = zmq.Context()
    sock = context.socket(zmq.PUSH)
    if len(sys.argv) > 4:
        # Configure encryption
        (server_public, _) = zmq.auth.load_certificate(sys.argv[4])
        sock.curve_serverkey = server_public

        (client_public, client_private) = zmq.auth.load_certificate(sys.argv[5])
        sock.curve_publickey = client_public
        sock.curve_secretkey = client_private

    sock.connect(url)

    for i in range(0, num_messages):
        sock.send_multipart([b(topic),
                             b(str(uuid.uuid1())),
                             b(datetime.datetime.utcnow().isoformat()),
                             b(username),
                             b(json.dumps({'id': i}))])

    def configure_curve(self, domain='*', location=''):
        try:
            domain = b(domain, self.encoding)
            location = b(location, self.encoding)
            self.pipe.send([b'CURVE', domain, location])

        except Exception as e:
            log.exception('configure-curve', e=e)

 def test_frame_more(self):
     """test Frame.more attribute"""
     frame = zmq.Frame(b("hello"))
     self.assertFalse(frame.more)
     sa,sb = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
     sa.send_multipart([b('hi'), b('there')])
     frame = self.recv(sb, copy=False)
     self.assertTrue(frame.more)
     frame = self.recv(sb, copy=False)
     self.assertFalse(frame.more)

 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 configure_curve(self, domain='*', location=''):
     '''
     Configure CURVE authentication for a given domain. CURVE authentication
     uses a directory that holds all public client certificates, i.e. their
     public keys. The certificates must be in zcert_save () format.
     To cover all domains, use "*".
     You can add and remove certificates in that directory at any time.
     To allow all client keys without checking, specify CURVE_ALLOW_ANY for
     the location.
     '''
     domain = b(domain, self.encoding)
     location = b(location, self.encoding)
     self.pipe.send_multipart([b'CURVE', domain, location])

 def test_lifecycle2(self):
     """Run through a different ref counting cycle with a copy."""
     for i in range(5, 16):  # 32, 64,..., 65536
         s = (2**i)*x
         rc = 2
         self.assertEqual(grc(s), rc)
         m = zmq.Frame(s)
         rc += 2
         self.assertEqual(grc(s), rc)
         m2 = copy.copy(m)
         rc += 1
         self.assertEqual(grc(s), rc)
         buf = m.buffer
         rc += view_rc
         self.assertEqual(grc(s), rc)
         self.assertEqual(s, b(str(m)))
         self.assertEqual(s, bytes(m2))
         self.assertEqual(s, m2.bytes)
         self.assertEqual(s, m.bytes)
         # self.assert_(s is str(m))
         # self.assert_(s is str(m2))
         del buf
         self.assertEqual(grc(s), rc)
         del m
         # m.buffer is kept until m is del'd
         rc -= view_rc
         rc -= 1
         self.assertEqual(grc(s), rc)
         del m2
         rc -= 2
         await_gc(s, rc)
         self.assertEqual(grc(s), rc)
         self.assertEqual(rc, 2)
         del s

    def test_lifecycle1(self):
        """Run through a ref counting cycle with a copy."""
        for i in range(5, 16):  # 32, 64,..., 65536
            s = (2**i)*x
            rc = 2
            self.assertEquals(grc(s), rc)
            m = zmq.Frame(s)
            rc += 2
            self.assertEquals(grc(s), rc)
            m2 = copy.copy(m)
            rc += 1
            self.assertEquals(grc(s), rc)
            buf = m2.buffer

            rc += view_rc
            self.assertEquals(grc(s), rc)

            self.assertEquals(s, b(str(m)))
            self.assertEquals(s, bytes(m2))
            self.assertEquals(s, m.bytes)
            # self.assert_(s is str(m))
            # self.assert_(s is str(m2))
            del m2
            rc -= 1
            self.assertEquals(grc(s), rc)
            rc -= view_rc
            del buf
            self.assertEquals(grc(s), rc)
            del m
            rc -= 2
            self.assertEquals(grc(s), rc)
            self.assertEquals(rc, 2)
            del s

 def configure_plain(self, domain='*', passwords=None):
     '''
     Configure PLAIN authentication for a given domain. PLAIN authentication
     uses a plain-text password file. To cover all domains, use "*".
     You can modify the password file at any time; it is reloaded automatically.
     '''
     self.pipe.send_multipart([b'PLAIN', b(domain, self.encoding), jsonapi.dumps(passwords or {})])

def send_multipart_u(sock, data):
    """ Wrapper around send_multipart that encode data as bytes.

    :param sock: The socket to use
    :param data: Data to convert to byte strings
    """
    return sock.send_multipart([b(d) for d in data])

 def test_str(self):
     """Test the str representations of the Frames."""
     for i in range(16):
         s = (2**i)*x
         m = zmq.Frame(s)
         m_str = str(m)
         m_str_b = b(m_str) # py3compat
         self.assertEqual(s, m_str_b)

 def test_buffer_out(self):
     """receiving buffered output"""
     ins = b("§§¶•ªº˜µ¬˚…∆˙åß∂©œ∑´†≈ç√")
     m = zmq.Frame(ins)
     outb = m.buffer
     self.assertTrue(isinstance(outb, view))
     self.assert_(outb is m.buffer)
     self.assert_(m.buffer is m.buffer)

 def deny(self, address):
     '''
     Deny (blacklist) a single IP address. For all security mechanisms, this
     rejects the connection without any further authentication. Use either a
     whitelist, or a blacklist, not not both. If you define both a whitelist
     and a blacklist, only the whitelist takes effect.
     '''
     self.pipe.send_multipart([b'DENY', b(address, self.encoding)])

    def notify_event(self, topic, msg):
        """Publish a message on the event publisher channel"""

        name = bytestring(self.res_name)

        multipart_msg = [b("watcher.%s.%s" % (name, topic)), json.dumps(msg)]

        if self.evpub_socket is not None and not self.evpub_socket.closed:
            self.evpub_socket.send_multipart(multipart_msg)

 def allow(self, address):
     '''
     Allow (whitelist) a single IP address. For NULL, all clients from this
     address will be accepted. For PLAIN and CURVE, they will be allowed to
     continue with authentication. You can call this method multiple times
     to whitelist multiple IP addresses. If you whitelist a single address,
     any non-whitelisted addresses are treated as blacklisted.
     '''
     self.pipe.send_multipart([b'ALLOW', b(address, self.encoding)])

 def test_retry_term(self):
     push = self.socket(zmq.PUSH)
     push.linger = self.timeout_ms
     push.connect('tcp://127.0.0.1:5555')
     push.send(b('ping'))
     time.sleep(0.1)
     self.alarm()
     self.context.destroy()
     assert self.timer_fired
     assert self.context.closed

 def test_init_socket(self):
     pub,sub = self.create_bound_pair(zmq.PUB, zmq.SUB)
     logger = self.logger
     handler = handlers.PUBHandler(pub)
     handler.setLevel(logging.DEBUG)
     handler.root_topic = self.topic
     logger.addHandler(handler)
     
     self.assertTrue(handler.socket is pub)
     self.assertTrue(handler.ctx is pub.context)
     self.assertTrue(handler.ctx is self.context)
     sub.setsockopt(zmq.SUBSCRIBE, b(self.topic))
     import time; time.sleep(0.1)
     msg1 = 'message'
     logger.info(msg1)
     
     (topic, msg2) = sub.recv_multipart()
     self.assertEquals(topic, b'zmq.INFO')
     self.assertEquals(msg2, b(msg1)+b'\n')
     logger.removeHandler(handler)

 def connect_handler(self, topic=None):
     topic = self.topic if topic is None else topic
     logger = self.logger
     pub,sub = self.create_bound_pair(zmq.PUB, zmq.SUB)
     handler = handlers.PUBHandler(pub)
     handler.setLevel(logging.DEBUG)
     handler.root_topic = topic
     logger.addHandler(handler)
     sub.setsockopt(zmq.SUBSCRIBE, b(topic))
     time.sleep(0.1)
     return logger, handler, sub

 def __init__(self, context=None, endpoint=DEFAULT_ENDPOINT_DEALER,
              timeout=5.0, ssh_server=None, ssh_keyfile=None):
     self._init_context(context)
     self.endpoint = endpoint
     self._id = b(uuid.uuid4().hex)
     self.socket = self.context.socket(zmq.DEALER)
     self.socket.setsockopt(zmq.IDENTITY, self._id)
     self.socket.setsockopt(zmq.LINGER, 0)
     get_connection(self.socket, endpoint, ssh_server, ssh_keyfile)
     self._timeout = timeout
     self.timeout = timeout * 1000
     self.stream = ZMQStream(self.socket, tornado.ioloop.IOLoop.instance())