Classic Messaging Kombu
This package provides implementation of interfaces in classic-messaging and base for consumers and messages handling.
Usage with publishing:
from classic.components import component
from classic.messaging import Message, Publisher
from classic.messaging_kombu import BrokerScheme, KombuPublisher
from kombu import Exchange, Queue, Connection
@component
class SomeService:
publisher: Publisher
def do_some_work(self):
message = Message('some', 'Some very useful info')
self.publisher.publish(message)
broker_scheme = BrokerScheme(
Queue('queue1', Exchange('some')),
)
connection = Connection('amqp://localhost:5672/')
publisher = KombuPublisher(
connection=connection,
scheme=broker_scheme
)
service = SomeService(publisher=publisher)
service.do_some_work()Message have 2 arguments - target and body. Target is a str with destination. In simple case it is an exchange name, in complex case - entry in mapping.
Body can be any serializable object.
Customization of target:
from classic.messaging import Message
from classic.messaging_kombu import BrokerScheme, KombuPublisher
from kombu import Exchange, Queue, Connection
targets = {
'FIRST': {
'exchange': 'exchange1',
},
'SECOND': {
'exchange': 'exchange2',
'timeout': 10,
}
}
broker_scheme = BrokerScheme(
Queue('queue1', Exchange('exchange1')),
Queue('queue2', Exchange('exchange2')),
)
connection = Connection('amqp://localhost:5672/')
publisher = KombuPublisher(
connection=connection,
scheme=broker_scheme,
messages_params=targets,
)
publisher = KombuPublisher(
connection=connection,
scheme=broker_scheme
)
publisher.publish(
Message('FIRST', 'Hello'), # Will be published to exchange1 and queue1
Message('SECOND', 'By'), # Will be published to exchange2, queue2 and timeout=10
)Usage with consuming:
from classic.messaging_kombu import BrokerScheme, KombuConsumer, MessageHandler
from kombu import Exchange, Queue, Connection
class SomeSerice:
def handle_message(self, message):
print(message)
class CustomHandler(MessageHandler):
def handle(self, message, body):
print(body)
message.ack()
broker_scheme = BrokerScheme(
Queue('queue1', Exchange('exchange1')),
Queue('queue2', Exchange('exchange2')),
)
connection = Connection('amqp://localhost:5672/')
consumer = KombuConsumer(
connection=connection,
scheme=broker_scheme,
)
service = SomeSerice()
handler = CustomHandler()
consumer.register_function(service.handle_message, 'queue1')
consumer.register_handler(handler, 'queue2')
