45 lines
1.2 KiB
Python
45 lines
1.2 KiB
Python
from typing import TypedDict
|
|
|
|
import json
|
|
import pika
|
|
from pika.adapters.blocking_connection import BlockingChannel
|
|
from pika.spec import Basic
|
|
|
|
conn = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
|
|
ch = conn.channel()
|
|
|
|
ch.basic_qos(prefetch_count=1)
|
|
|
|
# hover over on_messae_callback
|
|
# the docs specify signature
|
|
|
|
|
|
class OrderMessage(TypedDict):
|
|
order_id: int
|
|
user_id: str
|
|
amount: float
|
|
should_fail: bool
|
|
retry_count: int
|
|
|
|
|
|
def on_order_processed(channel: BlockingChannel, method: Basic.Deliver, properties, body):
|
|
order: OrderMessage = json.loads(body)
|
|
print(f"processed order: {order["order_id"]}")
|
|
channel.basic_ack(delivery_tag=method.delivery_tag)
|
|
|
|
|
|
def on_order_failed(channel: BlockingChannel, method: Basic.Deliver, properties, body):
|
|
order: OrderMessage = json.loads(body)
|
|
print(f"failed after retries: {order["order_id"]}")
|
|
channel.basic_ack(delivery_tag=method.delivery_tag)
|
|
|
|
|
|
ch.basic_consume(queue="orders.processed.q",
|
|
on_message_callback=on_order_processed, auto_ack=False)
|
|
ch.basic_consume(queue="orders.failed.q",
|
|
on_message_callback=on_order_failed, auto_ack=False)
|
|
|
|
# easy to forget
|
|
print("[notifier] waiting on messages")
|
|
ch.start_consuming()
|