class ZmqREQREPTwoFactoryConnectionTestCase(unittest.TestCase):
    """
    Test case for L{zmq.req_rep} with REQ/REP in two factories.
    """

    REQUEST_COUNT = 10000

    def setUp(self):
        self.factory1 = ZmqFactory()
        self.factory2 = ZmqFactory()
        c = ZmqEndpoint(ZmqEndpointType.connect, "tcp://127.0.0.1:7859")
        self.c1 = ZmqRequestConnection(self.factory1, c, identity=b'master')
        b = ZmqEndpoint(ZmqEndpointType.bind, "tcp://127.0.0.1:7859")
        self.c2 = ZmqReplyConnection(self.factory2, b, identity=b'slave')
        self.c1.d = defer.Deferred()

    def tearDown(self):
        self.factory2.shutdown()
        self.factory1.shutdown()

    def test_start(self):
        for _ in range(self.REQUEST_COUNT):
            reactor.callLater(0, self.c1.send, b'req')
        reactor.callLater(0, self.c1.send, b'stop')

        def checkResults(_):
            self.failUnlessEqual(self.c1.message_count, 3 * self.REQUEST_COUNT)
            self.failUnlessEqual(self.c2.message_count, self.REQUEST_COUNT)

        return self.c1.d.addCallback(checkResults)

class ZmqRouterDealerTwoFactoryConnectionTestCase(unittest.TestCase):
    """
    Test case for L{txzmq.req_rep} with ROUTER/DEALER in two factories.
    """

    REQUEST_COUNT = 10000

    def setUp(self):
        self.factory1 = ZmqFactory()
        dealer_endpoint = ZmqEndpoint(ZmqEndpointType.connect, "ipc://#7")
        self.dealer = ZmqTestDealerConnection(self.factory1, dealer_endpoint,
                                              identity='dealer')
        self.dealer.d = defer.Deferred()

        self.factory2 = ZmqFactory()
        router_endpoint = ZmqEndpoint(ZmqEndpointType.bind, "ipc://#7")
        self.router = ZmqTestRouterConnection(self.factory2, router_endpoint,
                                              identity='router')

    def tearDown(self):
        self.factory2.shutdown()
        self.factory1.shutdown()

    def test_start(self):
        for _ in range(self.REQUEST_COUNT):
            reactor.callLater(0, self.dealer.sendMsg, 'req')
        reactor.callLater(0, self.dealer.sendMsg, 'stop')

        def checkResults(_):
            self.failUnlessEqual(self.dealer.message_count,
                                 3 * self.REQUEST_COUNT)
            self.failUnlessEqual(self.router.message_count, self.REQUEST_COUNT)

        return self.dealer.d.addCallback(checkResults)

 def setUp(self):
     self.factory1 = ZmqFactory()
     self.factory2 = ZmqFactory()
     c = ZmqEndpoint(ZmqEndpointType.connect, "tcp://127.0.0.1:7859")
     self.c1 = ZmqRequestConnection(self.factory1, c, identity=b'master')
     b = ZmqEndpoint(ZmqEndpointType.bind, "tcp://127.0.0.1:7859")
     self.c2 = ZmqReplyConnection(self.factory2, b, identity=b'slave')
     self.c1.d = defer.Deferred()

class ZmqFactoryTestCase(unittest.TestCase):
    """
    Test case for L{zmq.twisted.factory.Factory}.
    """

    def setUp(self):
        self.factory = ZmqFactory()

    def test_shutdown(self):
        self.factory.shutdown()

    def setUp(self):
        self.factory1 = ZmqFactory()
        dealer_endpoint = ZmqEndpoint(ZmqEndpointType.connect, "ipc://#7")
        self.dealer = ZmqTestDealerConnection(self.factory1, dealer_endpoint,
                                              identity='dealer')
        self.dealer.d = defer.Deferred()

        self.factory2 = ZmqFactory()
        router_endpoint = ZmqEndpoint(ZmqEndpointType.bind, "ipc://#7")
        self.router = ZmqTestRouterConnection(self.factory2, router_endpoint,
                                              identity='router')

class BaseTestCase(unittest.TestCase):
    """
    This should be subclassed by the other test cases in this module.
    """
    def setUp(self):
        self.factory = ZmqFactory()
        self.factory.testMessage = ""
        ZmqPubConnection.allowLoopbackMulticast = True

    def tearDown(self):
        del ZmqPubConnection.allowLoopbackMulticast
        self.factory.shutdown()

 def setUp(self):
     self.factory = ZmqFactory()
     self.r = ZmqTestREPConnection(self.factory,
             ZmqEndpoint(ZmqEndpointType.bind, "ipc://#3"))
     self.s = ZmqREQConnection(self.factory,
             ZmqEndpoint(ZmqEndpointType.connect, "ipc://#3"),
             identity='client')

    def setUp(self):
        self.factory = ZmqFactory()
        ZmqXREQConnection.identity = "client"
        self.r = ZmqTestXREPConnection(ZmqEndpoint(ZmqEndpointType.bind, "ipc://#3"))
        self.r.listen(self.factory)
        self.s = ZmqXREQConnection(ZmqEndpoint(ZmqEndpointType.connect, "ipc://#3"))
        self.s.connect(self.factory)
        self.count = 0

        def get_next_id():
            self.count += 1
            return "msg_id_%d" % (self.count,)

        self.s._getNextId = get_next_id

class ZmqConnectionTestCase(unittest.TestCase):
    """
    Test case for L{zmq.twisted.connection.Connection}.

    In ZeroMQ 2.x, subscription is handled on receiving side:
    incoming messages are simply filtered, that's why connection.subscribe
    works immediately.

    In ZeroMQ 3.x, subscription is handled on publisher side:
    subscriber sends message to the publisher and publisher adjusts
    filtering on its side. So connection.subscribe doesn't start filtering
    immediately, it takes some time for messages to pass through the channel.
    """

    def setUp(self):
        self.factory = ZmqFactory()
        ZmqPubConnection.allowLoopbackMulticast = True

    def tearDown(self):
        del ZmqPubConnection.allowLoopbackMulticast
        self.factory.shutdown()

    def test_send_recv(self):
        r = ZmqTestSubConnection(
            self.factory, ZmqEndpoint(ZmqEndpointType.bind, "ipc://test-sock"))
        s = ZmqPubConnection(
            self.factory, ZmqEndpoint(ZmqEndpointType.connect,
                                      "ipc://test-sock"))

        r.subscribe('tag')

        def publish(ignore):
            s.publish('xyz', 'different-tag')
            s.publish('abcd', 'tag1')
            s.publish('efgh', 'tag2')

        def check(ignore):
            result = getattr(r, 'messages', [])
            expected = [['tag1', 'abcd'], ['tag2', 'efgh']]
            self.failUnlessEqual(
                result, expected, "Message should have been received")

        return _wait(0.01).addCallback(publish) \
            .addCallback(lambda _: _wait(0.01)).addCallback(check)

    def test_send_recv_pgm(self):
        r = ZmqTestSubConnection(self.factory, ZmqEndpoint(
            ZmqEndpointType.bind, "epgm://127.0.0.1;239.192.1.1:5556"))

        s = ZmqPubConnection(self.factory, ZmqEndpoint(
            ZmqEndpointType.connect, "epgm://127.0.0.1;239.192.1.1:5556"))

        r.subscribe('tag')

        def publish(ignore):
            s.publish('xyz', 'different-tag')
            s.publish('abcd', 'tag1')

        def check(ignore):
            result = getattr(r, 'messages', [])
            expected = [['tag1', 'abcd']]
            self.failUnlessEqual(
                result, expected, "Message should have been received")

        return _wait(0.2).addCallback(publish) \
            .addCallback(lambda _: _wait(0.2)).addCallback(check)

    def test_send_recv_multiple_endpoints(self):
        r = ZmqTestSubConnection(
            self.factory,
            ZmqEndpoint(ZmqEndpointType.bind, "tcp://127.0.0.1:5556"))
        r.addEndpoints([ZmqEndpoint(ZmqEndpointType.bind,
                                    "inproc://endpoint")])
        s1 = ZmqPubConnection(
            self.factory,
            ZmqEndpoint(ZmqEndpointType.connect, "tcp://127.0.0.1:5556"))
        s2 = ZmqPubConnection(
            self.factory,
            ZmqEndpoint(ZmqEndpointType.connect, "inproc://endpoint"))

        r.subscribe('')

        def publish(ignore):
            s1.publish('111', 'tag1')
            s2.publish('222', 'tag2')

        def check(ignore):
            result = getattr(r, 'messages', [])
            expected = [['tag1', '111'], ['tag2', '222']]
            self.failUnlessEqual(
                sorted(result), expected, "Message should have been received")

        return _wait(0.1).addCallback(publish) \
            .addCallback(lambda _: _wait(0.1)).addCallback(check)

    if not _detect_epgm():
        test_send_recv_pgm.skip = "epgm:// not available"

 def setUp(self):
     self.factory = ZmqFactory()
     ZmqPubConnection.allowLoopbackMulticast = True

 def setUp(self):
     self.factory = ZmqFactory()
     self.factory.testMessage = ""
     ZmqPubConnection.allowLoopbackMulticast = True

class ZmqREQREPConnectionTestCase(unittest.TestCase):
    """
    Test case for L{zmq.req_rep.ZmqREPConnection}.
    """

    def setUp(self):
        self.factory = ZmqFactory()
        b = ZmqEndpoint(ZmqEndpointType.bind, "ipc://#3")
        self.r = ZmqTestREPConnection(self.factory, b)
        c = ZmqEndpoint(ZmqEndpointType.connect, "ipc://#3")
        self.s = ZmqREQConnection(self.factory, c, identity=b'client')

    def tearDown(self):
        self.factory.shutdown()

    def test_getNextId(self):
        self.failUnlessEqual([], self.s._uuids)
        id1 = self.s._getNextId()
        self.failUnlessEqual(self.s.UUID_POOL_GEN_SIZE - 1, len(self.s._uuids))
        self.failUnlessIsInstance(id1, binary_string_type)

        id2 = self.s._getNextId()
        self.failUnlessIsInstance(id2, binary_string_type)

        self.failIfEqual(id1, id2)

        ids = [self.s._getNextId() for _ in range(1000)]
        self.failUnlessEqual(len(ids), len(set(ids)))

    def test_releaseId(self):
        self.s._releaseId(self.s._getNextId())
        self.failUnlessEqual(self.s.UUID_POOL_GEN_SIZE, len(self.s._uuids))

    def test_send_recv(self):
        self.count = 0

        def get_next_id():
            self.count += 1
            return b'msg_id_' + str(self.count).encode()

        self.s._getNextId = get_next_id

        self.s.sendMsg(b'aaa', b'aab')
        self.s.sendMsg(b'bbb')

        def check(ignore):
            result = getattr(self.r, 'messages', [])
            expected = [[b'msg_id_1', (b'aaa', b'aab')],
                        [b'msg_id_2', (b'bbb',)]]
            self.failUnlessEqual(
                result, expected, "Message should have been received")

        return _wait(0.01).addCallback(check)

    def test_send_recv_reply(self):
        d = self.s.sendMsg(b'aaa')

        def check_response(response):
            self.assertEqual(response, [b'aaa'])

        d.addCallback(check_response)
        return d

    def test_lot_send_recv_reply(self):
        deferreds = []
        for i in range(10):
            msg_id = "msg_id_%d" % (i,)
            d = self.s.sendMsg(b'aaa')

            def check_response(response, msg_id):
                self.assertEqual(response, [b'aaa'])

            d.addCallback(check_response, msg_id)
            deferreds.append(d)
        return defer.DeferredList(deferreds, fireOnOneErrback=True)

    def test_cleanup_requests(self):
        """The request dict is cleanedup properly."""
        def check(ignore):
            self.assertEqual(self.s._requests, {})
            self.failUnlessEqual(self.s.UUID_POOL_GEN_SIZE, len(self.s._uuids))

        return self.s.sendMsg(b'aaa').addCallback(check)

    def test_cancel(self):
        d = self.s.sendMsg(b'aaa')
        d.cancel()

        def check_requests(_):
            self.assertEqual(self.s._requests, {})
            self.failUnlessEqual(self.s.UUID_POOL_GEN_SIZE,
                                 len(self.s._uuids) + 1)

        return d.addCallbacks(lambda _: self.fail("Should have errored"),
                              lambda fail: fail.trap(
                              "twisted.internet.defer.CancelledError")) \
            .addCallback(check_requests) \
            .addCallback(lambda _: _wait(0.01))

    def test_send_timeout_ok(self):
#.........这里部分代码省略.........

class ZmqConnectionTestCase(unittest.TestCase):
    """
    Test case for L{zmq.twisted.connection.Connection}.
    """
    
    def make_pub(self, *args, **kwargs):
        from txzmq.pubsub import ZmqPubConnection
        return ZmqPubConnection(self.factory, *args, **kwargs)

    def make_sub(self, *args, **kwargs):
        from txzmq.pubsub import ZmqSubConnection
        return ZmqSubConnection(self.factory, *args, **kwargs)

    def setUp(self):
        from txzmq.factory import ZmqFactory
        self.factory = ZmqFactory()

    def tearDown(self):
        self.factory.shutdown()

    def test_pub_sub(self):
        import zmq
        from txzmq.test import _wait
        
        result = []
        def msg_func(msg):
            result.append(msg)
        
        sub = self.make_sub(callback=msg_func)
        sub.bind('inproc://#1')
        sub.setsockopt(zmq.SUBSCRIBE, '')
        
        pub = self.make_pub()
        pub.connect('inproc://#1')
        
        pub.send(['a', 'b', 'c'])
        pub.send(['1', '2', '3'])
        
        def check(ignore):
            expected = [['a', 'b', 'c'], ['1', '2', '3']]
            self.assertEqual(result, expected)

        return _wait(0.1).addCallback(check)
    
    def test_pub_sub_with_filter(self):
        import zmq
        from txzmq.test import _wait
        
        result = []
        def msg_func(msg):
            result.append(msg)
        
        sub = self.make_sub(callback=msg_func)
        sub.bind('inproc://#1')
        sub.setsockopt(zmq.SUBSCRIBE, 'a')
        
        pub = self.make_pub()
        pub.connect('inproc://#1')
        
        pub.send(['a', 'data1'])
        pub.send(['b', 'data2'])
        pub.send(['a', 'data3'])
        pub.send(['a', 'data4'])
        
        def check(ignore):
            expected = [['a', 'data1'], ['a', 'data3'], ['a', 'data4']]
            self.assertEqual(result, expected)

        return _wait(0.1).addCallback(check)

class ZmqConnectionTestCase(unittest.TestCase):
    """
    Test case for L{zmq.twisted.connection.Connection}.
    """
    
    def make_push(self, *args, **kwargs):
        from txzmq.pushpull import ZmqPushConnection
        return ZmqPushConnection(self.factory, *args, **kwargs)

    def make_pull(self, *args, **kwargs):
        from txzmq.pushpull import ZmqPullConnection
        return ZmqPullConnection(self.factory, *args, **kwargs)
    
    def setUp(self):
        from txzmq.factory import ZmqFactory
        self.factory = ZmqFactory()

    def tearDown(self):
        self.factory.shutdown()

    def test_send_recv(self):
        from txzmq.test import _wait
        
        result = []
        def msg_func(msg):
            result.append(msg)
        
        pull = self.make_pull(callback=msg_func)
        pull.bind("inproc://#1")
        
        push = self.make_push()
        push.connect("inproc://#1")
        push.send(['abcd'])

        def check(ignore):
            expected = [['abcd']]
            self.assertEqual(result, expected)

        return _wait(0.1).addCallback(check)
    
    def test_send_to_multiple_nodes(self):
        from txzmq.test import _wait
        
        result = []
        def msg_func(msg):
            result.append(msg)
        
        pull1 = self.make_pull(callback=msg_func)
        pull1.bind("inproc://#1")
        
        pull2 = self.make_pull(callback=msg_func)
        pull2.bind("inproc://#2")
        
        pull3 = self.make_pull(callback=msg_func)
        pull3.bind("inproc://#3")
        
        push = self.make_push()
        push.connect("inproc://#1")
        push.connect("inproc://#2")
        push.connect("inproc://#3")
        
        for i in range(1, 6+1):
            push.send(['msg%d' % i])
            
        def check(ignore):
            for i in range(1, 6+1):
                self.assertIn(['msg1'], result)

        return _wait(0.1).addCallback(check)

 def _test():
     factory = ZmqFactory()
     factory.reactor = reactor
     factory.registerForShutdown()
     reactor.stop()

class ZmqREQREPConnectionTestCase(unittest.TestCase):
    """
    Test case for L{zmq.req_rep.ZmqREPConnection}.
    """

    def setUp(self):
        self.factory = ZmqFactory()
        self.r = ZmqTestREPConnection(self.factory,
                ZmqEndpoint(ZmqEndpointType.bind, "ipc://#3"))
        self.s = ZmqREQConnection(self.factory,
                ZmqEndpoint(ZmqEndpointType.connect, "ipc://#3"),
                identity='client')

    def tearDown(self):
        self.factory.shutdown()

    def test_getNextId(self):
        self.failUnlessEqual([], self.s._uuids)
        id1 = self.s._getNextId()
        self.failUnlessEqual(self.s.UUID_POOL_GEN_SIZE - 1, len(self.s._uuids))
        self.failUnlessIsInstance(id1, str)

        id2 = self.s._getNextId()
        self.failUnlessIsInstance(id2, str)

        self.failIfEqual(id1, id2)

        ids = [self.s._getNextId() for _ in range(1000)]
        self.failUnlessEqual(len(ids), len(set(ids)))

    def test_releaseId(self):
        self.s._releaseId(self.s._getNextId())
        self.failUnlessEqual(self.s.UUID_POOL_GEN_SIZE, len(self.s._uuids))

    def test_send_recv(self):
        self.count = 0

        def get_next_id():
            self.count += 1
            return 'msg_id_%d' % (self.count,)

        self.s._getNextId = get_next_id

        self.s.sendMsg('aaa', 'aab')
        self.s.sendMsg('bbb')

        def check(ignore):
            result = getattr(self.r, 'messages', [])
            expected = [['msg_id_1', ('aaa', 'aab')], ['msg_id_2', ('bbb',)]]
            self.failUnlessEqual(
                result, expected, "Message should have been received")

        return _wait(0.01).addCallback(check)

    def test_send_recv_reply(self):
        d = self.s.sendMsg('aaa')

        def check_response(response):
            self.assertEqual(response, ['aaa'])

        d.addCallback(check_response)
        return d

    def test_lot_send_recv_reply(self):
        deferreds = []
        for i in range(10):
            msg_id = "msg_id_%d" % (i,)
            d = self.s.sendMsg('aaa')

            def check_response(response, msg_id):
                self.assertEqual(response, ['aaa'])

            d.addCallback(check_response, msg_id)
            deferreds.append(d)
        return defer.DeferredList(deferreds, fireOnOneErrback=True)

    def test_cleanup_requests(self):
        """The request dict is cleanedup properly."""
        def check(ignore):
            self.assertEqual(self.s._requests, {})
            self.failUnlessEqual(self.s.UUID_POOL_GEN_SIZE, len(self.s._uuids))

        return self.s.sendMsg('aaa').addCallback(check)

 def setUp(self):
     self.factory = ZmqFactory()

class ZmqConnectionTestCase(unittest.TestCase):
    """
    Test case for L{zmq.twisted.connection.Connection}.
    """

    def setUp(self):
        self.factory = ZmqFactory()
        ZmqPubConnection.allowLoopbackMulticast = True

    def tearDown(self):
        del ZmqPubConnection.allowLoopbackMulticast
        self.factory.shutdown()

    def test_send_recv(self):
        r = ZmqTestSubConnection(
            self.factory, ZmqEndpoint(ZmqEndpointType.bind, "ipc://test-sock"))
        s = ZmqPubConnection(
            self.factory, ZmqEndpoint(ZmqEndpointType.connect,
            "ipc://test-sock"))

        r.subscribe('tag')
        s.publish('xyz', 'different-tag')
        s.publish('abcd', 'tag1')
        s.publish('efgh', 'tag2')

        def check(ignore):
            result = getattr(r, 'messages', [])
            expected = [['tag1', 'abcd'], ['tag2', 'efgh']]
            self.failUnlessEqual(
                result, expected, "Message should have been received")

        return _wait(0.01).addCallback(check)

    def test_send_recv_pgm(self):
        r = ZmqTestSubConnection(self.factory, ZmqEndpoint(
            ZmqEndpointType.bind, "epgm://127.0.0.1;239.192.1.1:5556"))
        s = ZmqPubConnection(self.factory, ZmqEndpoint(
            ZmqEndpointType.connect, "epgm://127.0.0.1;239.192.1.1:5556"))

        r.subscribe('tag')
        s.publish('xyz', 'different-tag')
        s.publish('abcd', 'tag1')

        def check(ignore):
            result = getattr(r, 'messages', [])
            expected = [['tag1', 'abcd']]
            self.failUnlessEqual(
                result, expected, "Message should have been received")

        return _wait(0.2).addCallback(check)

    def test_send_recv_multiple_endpoints(self):
        r = ZmqTestSubConnection(
            self.factory,
            ZmqEndpoint(ZmqEndpointType.bind, "tcp://127.0.0.1:5556"))
        r.addEndpoints(
                [ZmqEndpoint(ZmqEndpointType.bind, "inproc://endpoint")])
        s1 = ZmqPubConnection(
            self.factory,
            ZmqEndpoint(ZmqEndpointType.connect, "tcp://127.0.0.1:5556"))
        s2 = ZmqPubConnection(
            self.factory,
            ZmqEndpoint(ZmqEndpointType.connect, "inproc://endpoint"))

        r.subscribe('')
        s1.publish('111', 'tag1')
        s2.publish('222', 'tag2')

        def check(ignore):
            result = getattr(r, 'messages', [])
            expected = [['tag1', '111'], ['tag2', '222']]
            self.failUnlessEqual(
                sorted(result), expected, "Message should have been received")

        return _wait(0.2).addCallback(check)

class ZmqConnectionTestCase(unittest.TestCase):
    """
    Test case for L{zmq.twisted.connection.Connection}.
    """
    
    def make_one(self, *args, **kwargs):
        from txzmq.pair import ZmqPairConnection
        return ZmqPairConnection(self.factory, *args, **kwargs)
    
    def setUp(self):
        from txzmq.factory import ZmqFactory
        self.factory = ZmqFactory()

    def tearDown(self):
        self.factory.shutdown()

    def test_send_recv(self):
        from txzmq.test import _wait
        
        a_result = []
        def a_msg_func(msg):
            a_result.append(msg)
        
        b_result = []
        def b_msg_func(msg):
            b_result.append(msg)
            
        a = self.make_one(callback=a_msg_func)
        a.bind("inproc://#1")
        
        b = self.make_one(callback=b_msg_func)
        b.connect("inproc://#1")
        b.send(['b2a'])
        
        def check(ignore):
            expected = [['a2b']]
            self.assertEqual(b_result, expected)

        def check_and_send(ignore):
            expected = [['b2a']]
            self.assertEqual(a_result, expected)
            a.send(['a2b'])
            return _wait(0.1).addCallback(check)

        return _wait(0.1).addCallback(check_and_send)
    
    def test_send_lot(self):
        from txzmq.test import _wait
        
        a_result = []
        def a_msg_func(msg):
            a_result.append(msg)
        
        b_result = []
        def b_msg_func(msg):
            b_result.append(msg)
            
        a = self.make_one(callback=a_msg_func)
        a.bind("inproc://#1")
        
        b = self.make_one(callback=b_msg_func)
        b.connect("inproc://#1")
        
        a2b_msgs = [['a2b %d' % i] for i in range(100)]
        for m in a2b_msgs:
            a.send(m)
        b2a_msgs = [['b2a %d' % i] for i in range(100)]
        for m in b2a_msgs:
            b.send(m)
        
        def check(ignore):
            self.assertEqual(a_result, b2a_msgs)
            self.assertEqual(b_result, a2b_msgs)

        return _wait(0.1).addCallback(check)

 def setUp(self):
     from txzmq.factory import ZmqFactory
     self.factory = ZmqFactory()