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
|
from nose import SkipTest
from kombu import Consumer, Producer, Exchange, Queue
from kombu.five import range
from kombu.utils import nested
from funtests import transport
class test_mongodb(transport.TransportCase):
transport = 'mongodb'
prefix = 'mongodb'
event_loop_max = 100
def before_connect(self):
try:
import pymongo # noqa
except ImportError:
raise SkipTest('pymongo not installed')
def after_connect(self, connection):
connection.channel().client # evaluate connection.
self.c = self.connection # shortcut
def test_fanout(self, name='test_mongodb_fanout'):
if not self.verify_alive():
return
c = self.connection
self.e = Exchange(name, type='fanout')
self.q = Queue(name, exchange=self.e, routing_key=name)
self.q2 = Queue(name + '2', exchange=self.e, routing_key=name + '2')
channel = c.default_channel
producer = Producer(channel, self.e)
consumer1 = Consumer(channel, self.q)
consumer2 = Consumer(channel, self.q2)
self.q2(channel).declare()
for i in range(10):
producer.publish({'foo': i}, routing_key=name)
for i in range(10):
producer.publish({'foo': i}, routing_key=name + '2')
_received1 = []
_received2 = []
def callback1(message_data, message):
_received1.append(message)
message.ack()
def callback2(message_data, message):
_received2.append(message)
message.ack()
consumer1.register_callback(callback1)
consumer2.register_callback(callback2)
with nested(consumer1, consumer2):
while 1:
if len(_received1) + len(_received2) == 20:
break
c.drain_events(timeout=60)
self.assertEqual(len(_received1) + len(_received2), 20)
# queue.delete
for i in range(10):
producer.publish({'foo': i}, routing_key=name)
self.assertTrue(self.q(channel).get())
self.q(channel).delete()
self.q(channel).declare()
self.assertIsNone(self.q(channel).get())
# queue.purge
for i in range(10):
producer.publish({'foo': i}, routing_key=name + '2')
self.assertTrue(self.q2(channel).get())
self.q2(channel).purge()
self.assertIsNone(self.q2(channel).get())
|
