15 AMQP Operations: Deployment and Broker Health

15.8.1 Publisher with Confirms

import pika
import json
from typing import Dict, Any

class AMQPPublisher:
    """Production-ready AMQP publisher with reliability features."""

    def __init__(self, host: str, exchange: str):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(
                host=host,
                heartbeat=600,  # Detect dead connections
                blocked_connection_timeout=300
            )
        )
        self.channel = self.connection.channel()
        self.exchange = exchange

        # Enable publisher confirms
        self.channel.confirm_delivery()

        # Declare durable exchange
        self.channel.exchange_declare(
            exchange=exchange,
            exchange_type='topic',
            durable=True
        )

    def publish(self, routing_key: str, message: Dict[str, Any],
                message_id: str = None) -> bool:
        """
        Publish message with persistence and confirmation.

        Args:
            routing_key: Topic routing key (e.g., 'sensor.temperature.line1')
            message: Dictionary to serialize as JSON
            message_id: Unique ID for idempotency (optional)

        Returns:
            True if broker acknowledged, False otherwise
        """
        properties = pika.BasicProperties(
            delivery_mode=2,  # Persistent
            content_type='application/json',
            message_id=message_id
        )

        try:
            self.channel.basic_publish(
                exchange=self.exchange,
                routing_key=routing_key,
                body=json.dumps(message),
                properties=properties,
                mandatory=True  # Return if unroutable
            )
            return True
        except pika.exceptions.UnroutableError:
            print(f"Message unroutable: {routing_key}")
            return False

    def close(self):
        """Clean shutdown."""
        self.connection.close()


# Usage example
publisher = AMQPPublisher('localhost', 'sensor-data')
publisher.publish(
    routing_key='sensor.temperature.line1.machine3',
    message={'temp': 75.3, 'timestamp': 1698765432},
    message_id='msg-001'
)

15.8.2 Consumer with Manual Acknowledgment

Read the Python consumer from queue declaration through bindings and prefetch before tracing the callback outcomes. The success path acknowledges only after processing; retryable failure requeues; terminal failure rejects without requeue so the configured dead-letter exchange preserves evidence.

import pika
import json
from typing import Callable

class AMQPConsumer:
    """Production-ready AMQP consumer with reliability features."""

    def __init__(self, host: str, queue: str, exchange: str,
                 binding_patterns: list, prefetch_count: int = 10):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters(host=host)
        )
        self.channel = self.connection.channel()
        self.queue = queue

        # Declare durable queue with dead letter exchange
        self.channel.queue_declare(
            queue=queue,
            durable=True,
            arguments={
                'x-dead-letter-exchange': 'dlx',
                'x-dead-letter-routing-key': f'{queue}.failed'
            }
        )

        # Bind to exchange with patterns
        for pattern in binding_patterns:
            self.channel.queue_bind(
                exchange=exchange,
                queue=queue,
                routing_key=pattern
            )

        # Set prefetch for batching
        self.channel.basic_qos(prefetch_count=prefetch_count)

    def consume(self, callback: Callable):
        """
        Start consuming messages with manual acknowledgment.

        Args:
            callback: Function(body: dict) -> bool
                     Returns True if processed successfully
        """
        def on_message(channel, method, properties, body):
            try:
                message = json.loads(body)
                success = callback(message)

                if success:
                    channel.basic_ack(delivery_tag=method.delivery_tag)
                else:
                    # Requeue for retry
                    channel.basic_nack(
                        delivery_tag=method.delivery_tag,
                        requeue=True
                    )
            except Exception as e:
                print(f"Processing error: {e}")
                # Send to dead letter queue
                channel.basic_nack(
                    delivery_tag=method.delivery_tag,
                    requeue=False
                )

        self.channel.basic_consume(
            queue=self.queue,
            on_message_callback=on_message,
            auto_ack=False  # Manual acknowledgment
        )

        print(f"Consuming from {self.queue}...")
        self.channel.start_consuming()


# Usage example
def process_sensor_reading(message: dict) -> bool:
    """Process a sensor reading. Returns True if successful."""
    print(f"Received: {message}")
    # Your processing logic here
    return True

consumer = AMQPConsumer(
    host='localhost',
    queue='analytics',
    exchange='sensor-data',
    binding_patterns=['sensor.#'],
    prefetch_count=100  # Batch processing
)
consumer.consume(process_sensor_reading)

15.9.1 Publisher with RabbitMQ Java Client

import com.rabbitmq.client.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class AMQPPublisher implements AutoCloseable {
    private final Connection connection;
    private final Channel channel;
    private final String exchange;
    private final ObjectMapper mapper = new ObjectMapper();

    public AMQPPublisher(String host, String exchange)
            throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setRequestedHeartbeat(60);

        this.connection = factory.newConnection();
        this.channel = connection.createChannel();
        this.exchange = exchange;

        // Enable publisher confirms
        channel.confirmSelect();

        // Declare durable topic exchange
        channel.exchangeDeclare(exchange, "topic", true);
    }

    public boolean publish(String routingKey, Object message,
            String messageId) throws Exception {
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
            .deliveryMode(2)  // Persistent
            .contentType("application/json")
            .messageId(messageId)
            .build();

        byte[] body = mapper.writeValueAsBytes(message);

        channel.basicPublish(exchange, routingKey, true, properties, body);

        // Wait for confirm (blocking)
        return channel.waitForConfirms(5000);
    }

    @Override
    public void close() throws Exception {
        channel.close();
        connection.close();
    }
}

15.9.2 Consumer with Manual Acknowledgment

Read the Java consumer in the same order as the Python contract: durable queue and dead-letter arguments first, binding patterns second, bounded prefetch third, and explicit basicAck or basicNack only after the callback reports its outcome.

import com.rabbitmq.client.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.util.Map;
import java.util.function.Function;

public class AMQPConsumer {
    private final Channel channel;
    private final String queue;
    private final ObjectMapper mapper = new ObjectMapper();

    public AMQPConsumer(String host, String queue, String exchange,
            String[] bindingPatterns, int prefetchCount) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);

        Connection connection = factory.newConnection();
        this.channel = connection.createChannel();
        this.queue = queue;

        // Declare queue with DLX
        Map<String, Object> args = Map.of(
            "x-dead-letter-exchange", "dlx",
            "x-dead-letter-routing-key", queue + ".failed"
        );
        channel.queueDeclare(queue, true, false, false, args);

        // Bind patterns
        for (String pattern : bindingPatterns) {
            channel.queueBind(queue, exchange, pattern);
        }

        // Set prefetch
        channel.basicQos(prefetchCount);
    }

    public void consume(Function<Map<String, Object>, Boolean> callback)
            throws IOException {
        DeliverCallback deliverCallback = (tag, delivery) -> {
            try {
                Map<String, Object> message = mapper.readValue(
                    delivery.getBody(), Map.class);

                boolean success = callback.apply(message);

                if (success) {
                    channel.basicAck(delivery.getEnvelope()
                        .getDeliveryTag(), false);
                } else {
                    channel.basicNack(delivery.getEnvelope()
                        .getDeliveryTag(), false, true);
                }
            } catch (Exception e) {
                channel.basicNack(delivery.getEnvelope()
                    .getDeliveryTag(), false, false);
            }
        };

        channel.basicConsume(queue, false, deliverCallback, tag -> {});
    }
}

15.10.1 Publisher with amqplib

const amqp = require('amqplib');

class AMQPPublisher {
    constructor() {
        this.connection = null;
        this.channel = null;
    }

    async connect(host, exchange) {
        this.connection = await amqp.connect(`amqp://${host}`);
        this.channel = await this.connection.createConfirmChannel();
        this.exchange = exchange;

        // Declare durable topic exchange
        await this.channel.assertExchange(exchange, 'topic', {
            durable: true
        });
    }

    async publish(routingKey, message, messageId = null) {
        const options = {
            persistent: true,  // delivery_mode = 2
            contentType: 'application/json',
            messageId: messageId
        };

        return new Promise((resolve, reject) => {
            this.channel.publish(
                this.exchange,
                routingKey,
                Buffer.from(JSON.stringify(message)),
                options,
                (err) => {
                    if (err) reject(err);
                    else resolve(true);
                }
            );
        });
    }

    async close() {
        await this.channel.close();
        await this.connection.close();
    }
}

// Usage
(async () => {
    const publisher = new AMQPPublisher();
    await publisher.connect('localhost', 'sensor-data');

    await publisher.publish(
        'sensor.temperature.line1.machine3',
        { temp: 75.3, timestamp: Date.now() },
        'msg-001'
    );

    await publisher.close();
})();

15.10.2 Consumer with Manual Acknowledgment

Read the Node.js consumer from connection and queue setup into pattern bindings, prefetch, and the asynchronous callback. The final ACK, requeueing NACK, or dead-lettering NACK must describe processing outcome; receiving the message alone is not completion evidence.

const amqp = require('amqplib');

class AMQPConsumer {
    async connect(host, queue, exchange, bindingPatterns, prefetchCount = 10) {
        this.connection = await amqp.connect(`amqp://${host}`);
        this.channel = await this.connection.createChannel();
        this.queue = queue;

        // Declare queue with DLX
        await this.channel.assertQueue(queue, {
            durable: true,
            arguments: {
                'x-dead-letter-exchange': 'dlx',
                'x-dead-letter-routing-key': `${queue}.failed`
            }
        });

        // Bind patterns
        for (const pattern of bindingPatterns) {
            await this.channel.bindQueue(queue, exchange, pattern);
        }

        // Set prefetch
        await this.channel.prefetch(prefetchCount);
    }

    async consume(callback) {
        console.log(`Consuming from ${this.queue}...`);

        this.channel.consume(this.queue, async (msg) => {
            if (msg === null) return;

            try {
                const message = JSON.parse(msg.content.toString());
                const success = await callback(message);

                if (success) {
                    this.channel.ack(msg);
                } else {
                    this.channel.nack(msg, false, true);  // Requeue
                }
            } catch (error) {
                console.error('Processing error:', error);
                this.channel.nack(msg, false, false);  // Dead letter
            }
        }, { noAck: false });
    }
}

// Usage
(async () => {
    const consumer = new AMQPConsumer();
    await consumer.connect(
        'localhost',
        'analytics',
        'sensor-data',
        ['sensor.#'],
        100
    );

    await consumer.consume((message) => {
        console.log('Received:', message);
        return true;  // Processing successful
    });
})();