-
Notifications
You must be signed in to change notification settings - Fork 9
/
kombu_rabbitmq_subscriber.py
executable file
·50 lines (37 loc) · 1.27 KB
/
kombu_rabbitmq_subscriber.py
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
#!/usr/bin/env python
from spylunking.log.setup_logging import build_colorized_logger
from celery_connectors.utils import ev
from celery_connectors.kombu_subscriber import KombuSubscriber
name = "kombu-rabbitmq-subscriber"
log = build_colorized_logger(
name=name)
log.info("Start - {}".format(name))
recv_msgs = []
def handle_message(body, message):
log.info(("callback received msg "
"body={}")
.format(body))
recv_msgs.append(body)
message.ack()
# end of handle_message
# Initialize KombuSubscriber
ssl_options = {}
sub = KombuSubscriber("kombu-rabbitmq-subscriber",
ev("SUB_BROKER_URL",
"pyamqp://rabbitmq:rabbitmq@localhost:5672//"),
ssl_options)
# Now consume:
seconds_to_consume = 10.0
heartbeat = 60
serializer = "application/json"
exchange = ev("CONSUME_EXCHANGE", "reporting")
routing_key = ev("CONSUME_ROUTING_KEY", "reporting.accounts")
queue = ev("CONSUME_QUEUE", "reporting.accounts")
sub.consume(callback=handle_message,
queue=queue,
exchange=exchange,
routing_key=routing_key,
serializer=serializer,
heartbeat=heartbeat,
time_to_wait=seconds_to_consume)
log.info("End - {}".format(name))