amqppy 0.0.19

Topic Subscriber

Firstly, we need to start the Topic Subscriber (also known as Consumer). In amqppy the class amqppy.consumer.Worker has this duty.

import amqppy
BROKER = 'amqp://guest:guest@localhost:5672//'

def on_topic_status(exchange, routing_key, headers, body):
    print('Received message from topic \'amqppy.publisher.topic.status\': {}'.format(body))

# subscribe to a topic: 'amqppy.publisher.topic.status'
worker = amqppy.Worker(BROKER)
worker.add_topic(exchange='amqppy.test',
                 routing_key='amqppy.publisher.topic.status',
                 on_topic_callback=on_topic_status)
# it will wait until worker is stopped or an uncaught exception
worker.run()

The subscriber worker will invoke the on_topic_callback every time a message is published with a topic that matches with the specified routing_key: ‘amqppy.publisher.topic.status’. Note that routing_key can contain wildcards therefore, one subscriber might be listening a set of Topics.

Once the topic subscriber is running we able to launch the publisher.

Topic Publisher

import amqppy
BROKER = 'amqp://guest:guest@localhost:5672//'

# publish my current status
amqppy.Topic(BROKER).publish(exchange='amqppy.test',
                             routing_key='amqppy.publisher.topic.status',
                             body='RUNNING')

The topic publisher will send a message to the AMQP exchange with the Topic routing_key: ‘amqppy.publisher.topic.status’, therefore, all the subscribed subscribers will receive the message unless they do not share the same queue. In case they share the same queue a round-robin dispatching policy would be applied among subscribers/consumers like happens in work queues.

RPC Reply

An object of type amqppy.consumer.Worker listens incoming RPC requests and computes the RPC reply in the on_request_callback. In the example below, the RPC consumer listens on Request rounting_key:’amqppy.requester.rpc.division’ and the division would be returned as the RPC reply.

import amqppy
BROKER = 'amqp://guest:guest@localhost:5672//'

def on_rpc_request_division(exchange, routing_key, headers, body):
    args = json.loads(body)
    return args['dividend'] / args['divisor']

# subscribe to a rpc request: 'amqppy.requester.rpc.division'
worker = Worker(BROKER)
worker.add_request(exchange='amqppy.test',
                   routing_key='amqppy.requester.rpc.division',
                   on_request_callback=on_rpc_request_division)
# it will wait until worker is stopped or an uncaught exception
worker.run()

RPC Request

The code below shows how to do a RPC Request using an instance of class amqppy.publisher.Rpc

import amqppy
BROKER = 'amqp://guest:guest@localhost:5672//'

# do a Rpc request 'amqppy.requester.rpc.division'
result = amqppy.Rpc(BROKER).request(exchange='amqppy.test',
                                    routing_key='amqppy.requester.rpc.division',
                                    body=json.dumps({'dividend': 3.23606797749979, 'divisor': 2.0}))
print('RPC result: {}.'.format(result))