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()
评论列表
文章目录