test_messenger.py 文件源码

def support_test_send_to_multiple_addresses(self, address1, address2):
        poller = zmq.Poller()

        socket1 = self.context.socket(roles['listener'])
        socket2 = self.context.socket(roles['listener'])
        try:
            socket1.bind("tcp://%s" % address1)
            socket2.bind("tcp://%s" % address2)
            poller.register(socket1, zmq.POLLIN)
            poller.register(socket2, zmq.POLLIN)
            polled = dict(poller.poll(2000))
            if socket1 in polled:
                socket1.recv()
                socket1.send(nw0.sockets._serialise(address1))
            elif socket2 in polled:
                socket2.recv()
                socket2.send(nw0.sockets._serialise(address2))
            else:
                raise RuntimeError("Nothing found")
        finally:
            socket1.close()
            socket2.close()