🦅 Dramatiq Header Middleware for RabbitMQ

🛠 Installation

pip install dramatiq-header

⬆️ Upgrade version

pip install dramatiq-header --upgrade

✏️ Usage

Worker code:

import dramatiq
from dramatiq.brokers.rabbitmq import RabbitmqBroker
from dramatiq_header import HeadersMessage # Import Middleware


rabbitmq_broker = RabbitmqBroker()
dramatiq.set_broker(rabbitmq_broker)


rabbitmq_broker.add_middleware(HeadersMessage()) # Add Middleware


@dramatiq.actor(queue_name='example')
def my_task(message):
    print(f'Message Received: {message}')
    print(HeadersMessage.get_headers()) # Get headers

[!TIP] You can add middleware specifically to monitor a header key. For example:

rabbitmq_broker.add_middleware(HeadersMessage('x-test-header'))

Sender example

import dramatiq
from dramatiq import Message
from dramatiq.brokers.rabbitmq import RabbitmqBroker


rabbitmq_broker = RabbitmqBroker()
dramatiq.set_broker(rabbitmq_broker)


def send_message(msg: str):
    message = Message(
        queue_name='example',
        actor_name='my_task',
        args=(msg, ),
        kwargs={},
        options={'x-test-header': 'test-header'}, # Send your entire header here
    )
    rabbitmq_broker.enqueue(message)


if __name__ == '__main__':
    send_message('test message')

Output

[!IMPORTANT]
This library does NOT transmit the header using the RabbitMQ header property; rather, it sends the header as metadata within the message that Dramatiq already dispatches.