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
|
#!/usr/bin/env python
"""
Test AMQP library.
Repeatedly receive messages from the demo_send.py
script, until it receives a message with 'quit' as the body.
2007-11-11 Barry Pederson <bp@barryp.org>
"""
from optparse import OptionParser
from functools import partial
import amqp
def callback(channel, msg):
for key, val in msg.properties.items():
print('%s: %s' % (key, str(val)))
for key, val in msg.delivery_info.items():
print('> %s: %s' % (key, str(val)))
print('')
print(msg.body)
print('-------')
print(msg.delivery_tag)
channel.basic_ack(msg.delivery_tag)
#
# Cancel this callback
#
if msg.body == 'quit':
channel.basic_cancel(msg.consumer_tag)
def main():
parser = OptionParser()
parser.add_option(
'--host', dest='host',
help='AMQP server to connect to (default: %default)',
default='localhost',
)
parser.add_option(
'-u', '--userid', dest='userid',
help='userid to authenticate as (default: %default)',
default='guest',
)
parser.add_option(
'-p', '--password', dest='password',
help='password to authenticate with (default: %default)',
default='guest',
)
parser.add_option(
'--ssl', dest='ssl', action='store_true',
help='Enable SSL (default: not enabled)',
default=False,
)
options, args = parser.parse_args()
conn = amqp.Connection(options.host, userid=options.userid,
password=options.password, ssl=options.ssl)
ch = conn.channel()
ch.exchange_declare('myfan', 'fanout')
qname, _, _ = ch.queue_declare()
ch.queue_bind(qname, 'myfan')
ch.basic_consume(qname, callback=partial(callback, ch))
#pyamqp://
#
# Loop as long as the channel has callbacks registered
#
while ch.callbacks:
ch.wait()
ch.close()
conn.close()
if __name__ == '__main__':
main()
|
