AMQPStorm

To implement a Remote Procedure Call (RPC) server using AMQPStorm, you can use a pattern where the server listens on a specific queue, processes incoming messages, and sends a response back to the reply_to queue specified in the message properties.

In a typical RPC implementation:

1. The server consumes messages from a dedicated request queue.
2. For each message, the server performs the requested logic.
3. The server publishes the result to the queue name provided in the reply_to property of the incoming message.
4. The server uses the correlation_id from the request to ensure the client can match the response to the original request.

import amqpstorm

def on_request(channel, method, properties, body):
    print(f'Received request: {body.decode()}
')
    # Perform some work
    response = body.decode().upper()

    # Send the response back to the reply_to queue
    channel.basic_publish(
        exchange='',
        routing_key=properties.reply_to,
        body=response.encode(),
        correlation_id=properties.correlation_id
    )
    # Acknowledge the message
    channel.basic_ack(delivery_tag=method.delivery_tag)

def main():
    connection = amqpstorm.Connection({
        'host': 'localhost',
        'port': 5672,
        'login': 'guest',
        'password': 'guest'
    })
    channel = connection.channel()

    # Declare the queue for requests
    channel.queue_declare(queue='rpc_queue')

    # Set up the consumer
    channel.basic_consume(queue='rpc_queue', on_message_callback=on_request)

    print('RPC Server is running. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

To consume messages from an AMQP queue using amqpstorm, you need to establish a connection, create a channel, and then use the consume method on the channel. The consume method accepts a callback function that is invoked whenever a message is received.

Key steps:

1. Create an AMQPConnection with your broker credentials.
2. Open an AMQPChannel from the connection.
3. Call channel.consume(queue_name, callback) to start listening for messages.
4. The callback function receives the AMQPMessage object, which contains the body (the message payload) and other metadata.

import amqpstorm

def callback(message):
    print(f'Received message: {message.body}')
    message.ack()

connection = amqpstorm.Connection(host='localhost', username='guest', password='guest')
connection.connect()

channel = connection.channel()
channel.queue.declare(name='test_queue')

# Start consuming from the queue
channel.consume('test_queue', callback)

# Keep the script running to listen for messages
connection.process_events()

To build a resilient AMQP consumer that can handle connection interruptions and unexpected errors, you should implement a loop that manages the connection and channel lifecycle. A robust pattern involves:

1. Connection Management: Wrapping the AMQPStorm.Connection instantiation and the connection.start() call in a retry loop.
2. Channel Management: Creating the channel within the connection context.
3. Consumer Loop: Using channel.consume() to start receiving messages.
4. Error Handling: Catching exceptions within the message processing logic to prevent the entire consumer from crashing, and catching connection-level exceptions to trigger a reconnection attempt.
5. Graceful Shutdown: Using try...finally blocks to ensure connection.stop() is called, which closes the connection and cleans up resources.

import AMQPStorm
import time

def run_consumer():
    while True:
        connection = None
        try:
            # 1. Establish connection
            connection = AMQPStorm.Connection(
                host='localhost',
                login='guest',
                password='guest'
            )
            
            with connection:
                channel = connection.channel()
                
                # 2. Setup queue and consumer
                channel.queue_declare(queue='robust_queue')
                
                print('[*] Waiting for messages. To exit press CTRL+C')
                
                # 3. Consume messages
                for method, properties, body in channel.consume('robust_queue'):
                    try:
                        print(f' [x] Received {body!r}')
                        # Simulate work
                        time.sleep(1)
                        # 4. Acknowledge message
                        channel.basic_ack(delivery_tag=method.delivery_tag)
                    except Exception as e:
                        print(f' [!] Error processing message: {e}')
                        # 5. Nack message on processing error
                        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

        except (AMQPStorm.exceptions.AMQPError, ConnectionError) as e:
            print(f' [!] Connection error: {e}. Retrying in 5 seconds...')
            time.sleep(5)
        except KeyboardInterrupt:
            print(' [!] Stopping consumer...')
            break
        finally:
            if connection:
                connection.stop()

if __name__ == '__main__':
    run_consumer()