curv-amqp 1.0.0

No project description provided

curv_amqp

Pika framework that handles reconnecting while using a blocking connection and has helpful defaults, building blocks, type hints, and a priority requeue method

Usage

from argparse import ArgumentParser

from curv_amqp.connection import Connection, URLParameters, ConnectionParameters
from curv_amqp.publisher import Publisher
from curv_amqp.consumer import Consumer, ConsumerMessage


def on_message_callback(message: ConsumerMessage):
    print('message.body:', message.body)
    message.ack()
    message.consumer.stop_consuming()


def main():
    parser = ArgumentParser()
    parser.add_argument('--url', type=str, default='localhost', help='amqp url or localhost - '
                                                                     'localhost assumes rabbitmq is installed - '
                                                                     '"brew install rabbitmq"')
    parser.add_argument('--queue', type=str, default='test-queue-name', help='amqp queue name')
    parser.add_argument('--body', type=str, default='your message', help='amqp message body')
    args = parser.parse_args()
    # pass in URLParameters or ConnectionParameters
    # its recommended that a single connection per process is used.
    url: str = args.url
    parameters = ConnectionParameters(url) if url is 'localhost' else URLParameters(url)
    queue_name = args.queue
    body = bytes(args.body, encoding='utf-8')

    connection = Connection(parameters=parameters)
    # its required that two different channels are used for a publisher and consumer
    # NOTE: will automatically declare queue for you
    publisher = Publisher(connection=connection)
    publisher.publish(routing_key=queue_name, body=body)

    consumer = Consumer(connection=connection)
    consumer.consume(queue=queue_name, prefetch_count=1, on_message_callback=on_message_callback)


if __name__ == '__main__':
    main()