AMQP 1.0 Client

Overview

The SwiftMQ AMQP 1.0 Client API is a standalone Java library for connecting to AMQP 1.0 brokers, sending and receiving messages, and managing messaging sessions and transactions. It provides direct, non-JMS access to AMQP concepts such as connections, sessions, producers, consumers, quality of service (QoS) modes, and transactional message delivery. The API supports advanced features like durable subscriptions, message selectors, and explicit QoS control.

Connection and Session Setup

To interact with an AMQP broker, instantiate a Connection with the target hostname, port, and optional credentials. Authentication uses SASL (PLAIN, CRAM-MD5, DIGEST-MD5, or ANONYMOUS). Before calling connect(), you may configure properties such as maxFrameSize, idleTimeout, or the SASL mechanism via setMechanism(). The containerId is automatically generated, but can be set explicitly (required for durable consumers). After connecting, create a Session using createSession(incomingWindowSize, outgoingWindowSize). Sessions are the context for producers, consumers, and transactions.

import com.swiftmq.amqp.v100.client.Connection;
import com.swiftmq.amqp.v100.client.Session;

Connection connection = new Connection(ctx, "broker.example.com", 5672, "user", "password");
connection.setContainerId("my-client-id");
connection.connect();
Session session = connection.createSession(100, 100);

Producer and Consumer Creation

Producers are created with Session.createProducer(target, qos), where target is the queue or topic name and qos is the desired quality of service. Consumers are created with Session.createConsumer(source, linkCredit, qos, noLocal, selector), where source is the queue or topic name, linkCredit controls prefetch, and selector is an optional message selector string. For queues, use the full queuename@routername format; for topics, use just the topic name. Durable consumers are supported via createDurableConsumer(linkName, source, ...) and require a set containerId.

// Producer example
Producer producer = session.createProducer("myqueue@router1", QoS.AT_LEAST_ONCE);

// Consumer example
Consumer consumer = session.createConsumer("myqueue@router1", 100, QoS.AT_LEAST_ONCE, false, null);

Quality of Service (QoS) Levels

The API supports three QoS levels: QoS.AT_MOST_ONCE (0), QoS.AT_LEAST_ONCE (1), and QoS.EXACTLY_ONCE (2). AT_MOST_ONCE settles messages before send, offering best performance but no delivery guarantees. AT_LEAST_ONCE settles after receive, ensuring at-least-once delivery. EXACTLY_ONCE uses two-phase settlement for strict once-only delivery. Specify the QoS when creating producers or consumers. For AT_LEAST_ONCE and EXACTLY_ONCE, received messages must be explicitly accepted or rejected by the application.

// Creating a producer with EXACTLY_ONCE
Producer producer = session.createProducer("myqueue@router1", QoS.EXACTLY_ONCE);

Message Sending and Receiving

To send a message, create an AMQPMessage, set its body and properties, and call Producer.send(). You can specify persistence, priority, and TTL. For consumers, use Consumer.receive() (blocking), receive(timeout), or receiveNoWait() (non-blocking). For QoS modes other than AT_MOST_ONCE, after processing a message, call accept() or reject() on the AMQPMessage to settle delivery. If the message is already settled, these calls are not required.

// Sending a message
AMQPMessage msg = new AMQPMessage();
msg.setAmqpValue(new AmqpValue("Hello, world!"));
producer.send(msg);

// Receiving and accepting a message
AMQPMessage received = consumer.receive();
if (received != null && !received.isSettled()) {
    received.accept();
}

Transactions

Transactional messaging is supported via the TransactionController obtained from a session. Begin a transaction with createTxnId(), then send or acquire messages within the transaction by associating the TxnIdIF with messages or consumer acquisition. Commit or rollback with commit(txnId) or rollback(txnId). Transactional state is handled automatically when sending or settling messages with an associated transaction ID.

// Transactional send
TransactionController txCtrl = session.getTransactionController();
TxnIdIF txnId = txCtrl.createTxnId();
AMQPMessage msg = new AMQPMessage();
msg.setAmqpValue(new AmqpValue("Transactional message"));
msg.setTxnIdIF(txnId);
producer.send(msg);
txCtrl.commit(txnId);

Error Handling

The API throws checked exceptions for connection, authentication, protocol, and state errors. Register an ExceptionListener on the Connection to receive asynchronous notifications of connection closure or fatal errors. If a link or session is closed unexpectedly, pending operations are notified and further use will throw exceptions. Always close connections, sessions, producers, and consumers explicitly to release resources.

// Registering an exception listener
connection.setExceptionListener(new ExceptionListener() {
    public void onException(Exception e) {
        System.err.println("Connection error: " + e);
    }
});