1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
|
from eventlet import event, spawn, sleep, patcher
from eventlet.hubs import get_hub, _threadlocal, use_hub
from nose.tools import *
from tests import mock, LimitedTestCase, using_pyevent, skip_unless
from unittest import TestCase
from threading import Thread
try:
from eventlet.green import zmq
except ImportError:
zmq = {} # for systems lacking zmq, skips tests instead of barfing
def zmq_supported(_):
try:
import zmq
except ImportError:
return False
return not using_pyevent(_)
class TestUpstreamDownStream(LimitedTestCase):
sockets = []
def tearDown(self):
self.clear_up_sockets()
super(TestUpstreamDownStream, self).tearDown()
def create_bound_pair(self, type1, type2, interface='tcp://127.0.0.1'):
"""Create a bound socket pair using a random port."""
self.context = context = zmq.Context()
s1 = context.socket(type1)
port = s1.bind_to_random_port(interface)
s2 = context.socket(type2)
s2.connect('%s:%s' % (interface, port))
self.sockets = [s1, s2]
return s1, s2, port
def clear_up_sockets(self):
for sock in self.sockets:
sock.close()
def assertRaisesErrno(self, errno, func, *args):
try:
func(*args)
except zmq.ZMQError, e:
self.assertEqual(e.errno, errno, "wrong error raised, expected '%s' \
got '%s'" % (zmq.ZMQError(errno), zmq.ZMQError(e.errno)))
else:
self.fail("Function did not raise any error")
@skip_unless(zmq_supported)
def test_recv_spawned_before_send_is_non_blocking(self):
req, rep, port = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
# req.connect(ipc)
# rep.bind(ipc)
sleep()
msg = dict(res=None)
done = event.Event()
def rx():
msg['res'] = rep.recv()
done.send('done')
spawn(rx)
req.send('test')
done.wait()
self.assertEqual(msg['res'], 'test')
@skip_unless(zmq_supported)
def test_close_socket_raises_enotsup(self):
req, rep, port = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
rep.close()
req.close()
self.assertRaisesErrno(zmq.ENOTSUP, rep.recv)
self.assertRaisesErrno(zmq.ENOTSUP, req.send, 'test')
@skip_unless(zmq_supported)
def test_send_1k_req_rep(self):
req, rep, port = self.create_bound_pair(zmq.REQ, zmq.REP)
sleep()
done = event.Event()
def tx():
tx_i = 0
req.send(str(tx_i))
while req.recv() != 'done':
tx_i += 1
req.send(str(tx_i))
def rx():
while True:
rx_i = rep.recv()
if rx_i == "1000":
rep.send('done')
sleep()
done.send(0)
break
rep.send('i')
spawn(tx)
spawn(rx)
final_i = done.wait()
self.assertEqual(final_i, 0)
@skip_unless(zmq_supported)
def test_send_1k_push_pull(self):
down, up, port = self.create_bound_pair(zmq.PUSH, zmq.PULL)
sleep()
done = event.Event()
def tx():
tx_i = 0
while tx_i <= 1000:
tx_i += 1
down.send(str(tx_i))
def rx():
while True:
rx_i = up.recv()
if rx_i == "1000":
done.send(0)
break
spawn(tx)
spawn(rx)
final_i = done.wait()
self.assertEqual(final_i, 0)
@skip_unless(zmq_supported)
def test_send_1k_pub_sub(self):
pub, sub_all, port = self.create_bound_pair(zmq.PUB, zmq.SUB)
sub1 = self.context.socket(zmq.SUB)
sub2 = self.context.socket(zmq.SUB)
self.sockets.extend([sub1, sub2])
addr = 'tcp://127.0.0.1:%s' % port
sub1.connect(addr)
sub2.connect(addr)
sub_all.setsockopt(zmq.SUBSCRIBE, '')
sub1.setsockopt(zmq.SUBSCRIBE, 'sub1')
sub2.setsockopt(zmq.SUBSCRIBE, 'sub2')
sub_all_done = event.Event()
sub1_done = event.Event()
sub2_done = event.Event()
sleep(0.2)
def rx(sock, done_evt, msg_count=10000):
count = 0
while count < msg_count:
msg = sock.recv()
sleep()
if 'LAST' in msg:
break
count += 1
done_evt.send(count)
def tx(sock):
for i in range(1, 1001):
msg = "sub%s %s" % ([2,1][i % 2], i)
sock.send(msg)
sleep()
sock.send('sub1 LAST')
sock.send('sub2 LAST')
spawn(rx, sub_all, sub_all_done)
spawn(rx, sub1, sub1_done)
spawn(rx, sub2, sub2_done)
spawn(tx, pub)
sub1_count = sub1_done.wait()
sub2_count = sub2_done.wait()
sub_all_count = sub_all_done.wait()
self.assertEqual(sub1_count, 500)
self.assertEqual(sub2_count, 500)
self.assertEqual(sub_all_count, 1000)
@skip_unless(zmq_supported)
def test_change_subscription(self):
pub, sub, port = self.create_bound_pair(zmq.PUB, zmq.SUB)
sub.setsockopt(zmq.SUBSCRIBE, 'test')
sleep(0.2)
sub_done = event.Event()
def rx(sock, done_evt):
count = 0
sub = 'test'
while True:
msg = sock.recv()
sleep()
if 'DONE' in msg:
break
if 'LAST' in msg and sub == 'test':
sock.setsockopt(zmq.UNSUBSCRIBE, 'test')
sock.setsockopt(zmq.SUBSCRIBE, 'done')
sub = 'done'
count += 1
done_evt.send(count)
def tx(sock):
for i in range(1, 101):
msg = "test %s" % i
if i != 50:
sock.send(msg)
else:
sock.send('test LAST')
sleep()
sock.send('done DONE')
spawn(rx, sub, sub_done)
spawn(tx, pub)
rx_count = sub_done.wait()
self.assertEqual(rx_count, 50)
@skip_unless(zmq_supported)
def test_recv_multipart_bug68(self):
req, rep, port = self.create_bound_pair(zmq.REQ, zmq.REP)
msg = ['']
req.send_multipart(msg)
recieved_msg = rep.recv_multipart()
self.assertEqual(recieved_msg, msg)
# Send a message back the other way
msg2 = [""]
rep.send_multipart(msg2, copy=False)
# When receiving a copy it's a zmq.core.message.Message you get back
recieved_msg = req.recv_multipart(copy=False)
# So it needs to be converted to a string
# I'm calling str(m) consciously here; Message has a .data attribute
# but it's private __str__ appears to be the way to go
self.assertEqual([str(m) for m in recieved_msg], msg2)
@skip_unless(zmq_supported)
def test_recv_noblock_bug76(self):
req, rep, port = self.create_bound_pair(zmq.REQ, zmq.REP)
self.assertRaisesErrno(zmq.EAGAIN, rep.recv, zmq.NOBLOCK)
self.assertRaisesErrno(zmq.EAGAIN, rep.recv, zmq.NOBLOCK, True)
class TestThreadedContextAccess(TestCase):
"""zmq's Context must be unique within a hub
The zeromq API documentation states:
All zmq sockets passed to the zmq_poll() function must share the same zmq
context and must belong to the thread calling zmq_poll()
As zmq_poll is what's eventually being called then we need to ensure that
all sockets that are going to be passed to zmq_poll (via hub.do_poll) are
in the same context
"""
if zmq: # don't call decorators if zmq module unavailable
@skip_unless(zmq_supported)
def test_context_factory_function(self):
ctx = zmq.Context()
self.assertTrue(ctx is not None)
@skip_unless(zmq_supported)
def test_threadlocal_context(self):
context = zmq.Context()
self.assertEqual(context, _threadlocal.context)
next_context = zmq.Context()
self.assertTrue(context is next_context)
@skip_unless(zmq_supported)
def test_different_context_in_different_thread(self):
context = zmq.Context()
test_result = []
def assert_different(ctx):
try:
this_thread_context = zmq.Context()
except:
test_result.append('fail')
raise
test_result.append(ctx is this_thread_context)
Thread(target=assert_different, args=(context,)).start()
while not test_result:
sleep(0.1)
self.assertFalse(test_result[0])
|
