def test_multisend(self):
"""ensure that a message remains intact after multiple sends"""
a,b = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
s = b"message"
m = zmq.Frame(s)
self.assertEqual(s, m.bytes)
a.send(m, copy=False)
time.sleep(0.1)
self.assertEqual(s, m.bytes)
a.send(m, copy=False)
time.sleep(0.1)
self.assertEqual(s, m.bytes)
a.send(m, copy=True)
time.sleep(0.1)
self.assertEqual(s, m.bytes)
a.send(m, copy=True)
time.sleep(0.1)
self.assertEqual(s, m.bytes)
for i in range(4):
r = b.recv()
self.assertEqual(s,r)
self.assertEqual(s, m.bytes)
评论列表
文章目录