56 lines
1.9 KiB
Python
56 lines
1.9 KiB
Python
import pika
|
|
|
|
# this is the physical conn
|
|
connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
|
|
# a channel is a lightweight multiplexer over it
|
|
channel = connection.channel()
|
|
|
|
# first the exchanges
|
|
|
|
# this is A DECLARE, basically so can run multuple times
|
|
# .x = exchange
|
|
channel.exchange_declare(
|
|
exchange="orders.x", exchange_type="direct", durable=True)
|
|
# .dlx = dead letter exchange for failed messages
|
|
channel.exchange_declare(exchange="orders.dlx",
|
|
exchange_type="direct", durable=True)
|
|
# the exchange_types are
|
|
# - direct: exact key match for routing
|
|
# - topic: pattern based
|
|
# - fanout: all
|
|
# - headers: defined in headers
|
|
|
|
# then the queues
|
|
# when changing settings, you might need to delete q from adming
|
|
# as it'll aready exist and this declare does not delete
|
|
channel.queue_declare(
|
|
queue="orders.q",
|
|
durable=True,
|
|
arguments={
|
|
# what to do with "dead letters"
|
|
# which can be when consumder signals issue
|
|
# or q overflow
|
|
# or ttl stuff
|
|
"x-dead-letter-exchange": "orders.dlx",
|
|
"x-dead-letter-routing-key": "order.dead",
|
|
},
|
|
)
|
|
channel.queue_declare(queue="orders.processed.q", durable=True)
|
|
channel.queue_declare(queue="orders.failed.q", durable=True)
|
|
channel.queue_declare(queue="orders.dlq", durable=True)
|
|
|
|
# then the bindings
|
|
# these are rules on how exchange will route what message
|
|
# read as, on order.x, if key is order.created, push to orders.q
|
|
channel.queue_bind(exchange="orders.x", queue="orders.q",
|
|
routing_key="order.created")
|
|
channel.queue_bind(exchange="orders.x", queue="orders.processed.q",
|
|
routing_key="order.processed")
|
|
channel.queue_bind(exchange="orders.x", queue="orders.failed.q",
|
|
routing_key="order.failed")
|
|
channel.queue_bind(exchange="orders.dlx", queue="orders.dlq",
|
|
routing_key="order.dead")
|
|
|
|
|
|
connection.close()
|