1.

Contents
Messaging Scenarios – Ecommerce............................................................................................................................ 4
Wholesaling And Warehousing............................................................................................................................. 4
2. Java Client API Guide........................................................................................................................................ 8
Overview................................................................................................................................................................ 9
3. Java Client API Guide...................................................................................................................................... 12
Overview................................................................................................................................................................. 13
Connections and Channels.................................................................................................................................... 14
Connecting to RabbitMQ....................................................................................................................................... 14
Disconnecting from RabbitMQ.............................................................................................................................. 16
Connection and Channel Lifespan......................................................................................................................... 17
Using Exchanges and Queues................................................................................................................................ 17
Passive Declaration............................................................................................................................................. 19
Operations with Optional Responses................................................................................................................. 20
Deleting Entities and Purging Messages........................................................................................................... 20
Publishing Messages.............................................................................................................................................. 21
Channels and Concurrency Considerations (Thread Safety)............................................................................... 23
Receiving Messages by Subscription ("Push API").............................................................................................. 25
 Retrieving Individual Messages ("Pull API")....................................................................................................... 28
Handling unroutable messages............................................................................................................................. 29
Shutdown Protocol................................................................................................................................................ 30
Overview of the Client Shutdown Process......................................................................................................... 30
Information about the circumstances of a shutdown........................................................................................ 31
Atomicity and use of the isOpen() method........................................................................................................ 33
Advanced Connection options............................................................................................................................... 34
Consumer thread pool........................................................................................................................................ 34
4. Java Client API Guide...................................................................................................................................... 35
Overview................................................................................................................................................................. 37
Connections and Channels.................................................................................................................................... 37
Connecting to RabbitMQ....................................................................................................................................... 38
Disconnecting from RabbitMQ............................................................................................................................. 40
Connection and Channel Lifespan........................................................................................................................ 40
Using Exchanges and Queues................................................................................................................................ 41
Passive Declaration............................................................................................................................................. 42
Operations with Optional Responses................................................................................................................. 43
Deleting Entities and Purging Messages............................................................................................................ 44
Publishing Messages.............................................................................................................................................. 44
 Channels and Concurrency Considerations (Thread Safety)............................................................................... 46
Receiving Messages by Subscription ("Push API").............................................................................................. 48
Retrieving Individual Messages ("Pull API")........................................................................................................ 51
Handling unroutable messages............................................................................................................................. 52
Shutdown Protocol................................................................................................................................................ 53
Overview of the Client Shutdown Process......................................................................................................... 53
Information about the circumstances of a shutdown........................................................................................ 54
Atomicity and use of the isOpen() method........................................................................................................ 56
Advanced Connection options............................................................................................................................... 57
Consumer thread pool........................................................................................................................................ 57
Using Lists of Hosts............................................................................................................................................ 58
5. Security............................................................................................................................................................. 60
Enabling SSL Support in RabbitMQ...................................................................................................................... 70
Trust the Client's Root CA................................................................................................................................... 73
Key Managers, Trust Managers and Stores......................................................................................................... 73
Connecting without validating certificates.......................................................................................................... 74
 Messaging Scenarios – Ecommerce.

Is your Point Of Sale software communicating with your e-commerce business? If not, you could be
losing money.

We can integrate QuickBook’s POS software with your online store, allowing your two systems to work as one.
Our Integrator enables communication between QuickBooks POS and your website so that information is
updated quickly and easily.

Platforms
Whether your e-commerce business has been built on a Bigcommerce, Magento, Shopify or
WooCommerce platform, we can help. Our integrator is designed to work with many platforms, allowing your
POS system and e-commerce business to share information. Simply select your platform and  purchase our
QuickBooks POS Integrator.

Benefits of Modern Retail’s Integrator:

Aside from making your life a whole lot easier, following are a few things that our integrator can do:

 Pull product catalog updates from your QuickBooks POS system and send it to your website.
 Send completed order and customer information to your POS system.
 Provide up-to-date pricing and inventory information with no intervention from you.
 Allow editing of information coming from your POS system so you can merchandise products in a way
that sells online.
The bottom line is that integrating your QuickBooks POS system saves you time, improves efficiencies and
increases your customer satisfaction.
Wholesaling And Warehousing
Wholesaling and warehousing ecommerce businesses require a lot of investment at the start – you need to manage
inventory and stock, keep track of customer orders and shipping information, and invest in the warehouse space
itself.

DollarDays is an online wholesaler with a massive product catalog that includes more than 260,000 products. They
employ a key strategy for retailers in this space – by offering case prices AND piece prices, they can sell to the
general public and to retailers. This gives them a higher profit margin than a strictly wholesale model.

Highlights the message to be sent in between various stake holders?

What will the traditional ways? Each having different technologies . Integration in Ecommerce
Domain?
Ask ? What will be the solution? Solve using Messaging with the advantages.

Another case: Business Process with Microservices. How micro services need Messaging.
Taxi-hailing application
The following diagram shows how services in a taxi-hailing application might interact when the user requests a
trip.
The services use a combination of notifications, request/response, and publish/subscribe. For example, the
passenger’s smartphone sends a notification to the Trip Management service to request a pickup. The
Trip Management service verifies that the passenger’s account is active by using request/response to invoke the
Passenger Service. The Trip Management service then creates the trip and uses publish/subscribe to notify other
services including the Dispatcher, which locates an available driver.
Now that we have looked at interaction styles, let’s take a look at how to define APIs.
[Link]
 2. Java Client API Guide

This guide covers RabbitMQ Java client and its public API. It assumes that the most recent major
version of the client is used and the reader is familiar with the basics. Key sections of the guide
are:

 Connecting to RabbitMQ
 Connection and Channel Lifespan
 Using Exchanges and Queues
 Publishing Messages
 Consuming Using a Subscription
 Concurrency Considerations and Safety
 Automatic Recovery From Network Failures

5.x release series of this library require JDK 8, both for compilation and at runtime. On Android,
this means only Android 7.0 or later versions are supported. 4.x release series support JDK 6 and
Android versions prior to 7.0.

The library is open source, developed on GitHub, and is triple-licensed under

 Apache Public License 2.0

 Mozilla Public License
 GPL 2.0
 This means that the user can consider the library to be licensed under any of the licenses from the
list above. For example, the user may choose the Apache Public License 2.0 and include this
client into a commercial product. Codebases that are licensed under the GPLv2 may choose
GPLv2, and so on.

There are also command line tools that used to be shipped with the Java client.

The client API is closely modelled on the AMQP 0-9-1 protocol model, with additional
abstractions for ease of use.

An API reference (JavaDoc) is available separately.

Overview

RabbitMQ Java client uses [Link] as its top-level package. The key classes and
interfaces are:

 Channel: represents an AMQP 0-9-1 channel, and provides most of the operations (protocol
methods).
 Connection: represents an AMQP 0-9-1 connection
 ConnectionFactory: constructs Connection instances
 Consumer: represents a message consumer
 DefaultConsumer: commonly used base class for consumers
 BasicProperties: message properties (metadata)
 [Link]: builder for BasicProperties
Protocol operations are available through the Channel interface. Connection is used to open
channels, register connection lifecycle event handlers, and close connections that are no longer
needed. Connections are instantiated through ConnectionFactory, which is how you configure
various connection settings, such as the vhost or username.

This sends a message with delivery mode 2 (persistent), priority 1 and content-type "text/plain".
You can build your own message properties object, using a Builder class mentioning as many
properties as you like, for example:
The easiest way to implement a Consumer is to subclass the convenience class DefaultConsumer.
An object of this subclass can be passed on a basicConsume call to set up the subscription:
Here, since we specified autoAck = false, it is necessary to acknowledge messages delivered to
the Consumer, most conveniently done in the handleDelivery method, as illustrated.
3. Java Client API Guide

This guide covers RabbitMQ Java client and its public API. It assumes that the most recent major
version of the client is used and the reader is familiar with the basics. Key sections of the guide
are:

 Connecting to RabbitMQ
 Connection and Channel Lifespan
 Using Exchanges and Queues
 Publishing Messages
 Consuming Using a Subscription
 Concurrency Considerations and Safety
 Automatic Recovery From Network Failures

5.x release series of this library require JDK 8, both for compilation and at runtime. On Android,
this means only Android 7.0 or later versions are supported. 4.x release series support JDK 6 and
Android versions prior to 7.0.

The library is open source, developed on GitHub, and is triple-licensed under

 Apache Public License 2.0

 Mozilla Public License
 GPL 2.0
 This means that the user can consider the library to be licensed under any of the licenses from the
list above. For example, the user may choose the Apache Public License 2.0 and include this
client into a commercial product. Codebases that are licensed under the GPLv2 may choose
GPLv2, and so on.

There are also command line tools that used to be shipped with the Java client.

The client API is closely modelled on the AMQP 0-9-1 protocol model, with additional
abstractions for ease of use.

An API reference (JavaDoc) is available separately.

Overview

RabbitMQ Java client uses [Link] as its top-level package. The key classes and
interfaces are:

 Channel: represents an AMQP 0-9-1 channel, and provides most of the operations (protocol
methods).
 Connection: represents an AMQP 0-9-1 connection
 ConnectionFactory: constructs Connection instances
 Consumer: represents a message consumer
 DefaultConsumer: commonly used base class for consumers
 BasicProperties: message properties (metadata)
 [Link]: builder for BasicProperties
Protocol operations are available through the Channel interface. Connection is used to open
channels, register connection lifecycle event handlers, and close connections that are no longer
needed. Connections are instantiated through ConnectionFactory, which is how you configure
various connection settings, such as the vhost or username.

Connections and Channels

The core API classes are Connection and Channel, representing an AMQP 0-9-1 connection and
channel, respectively. They are typically imported before used:

import [Link];
import [Link];

Connecting to RabbitMQ

The following code connects to a RabbitMQ node using the given parameters (host name, port
number, etc):

ConnectionFactory factory = new ConnectionFactory();

// "guest"/"guest" by default, limited to localhost connections
[Link](userName);
[Link](password);
[Link](virtualHost);
[Link](hostName);
[Link](portNumber);

Connection conn = [Link]();

All of these parameters have sensible defaults for a RabbitMQ node running locally. The default
value for a property will be used if the property remains unassigned prior to creating a
connection:

Property Default Value

Username "guest"

Password "guest"

Virtual "/"
host

Hostname "localhost"

port 5672 for regular connections, 5671 for connections that use

TLS

Alternatively, URIs may be used:

ConnectionFactory factory = new ConnectionFactory();

[Link]("amqp://userName:password@hostName:portNumber/virtualHost");
Connection conn = [Link]();

All of these parameters have sensible defaults for a stock RabbitMQ server running locally.

Note that user guest can only connect from localhost  by default. This is to limit well-known
credential use in production systems.

The Connection interface can then be used to open a channel:

Channel channel = [Link]();

The channel can now be used to send and receive messages, as described in subsequent sections.

Successful and unsuccessful client connection events can be observed in server node logs.

Disconnecting from RabbitMQ

To disconnect, simply close the channel and the connection:

[Link]();
[Link]();

Note that closing the channel may be considered good practice, but isn’t strictly necessary here -
it will be done automatically anyway when the underlying connection is closed.

Client disconnection events can be observed in server node logs .

Connection and Channel Lifespan

Connections are meant to be long-lived. The underlying protocol is designed and optimized for
long running connections. That means that opening a new connection per operation, e.g. a
message published, is unnecessary and strongly discouraged as it will introduce a lot of network
roundtrips and overhead.

Channels are also meant to be long-lived but since many recoverable protocol errors will result in
channel closure, channel lifespan could be shorter than that of its connection. Closing and
opening new channels per operation is usually unnecessary but can be appropriate. When in
doubt, consider reusing channels fist.

Channel-level exceptions such as attempts to consume from a queue that does not exist will result
in channel closure. A closed channel can no longer be used and will not receive any more events
from the server (such as message deliveries). Channel-level exceptions will be logged by
RabbitMQ and will initiate a shutdown sequence for the channel (see below).

Using Exchanges and Queues

Client applications work with exchanges and queues, the high-level building blocks of the
protocol. These must be declared before they can be used. Declaring either type of object simply
ensures that one of that name exists, creating it if necessary.

Continuing the previous example, the following code declares an exchange and a server-named
queue, then binds them together.
[Link](exchangeName, "direct", true);
String queueName = [Link]().getQueue();
[Link](queueName, exchangeName, routingKey);

This will actively declare the following objects, both of which can be customised by using
additional parameters. Here neither of them have any special arguments.

1. a durable, non-autodelete exchange of "direct" type

2. a non-durable, exclusive, autodelete queue with a generated name

The above function calls then bind the queue to the exchange with the given routing key.

Note that this would be a typical way to declare a queue when only one client wants to work with
it: it doesn’t need a well-known name, no other client can use it (exclusive) and will be cleaned up
automatically (autodelete). If several clients want to share a queue with a well-known name, this
code would be appropriate:

[Link](exchangeName, "direct", true);

[Link](queueName, true, false, false, null);
[Link](queueName, exchangeName, routingKey);

This will actively declare:

1. a durable, non-autodelete exchange of "direct" type

2. a durable, non-exclusive, non-autodelete queue with a well-known name
Many Channel API methods are overloaded. These convenient short forms
of exchangeDeclare, queueDeclare and queueBind use sensible defaults. There are also longer
forms with more parameters, to let you override these defaults as necessary, giving full control
where needed.

This "short form, long form" pattern is used throughout the client API uses.

Passive Declaration

Queues and exchanges can be declared "passively". A passive declare simply checks that the
entity with the provided name exists. If it does, the operation is a no-op. For queues successful
passive declares will return the same information as non-passive ones, namely the number of
consumers and messages in ready state in the queue. If the entity does not exist, the operation
fails with a channel level exception. The channel cannot be used after that. A new channel should
be opened. It is common to use one-off (temporary) channels for passive declarations.

Channel#queueDeclarePassive and Channel#exchangeDeclarePassive are the methods used for

passive declaration. The following example demonstrates Channel#queueDeclarePassive:

[Link] response = [Link]("queue-name");

// returns the number of messages in Ready state in the queue
[Link]();
// returns the number of consumers the queue has
[Link]();
Channel#exchangeDeclarePassive's return value contains no useful information. Therefore if the
method returns and no channel exceptions occurs, it means that the exchange does exist.
Operations with Optional Responses

Some common operations also have a "no wait" version which won't wait for server response. For
example, to declare a queue and instruct the server to not send any response, use

[Link](queueName, true, false, false, null);

The "no wait" versions are more efficient but offer lower safety guarantees, e.g. they are more
dependent on the heartbeat mechanism for detection of failed operations. When in doubt, start
with the standard version. The "no wait" versions are only needed in scenarios with high topology
(queue, binding) churn.
Deleting Entities and Purging Messages

A queue or exchange can be explicitly deleted:

[Link]("queue-name")
It is possible to delete a queue only if it is empty:
[Link]("queue-name", false, true)
or if it is not used (does not have any consumers):
[Link]("queue-name", true, false)
A queue can be purged (all of its messages deleted):

[Link]("queue-name")

Publishing Messages

To publish a message to an exchange, use [Link] as follows:

byte[] messageBodyBytes = "Hello, world!".getBytes();

[Link](exchangeName, routingKey, null, messageBodyBytes);

For fine control, you can use overloaded variants to specify the mandatory flag, or send messages
with pre-set message properties:

[Link](exchangeName, routingKey, mandatory,

MessageProperties.PERSISTENT_TEXT_PLAIN,
messageBodyBytes);

This sends a message with delivery mode 2 (persistent), priority 1 and content-type "text/plain".
You can build your own message properties object, using a Builder class mentioning as many
properties as you like, for example:

[Link](exchangeName, routingKey,
new [Link]()
.contentType("text/plain")
.deliveryMode(2)
 .priority(1)
.userId("bob")
.build()),
messageBodyBytes);

This example publishes a message with custom headers:

Map<String, Object> headers = new HashMap<String, Object>();

[Link]("latitude", 51.5252949);
[Link]("longitude", -0.0905493);

[Link](exchangeName, routingKey,
new [Link]()
.headers(headers)
.build()),
messageBodyBytes);

This example publishes a message with expiration:

[Link](exchangeName, routingKey,
new [Link]()
.expiration("60000")
.build()),
messageBodyBytes);
We have not illustrated all the possibilities here.

Note that BasicProperties is an inner class of the autogenerated holder class AMQP.

Invocations of Channel#basicPublish will eventually block if a resource-driven alarm is in effect.

Channels and Concurrency Considerations (Thread Safety)

As a rule of thumb, sharing Channel instances between threads is something to be avoided.

Applications should prefer using a Channel per thread instead of sharing the same Channelacross
multiple threads.

While some operations on channels are safe to invoke concurrently, some are not and will result
in incorrect frame interleaving on the wire, double acknowledgements and so on.

Concurrent publishing on a shared channel can result in incorrect frame interleaving on the wire,
triggering a connection-level protocol exception and immediate connection closure by the broker.
It therefore requires explicit synchronization in application code (Channel#basicPublish must be
invoked in a critical section). Sharing channels between threads will also interfere with Publisher
Confirms. Concurrent publishing on a shared channel is best avoided entirely, e.g. by using a
channel per thread.

It is possible to use channel pooling to avoid concurrent publishing on a shared channel: once a
thread is done working with a channel, it returns it to the pool, making the channel available for
another thread. Channel pooling can be thought of as a specific synchronization solution. It is
recommended that an existing pooling library is used instead of a homegrown solution. For
example, Spring AMQP which comes with a ready-to-use channel pooling feature.

Channels consume resources and in most cases applications very rarely need more than a few
hundreds open channels in the same JVM process. If we assume that the application has a thread
for each channel (as channel shouldn't be used concurrently), thousands of threads for a single
JVM is already a fair amount of overhead that likely can be avoided. Moreover a few fast
publishers can easily saturate a network interface and a broker node: publishing involves less
work than routing, storing and delivering messages.

A classic anti-pattern to be avoided is opening a channel for each published message. Channels
are supposed to be reasonably long-lived and opening a new one is a network round-trip which
makes this pattern extremely inefficient.

Consuming in one thread and publishing in another thread on a shared channel can be safe.

Server-pushed deliveries (see the section below) are dispatched concurrently with a guarantee
that per-channel ordering is preserved. The dispatch mechanism uses
a [Link] , one per connection. It is possible to provide a custom
executor that will be shared by all connections produced by a single ConnectionFactory using
the ConnectionFactory#setSharedExecutor setter.

When manual acknowledgements  are used, it is important to consider what thread does the
acknowledgement. If it's different from the thread that received the delivery
(e.g. Consumer#handleDelivery delegated delivery handling to a different thread),
acknowledging with the multiple parameter set to true is unsafe and will result in double-
acknowledgements, and therefore a channel-level protocol exception that closes the channel.
Acknowledging a single message at a time can be safe.

Receiving Messages by Subscription ("Push API")

import [Link];
import [Link];

The most efficient way to receive messages is to set up a subscription using

the Consumerinterface. The messages will then be delivered automatically as they arrive, rather
than having to be explicitly requested.

When calling the API methods relating to Consumers, individual subscriptions are always
referred to by their consumer tags. A consumer tag is a consumer identifier which can be either
client- or server-generated. To let RabbitMQ generate a node-wide unique tag, use
a Channel#basicConsume override that doesn't take a consumer tag argument or pass an empty
string for consumer tag and use the value returned by Channel#basicConsume. Consumer tags
are used to cancel consumers.

Distinct Consumer instances must have distinct consumer tags. Duplicate consumer tags on a

connection is strongly discouraged and can lead to issues with automatic connection recoveryand
confusing monitoring data when consumers are monitored.

The easiest way to implement a Consumer is to subclass the convenience class DefaultConsumer.

An object of this subclass can be passed on a basicConsume call to set up the subscription:
boolean autoAck = false;
[Link](queueName, autoAck, "myConsumerTag",
new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
[Link] properties,
byte[] body)
throws IOException
{
String routingKey = [Link]();
String contentType = [Link]();
long deliveryTag = [Link]();
// (process the message components here ...)
[Link](deliveryTag, false);
}
});

Here, since we specified autoAck = false, it is necessary to acknowledge messages delivered to

the Consumer, most conveniently done in the handleDelivery method, as illustrated.

More sophisticated Consumers will need to override further methods. In

particular, handleShutdownSignal is called when channels and connections close,
and handleConsumeOk is passed the consumer tag before any other callbacks to
that Consumer are called.
Consumers can also implement the handleCancelOk and handleCancel methods to be notified of
explicit and implicit cancellations, respectively.

You can explicitly cancel a particular Consumer with [Link]:

[Link](consumerTag);

passing the consumer tag.

Just like with publishers, it is important to consider concurrency hazard safety for consumers.

Callbacks to Consumers are dispatched in a thread pool separate from the thread that
instantiated its Channel. This means that Consumers can safely call blocking methods on
the Connection or Channel, such as Channel#queueDeclare or Channel#basicCancel.

Each Channel has its own dispatch thread. For the most common use case of
one Consumer per Channel, this means Consumers do not hold up other Consumers. If you have
multiple Consumers per Channel be aware that a long-running Consumer may hold up dispatch
of callbacks to other Consumers on that Channel.

Please refer to the Concurrency Considerations (Thread Safety) section for other topics related to
concurrency and concurrency hazard safety.
Retrieving Individual Messages ("Pull API")

To explicitly retrieve messages, use [Link]. The returned value is an instance

of GetResponse, from which the header information (properties) and message body can be
extracted:

boolean autoAck = false;

GetResponse response = [Link](queueName, autoAck);
if (response == null) {
// No message retrieved.
} else {
[Link] props = [Link]();
byte[] body = [Link]();
long deliveryTag = [Link]().getDeliveryTag();
...

and since the autoAck = false above, you must also call [Link] to acknowledge that

you have successfully received the message:

...
[Link]([Link], false); // acknowledge receipt of the message
}
Handling unroutable messages

If a message is published with the "mandatory" flags set, but cannot be routed, the broker will
return it to the sending client (via a [Link] command).

To be notified of such returns, clients can implement the ReturnListener interface and

call [Link]. If the client has not configured a return listener for a particular
channel, then the associated returned messages will be silently dropped.

[Link](new ReturnListener() {
public void handleReturn(int replyCode,
String replyText,
String exchange,
String routingKey,
[Link] properties,
byte[] body)
throws IOException {
...
}
});

A return listener will be called, for example, if the client publishes a message with the
"mandatory" flag set to an exchange of "direct" type which is not bound to a queue.
 Shutdown Protocol
Overview of the Client Shutdown Process

The AMQP 0-9-1 connection and channel share the same general approach to managing network
failure, internal failure, and explicit local shutdown.

The AMQP 0-9-1 connection and channel have the following lifecycle states:

 open: the object is ready to use

 closing: the object has been explicitly notified to shut down locally, has issued a shutdown
request to any supporting lower-layer objects, and is waiting for their shutdown procedures to
complete
 closed: the object has received all shutdown-complete notification(s) from any lower-layer
objects, and as a consequence has shut itself down

Those objects always end up in the closed state, regardless of the reason that caused the closure,
like an application request, an internal client library failure, a remote network request or network
failure.

The AMQP connection and channel objects possess the following shutdown-related methods:

 addShutdownListener(ShutdownListener
listener) and removeShutdownListener(ShutdownListener listener), to manage any listeners,
which will be fired when the object transitions to closedstate. Note that, adding a
ShutdownListener to an object that is already closed will fire the listener immediately
 getCloseReason(), to allow the investigation of what was the reason of the object’s shutdown
 isOpen(), useful for testing whether the object is in an open state
 close(int closeCode, String closeMessage), to explicitly notify the object to shut down

Simple usage of listeners would look like:

import [Link];
import [Link];

[Link](new ShutdownListener() {
public void shutdownCompleted(ShutdownSignalException cause)
{
...
}
});
Information about the circumstances of a shutdown

One can retrieve the ShutdownSignalException, which contains all the information available
about the close reason, either by explicitly calling the getCloseReason() method or by using
the causeparameter in the service(ShutdownSignalException cause) method of
the ShutdownListener class.

The ShutdownSignalException class provides methods to analyze the reason of the shutdown. By

calling the isHardError() method we get information whether it was a connection or a channel
error, and getReason() returns information about the cause, in the form an AMQP method -
either [Link] or [Link] (or null if the cause was some exception
in the library, such as a network communication failure, in which case that exception can be
retrieved with getCause()).

public void shutdownCompleted(ShutdownSignalException cause)

{
if ([Link]())
{
Connection conn = (Connection)[Link]();
if (![Link]())
{
Method reason = [Link]();
...
}
...
} else {
Channel ch = (Channel)[Link]();
...
}
}
Atomicity and use of the isOpen() method

Use of the isOpen() method of channel and connection objects is not recommended for

production code, because the value returned by the method is dependent on the existence of the
shutdown cause. The following code illustrates the possibility of race conditions:

public void brokenMethod(Channel channel)

{
if ([Link]())
{
// The following code depends on the channel being in open state.
// However there is a possibility of the change in the channel state
// between isOpen() and basicQos(1) call
...
[Link](1);
}
}

Instead, we should normally ignore such checking, and simply attempt the action desired. If
during the execution of the code the channel of the connection is closed,
a ShutdownSignalException will be thrown indicating that the object is in an invalid state. We
should also catch for IOException caused either by SocketException, when broker closes the
connection unexpectedly, or ShutdownSignalException, when broker initiated clean close.

public void validMethod(Channel channel)

{
try {
...
[Link](1);
} catch (ShutdownSignalException sse) {
// possibly check if channel was closed
// by the time we started action and reasons for
// closing it
...
} catch (IOException ioe) {
// check why connection was closed
...
}
}

Advanced Connection options

Consumer thread pool

Consumer threads (see Receiving below) are automatically allocated in a

new ExecutorServicethread pool by default. If greater control is required supply
an ExecutorService on the newConnection() method, so that this pool of threads is used instead.
Here is an example where a larger thread pool is supplied than is normally allocated:

ExecutorService es = [Link](20);
 Connection conn = [Link](es);
Both Executors and ExecutorService classes are in the [Link] package.

When the connection is closed a default ExecutorService will be shutdown(), but a user-

suppliedExecutorService (like es above) will not be shutdown(). Clients that supply a
custom ExecutorService must ensure it is shutdown eventually (by calling
its shutdown() method), or else the pool’s threads may prevent JVM termination.

The same executor service may be shared between multiple connections, or serially re-used on re-
connection but it cannot be used after it is shutdown().

4. Java Client API Guide

This guide covers RabbitMQ Java client and its public API. It assumes that the most recent major
version of the client is used and the reader is familiar with the basics. Key sections of the guide
are:

 Connecting to RabbitMQ
 Connection and Channel Lifespan
 Using Exchanges and Queues
 Publishing Messages
 Consuming Using a Subscription
 Concurrency Considerations and Safety
 Automatic Recovery From Network Failures

5.x release series of this library require JDK 8, both for compilation and at runtime. On Android,
this means only Android 7.0 or later versions are supported. 4.x release series support JDK 6 and
Android versions prior to 7.0.

The library is open source, developed on GitHub, and is triple-licensed under

 Apache Public License 2.0

 Mozilla Public License
 GPL 2.0
This means that the user can consider the library to be licensed under any of the licenses from the
list above. For example, the user may choose the Apache Public License 2.0 and include this
client into a commercial product. Codebases that are licensed under the GPLv2 may choose
GPLv2, and so on.

There are also command line tools that used to be shipped with the Java client.

The client API is closely modelled on the AMQP 0-9-1 protocol model, with additional
abstractions for ease of use.

An API reference (JavaDoc) is available separately.

 Overview

RabbitMQ Java client uses [Link] as its top-level package. The key classes and
interfaces are:

 Channel: represents an AMQP 0-9-1 channel, and provides most of the operations (protocol
methods).
 Connection: represents an AMQP 0-9-1 connection
 ConnectionFactory: constructs Connection instances
 Consumer: represents a message consumer
 DefaultConsumer: commonly used base class for consumers
 BasicProperties: message properties (metadata)
 [Link]: builder for BasicProperties
Protocol operations are available through the Channel interface. Connection is used to open
channels, register connection lifecycle event handlers, and close connections that are no longer
needed. Connections are instantiated through ConnectionFactory, which is how you configure
various connection settings, such as the vhost or username.

Connections and Channels

The core API classes are Connection and Channel, representing an AMQP 0-9-1 connection and
channel, respectively. They are typically imported before used:

import [Link];
import [Link];

Connecting to RabbitMQ

The following code connects to a RabbitMQ node using the given parameters (host name, port
number, etc):

ConnectionFactory factory = new ConnectionFactory();

// "guest"/"guest" by default, limited to localhost connections
[Link](userName);
[Link](password);
[Link](virtualHost);
[Link](hostName);
[Link](portNumber);

Connection conn = [Link]();

All of these parameters have sensible defaults for a RabbitMQ node running locally. The default
value for a property will be used if the property remains unassigned prior to creating a
connection:

Property Default Value

Username "guest"
Property Default Value

Password "guest"

Virtual "/"
host

Hostname "localhost"

port 5672 for regular connections, 5671 for connections that use

TLS

Alternatively, URIs may be used:

ConnectionFactory factory = new ConnectionFactory();

[Link]("amqp://userName:password@hostName:portNumber/virtualHost");
Connection conn = [Link]();

All of these parameters have sensible defaults for a stock RabbitMQ server running locally.

Note that user guest can only connect from localhost  by default. This is to limit well-known
credential use in production systems.

The Connection interface can then be used to open a channel:

Channel channel = [Link]();

The channel can now be used to send and receive messages, as described in subsequent sections.

Successful and unsuccessful client connection events can be observed in server node logs.

Disconnecting from RabbitMQ

To disconnect, simply close the channel and the connection:

[Link]();
[Link]();

Note that closing the channel may be considered good practice, but isn’t strictly necessary here -
it will be done automatically anyway when the underlying connection is closed.

Client disconnection events can be observed in server node logs .

Connection and Channel Lifespan

Connections are meant to be long-lived. The underlying protocol is designed and optimized for
long running connections. That means that opening a new connection per operation, e.g. a
message published, is unnecessary and strongly discouraged as it will introduce a lot of network
roundtrips and overhead.
Channels are also meant to be long-lived but since many recoverable protocol errors will result in
channel closure, channel lifespan could be shorter than that of its connection. Closing and
opening new channels per operation is usually unnecessary but can be appropriate. When in
doubt, consider reusing channels fist.

Channel-level exceptions such as attempts to consume from a queue that does not exist will result
in channel closure. A closed channel can no longer be used and will not receive any more events
from the server (such as message deliveries). Channel-level exceptions will be logged by
RabbitMQ and will initiate a shutdown sequence for the channel (see below).

Using Exchanges and Queues

Client applications work with exchanges and queues, the high-level building blocks of the
protocol. These must be declared before they can be used. Declaring either type of object simply
ensures that one of that name exists, creating it if necessary.

Continuing the previous example, the following code declares an exchange and a server-named
queue, then binds them together.

[Link](exchangeName, "direct", true);

String queueName = [Link]().getQueue();
[Link](queueName, exchangeName, routingKey);

This will actively declare the following objects, both of which can be customised by using
additional parameters. Here neither of them have any special arguments.
 1. a durable, non-autodelete exchange of "direct" type
2. a non-durable, exclusive, autodelete queue with a generated name

The above function calls then bind the queue to the exchange with the given routing key.

Note that this would be a typical way to declare a queue when only one client wants to work with
it: it doesn’t need a well-known name, no other client can use it (exclusive) and will be cleaned up
automatically (autodelete). If several clients want to share a queue with a well-known name, this
code would be appropriate:

[Link](exchangeName, "direct", true);

[Link](queueName, true, false, false, null);
[Link](queueName, exchangeName, routingKey);

This will actively declare:

1. a durable, non-autodelete exchange of "direct" type

2. a durable, non-exclusive, non-autodelete queue with a well-known name

Many Channel API methods are overloaded. These convenient short forms

of exchangeDeclare, queueDeclare and queueBind use sensible defaults. There are also longer
forms with more parameters, to let you override these defaults as necessary, giving full control
where needed.

This "short form, long form" pattern is used throughout the client API uses.
Passive Declaration

Queues and exchanges can be declared "passively". A passive declare simply checks that the
entity with the provided name exists. If it does, the operation is a no-op. For queues successful
passive declares will return the same information as non-passive ones, namely the number of
consumers and messages in ready state in the queue. If the entity does not exist, the operation
fails with a channel level exception. The channel cannot be used after that. A new channel should
be opened. It is common to use one-off (temporary) channels for passive declarations.

Channel#queueDeclarePassive and Channel#exchangeDeclarePassive are the methods used for

passive declaration. The following example demonstrates Channel#queueDeclarePassive:

[Link] response = [Link]("queue-name");

// returns the number of messages in Ready state in the queue
[Link]();
// returns the number of consumers the queue has
[Link]();
Channel#exchangeDeclarePassive's return value contains no useful information. Therefore if the
method returns and no channel exceptions occurs, it means that the exchange does exist.
Operations with Optional Responses

Some common operations also have a "no wait" version which won't wait for server response. For
example, to declare a queue and instruct the server to not send any response, use
[Link](queueName, true, false, false, null);
The "no wait" versions are more efficient but offer lower safety guarantees, e.g. they are more
dependent on the heartbeat mechanism for detection of failed operations. When in doubt, start
with the standard version. The "no wait" versions are only needed in scenarios with high topology
(queue, binding) churn.
Deleting Entities and Purging Messages

A queue or exchange can be explicitly deleted:

[Link]("queue-name")
It is possible to delete a queue only if it is empty:
[Link]("queue-name", false, true)
or if it is not used (does not have any consumers):
[Link]("queue-name", true, false)

A queue can be purged (all of its messages deleted):

[Link]("queue-name")

Publishing Messages

To publish a message to an exchange, use [Link] as follows:

byte[] messageBodyBytes = "Hello, world!".getBytes();
[Link](exchangeName, routingKey, null, messageBodyBytes);

For fine control, you can use overloaded variants to specify the mandatory flag, or send messages
with pre-set message properties:

[Link](exchangeName, routingKey, mandatory,

MessageProperties.PERSISTENT_TEXT_PLAIN,
messageBodyBytes);

This sends a message with delivery mode 2 (persistent), priority 1 and content-type "text/plain".
You can build your own message properties object, using a Builder class mentioning as many
properties as you like, for example:

[Link](exchangeName, routingKey,
new [Link]()
.contentType("text/plain")
.deliveryMode(2)
.priority(1)
.userId("bob")
.build()),
messageBodyBytes);

This example publishes a message with custom headers:

Map<String, Object> headers = new HashMap<String, Object>();

[Link]("latitude", 51.5252949);
[Link]("longitude", -0.0905493);

[Link](exchangeName, routingKey,
new [Link]()
.headers(headers)
.build()),
messageBodyBytes);

This example publishes a message with expiration:

[Link](exchangeName, routingKey,
new [Link]()
.expiration("60000")
.build()),
messageBodyBytes);

We have not illustrated all the possibilities here.

Note that BasicProperties is an inner class of the autogenerated holder class AMQP.

Invocations of Channel#basicPublish will eventually block if a resource-driven alarm is in effect.

Channels and Concurrency Considerations (Thread Safety)

As a rule of thumb, sharing Channel instances between threads is something to be avoided.

Applications should prefer using a Channel per thread instead of sharing the same Channelacross
multiple threads.

While some operations on channels are safe to invoke concurrently, some are not and will result
in incorrect frame interleaving on the wire, double acknowledgements and so on.

Concurrent publishing on a shared channel can result in incorrect frame interleaving on the wire,
triggering a connection-level protocol exception and immediate connection closure by the broker.
It therefore requires explicit synchronization in application code (Channel#basicPublish must be
invoked in a critical section). Sharing channels between threads will also interfere with Publisher
Confirms. Concurrent publishing on a shared channel is best avoided entirely, e.g. by using a
channel per thread.

It is possible to use channel pooling to avoid concurrent publishing on a shared channel: once a
thread is done working with a channel, it returns it to the pool, making the channel available for
another thread. Channel pooling can be thought of as a specific synchronization solution. It is
recommended that an existing pooling library is used instead of a homegrown solution. For
example, Spring AMQP which comes with a ready-to-use channel pooling feature.

Channels consume resources and in most cases applications very rarely need more than a few
hundreds open channels in the same JVM process. If we assume that the application has a thread
for each channel (as channel shouldn't be used concurrently), thousands of threads for a single
JVM is already a fair amount of overhead that likely can be avoided. Moreover a few fast
publishers can easily saturate a network interface and a broker node: publishing involves less
work than routing, storing and delivering messages.

A classic anti-pattern to be avoided is opening a channel for each published message. Channels
are supposed to be reasonably long-lived and opening a new one is a network round-trip which
makes this pattern extremely inefficient.

Consuming in one thread and publishing in another thread on a shared channel can be safe.

Server-pushed deliveries (see the section below) are dispatched concurrently with a guarantee
that per-channel ordering is preserved. The dispatch mechanism uses
a [Link] , one per connection. It is possible to provide a custom
executor that will be shared by all connections produced by a single ConnectionFactory using
the ConnectionFactory#setSharedExecutor setter.

When manual acknowledgements  are used, it is important to consider what thread does the
acknowledgement. If it's different from the thread that received the delivery
(e.g. Consumer#handleDelivery delegated delivery handling to a different thread),
acknowledging with the multiple parameter set to true is unsafe and will result in double-
acknowledgements, and therefore a channel-level protocol exception that closes the channel.
Acknowledging a single message at a time can be safe.

Receiving Messages by Subscription ("Push API")

import [Link];
import [Link];

The most efficient way to receive messages is to set up a subscription using

the Consumerinterface. The messages will then be delivered automatically as they arrive, rather
than having to be explicitly requested.

When calling the API methods relating to Consumers, individual subscriptions are always
referred to by their consumer tags. A consumer tag is a consumer identifier which can be either
client- or server-generated. To let RabbitMQ generate a node-wide unique tag, use
a Channel#basicConsume override that doesn't take a consumer tag argument or pass an empty
string for consumer tag and use the value returned by Channel#basicConsume. Consumer tags
are used to cancel consumers.

Distinct Consumer instances must have distinct consumer tags. Duplicate consumer tags on a

connection is strongly discouraged and can lead to issues with automatic connection recoveryand
confusing monitoring data when consumers are monitored.

The easiest way to implement a Consumer is to subclass the convenience class DefaultConsumer.

An object of this subclass can be passed on a basicConsume call to set up the subscription:

boolean autoAck = false;

[Link](queueName, autoAck, "myConsumerTag",
new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
 Envelope envelope,
[Link] properties,
byte[] body)
throws IOException
{
String routingKey = [Link]();
String contentType = [Link]();
long deliveryTag = [Link]();
// (process the message components here ...)
[Link](deliveryTag, false);
}
});

Here, since we specified autoAck = false, it is necessary to acknowledge messages delivered to

the Consumer, most conveniently done in the handleDelivery method, as illustrated.

More sophisticated Consumers will need to override further methods. In

particular, handleShutdownSignal is called when channels and connections close,
and handleConsumeOk is passed the consumer tag before any other callbacks to
that Consumer are called.

Consumers can also implement the handleCancelOk and handleCancel methods to be notified of

explicit and implicit cancellations, respectively.

You can explicitly cancel a particular Consumer with [Link]:

[Link](consumerTag);

passing the consumer tag.

Just like with publishers, it is important to consider concurrency hazard safety for consumers.

Callbacks to Consumers are dispatched in a thread pool separate from the thread that
instantiated its Channel. This means that Consumers can safely call blocking methods on
the Connection or Channel, such as Channel#queueDeclare or Channel#basicCancel.

Each Channel has its own dispatch thread. For the most common use case of
one Consumer per Channel, this means Consumers do not hold up other Consumers. If you have
multiple Consumers per Channel be aware that a long-running Consumer may hold up dispatch
of callbacks to other Consumers on that Channel.

Please refer to the Concurrency Considerations (Thread Safety) section for other topics related to
concurrency and concurrency hazard safety.

Retrieving Individual Messages ("Pull API")

To explicitly retrieve messages, use [Link]. The returned value is an instance

of GetResponse, from which the header information (properties) and message body can be
extracted:

boolean autoAck = false;

GetResponse response = [Link](queueName, autoAck);
if (response == null) {
// No message retrieved.
} else {
[Link] props = [Link]();
byte[] body = [Link]();
long deliveryTag = [Link]().getDeliveryTag();
...

and since the autoAck = false above, you must also call [Link] to acknowledge that

you have successfully received the message:

...
[Link]([Link], false); // acknowledge receipt of the message
}

Handling unroutable messages

If a message is published with the "mandatory" flags set, but cannot be routed, the broker will
return it to the sending client (via a [Link] command).

To be notified of such returns, clients can implement the ReturnListener interface and

call [Link]. If the client has not configured a return listener for a particular
channel, then the associated returned messages will be silently dropped.

[Link](new ReturnListener() {
public void handleReturn(int replyCode,
 String replyText,
String exchange,
String routingKey,
[Link] properties,
byte[] body)
throws IOException {
...
}
});

A return listener will be called, for example, if the client publishes a message with the
"mandatory" flag set to an exchange of "direct" type which is not bound to a queue.

Shutdown Protocol
Overview of the Client Shutdown Process

The AMQP 0-9-1 connection and channel share the same general approach to managing network
failure, internal failure, and explicit local shutdown.

The AMQP 0-9-1 connection and channel have the following lifecycle states:

 open: the object is ready to use

 closing: the object has been explicitly notified to shut down locally, has issued a shutdown
request to any supporting lower-layer objects, and is waiting for their shutdown procedures to
complete
 closed: the object has received all shutdown-complete notification(s) from any lower-layer
objects, and as a consequence has shut itself down

Those objects always end up in the closed state, regardless of the reason that caused the closure,
like an application request, an internal client library failure, a remote network request or network
failure.

The AMQP connection and channel objects possess the following shutdown-related methods:

 addShutdownListener(ShutdownListener
listener) and removeShutdownListener(ShutdownListener listener), to manage any listeners,
which will be fired when the object transitions to closedstate. Note that, adding a
ShutdownListener to an object that is already closed will fire the listener immediately
 getCloseReason(), to allow the investigation of what was the reason of the object’s shutdown
 isOpen(), useful for testing whether the object is in an open state
 close(int closeCode, String closeMessage), to explicitly notify the object to shut down

Simple usage of listeners would look like:

import [Link];
import [Link];

[Link](new ShutdownListener() {
public void shutdownCompleted(ShutdownSignalException cause)
{
 ...
}
});
Information about the circumstances of a shutdown

One can retrieve the ShutdownSignalException, which contains all the information available
about the close reason, either by explicitly calling the getCloseReason() method or by using
the causeparameter in the service(ShutdownSignalException cause) method of
the ShutdownListener class.

The ShutdownSignalException class provides methods to analyze the reason of the shutdown. By

calling the isHardError() method we get information whether it was a connection or a channel
error, and getReason() returns information about the cause, in the form an AMQP method -
either [Link] or [Link] (or null if the cause was some exception
in the library, such as a network communication failure, in which case that exception can be
retrieved with getCause()).

public void shutdownCompleted(ShutdownSignalException cause)

{
if ([Link]())
{
Connection conn = (Connection)[Link]();
if (![Link]())
{
 Method reason = [Link]();
...
}
...
} else {
Channel ch = (Channel)[Link]();
...
}
}
Atomicity and use of the isOpen() method

Use of the isOpen() method of channel and connection objects is not recommended for

production code, because the value returned by the method is dependent on the existence of the
shutdown cause. The following code illustrates the possibility of race conditions:

public void brokenMethod(Channel channel)

{
if ([Link]())
{
// The following code depends on the channel being in open state.
// However there is a possibility of the change in the channel state
// between isOpen() and basicQos(1) call
...
[Link](1);
 }
}

Instead, we should normally ignore such checking, and simply attempt the action desired. If
during the execution of the code the channel of the connection is closed,
a ShutdownSignalException will be thrown indicating that the object is in an invalid state. We
should also catch for IOException caused either by SocketException, when broker closes the
connection unexpectedly, or ShutdownSignalException, when broker initiated clean close.

public void validMethod(Channel channel)

{
try {
...
[Link](1);
} catch (ShutdownSignalException sse) {
// possibly check if channel was closed
// by the time we started action and reasons for
// closing it
...
} catch (IOException ioe) {
// check why connection was closed
...
}
}
Advanced Connection options
Consumer thread pool

Consumer threads (see Receiving below) are automatically allocated in a

new ExecutorServicethread pool by default. If greater control is required supply
an ExecutorService on the newConnection() method, so that this pool of threads is used instead.
Here is an example where a larger thread pool is supplied than is normally allocated:

ExecutorService es = [Link](20);
Connection conn = [Link](es);
Both Executors and ExecutorService classes are in the [Link] package.

When the connection is closed a default ExecutorService will be shutdown(), but a user-

suppliedExecutorService (like es above) will not be shutdown(). Clients that supply a
custom ExecutorService must ensure it is shutdown eventually (by calling
its shutdown() method), or else the pool’s threads may prevent JVM termination.

The same executor service may be shared between multiple connections, or serially re-used on re-
connection but it cannot be used after it is shutdown().

Use of this feature should only be considered if there is evidence that there is a severe bottleneck
in the processing of Consumer callbacks. If there are no Consumer callbacks executed, or very
few, the default allocation is more than sufficient. The overhead is initially minimal and the total
thread resources allocated are bounded, even if a burst of consumer activity may occasionally
occur.

Using Lists of Hosts

It is possible to pass an Address array to newConnection(). An Address is simply a convenience

class in the [Link] package with host and port components. For example:

Address[] addrArr = new Address[]{ new Address(hostname1, portnumber1)

, new Address(hostname2, portnumber2)};
Connection conn = [Link](addrArr);
will attempt to connect to hostname1:portnumber1, and if that fails to hostname2:portnumber2.
The connection returned is the first in the array that succeeds (without throwing IOException).
This is entirely equivalent to repeatedly setting host and port on a factory,
calling [Link]() each time, until one of them succeeds.

If an ExecutorService is provided as well (using the form [Link](es, addrArr))

the thread pool is associated with the (first) successful connection.
5. Security

AMQP commands permission checks

Exchange - passive=false
bit passive
If set, the server will reply with Declare-Ok if the queue already exists with the same name, and
raise an error if not. The client can use this to check whether a queue exists without modifying
the server state. When set, all other method fields except name and no-wait are ignored. A
declare with both passive and no-wait has no effect. Arguments are compared for semantic
equivalence.
The client MAY ask the server to assert that a queue exists without creating the queue if not. If
the queue does not exist, the server treats this as a failure. Error code: not-found
If not set and the queue exists, the server MUST check that the existing queue has the same
values for durable, exclusive, auto-delete, and arguments fields. The server MUST respond with
Declare-Ok if the requested queue matches these fields, and MUST raise a channel exception if
not.
Alternate Exchange (“AE”)

It is sometimes desirable to let clients handle messages that an exchange was unable to route (i.e.
either because there were no bound queues our no matching bindings). Typical examples of this
are

 detecting when clients accidentally or maliciously publish messages that cannot be

routed
 “or else” routing semantics where some messages are handled specially and the rest by a
generic handler

RabbitMQ’s Alternate Exchange (“AE”) feature addresses these use cases.

For any given exchange, an AE can be defined by clients using the exchange’s arguments, or in the
server using policies. In the case where both policy and arguments specify an AE, the one specified
in arguments overrules the one specified in policy.

Dead Letter Exchanges

Messages from a queue can be ‘dead-lettered’; that is, republished to another exchange when any
of the following events occur:

 The message is rejected ([Link] or [Link]) with requeue=false,

 The TTL for the message expires; or
 The queue length limit is exceeded.
Dead letter exchanges (DLXs) are normal exchanges. They can be any of the usual types and are
declared as usual.
SSL/TLS Handshake

A typical TCP connection is established through a process known as a three-way handshake. In it, a

client sends out a connection request (SYN), receives an acknowledgment (SYN/ACK) and then
responds with an acknowledgment of its own (ACK). This closes the loop and establishes the
connection.

One, two and three…forth, back and forth. Assuming there are no hiccups, the time it takes to
complete the handshake should be exactly equal to a single round trip time (RTT).

On the other hand, negotiating a SSL/TLS connection requires a few additional back-and-forths.
This is because the browser and server now also need to:

1. Agree upon a mutually-compatible method of encryption.

2. Go through a process of mutual verification.

3. Generate symmetric keys, used to encode and decode all information exchanged during the
session.

These extra interactions add overhead to the process, resulting in two additional round trips—or
more, depending on your server’s configuration.
Keys, Certificates and CA Certificates

The root certificate is a self-signed certificate, made available by a Certificate Authority. These

exist as commercial companies, and will, for a fee, sign SSL Certificates that you have generated.

For the purposes of this guide, we will start by creating our own Certificate Authority. Once we
have done this, we will generate signed certificates for the server and clients, in a number of
formats. These will we then use with the Java, .Net and Erlang AMQP clients. Note that Mono has
more stringent requirements on OpenSSL certificates (and a few bugs too), so we will be specifying
slightly more stringent key usage constraints than is normally necessary.

# mkdir testca
# cd testca
# mkdir certs private
# chmod 700 private
# echo 01 > serial
# touch [Link]

Now place the following in [Link] within the testca directory we've just created:

[ ca ]
default_ca = testca

[ testca ]
dir = .
certificate = $dir/[Link]
database = $dir/[Link]
new_certs_dir = $dir/certs
private_key = $dir/private/[Link]
serial = $dir/serial

default_crl_days = 7
default_days = 365
default_md = sha1

policy = testca_policy
x509_extensions = certificate_extensions

[ testca_policy ]
commonName = supplied
stateOrProvinceName = optional
countryName = optional
emailAddress = optional
organizationName = optional
organizationalUnitName = optional

[ certificate_extensions ]
basicConstraints = CA:false
[ req ]
default_bits = 2048
default_keyfile = ./private/[Link]
default_md = sha1
prompt = yes
distinguished_name = root_ca_distinguished_name
x509_extensions = root_ca_extensions

[ root_ca_distinguished_name ]
commonName = hostname

[ root_ca_extensions ]
basicConstraints = CA:true
keyUsage = keyCertSign, cRLSign

[ client_ca_extensions ]
basicConstraints = CA:false
keyUsage = digitalSignature
extendedKeyUsage = [Link].[Link].2

[ server_ca_extensions ]
basicConstraints = CA:false
keyUsage = keyEncipherment
extendedKeyUsage = [Link].[Link].1

Now we can generate the key and certificates that our test Certificate Authority will use. Still
within the testca directory:

# openssl req -x509 -config [Link] -newkey rsa:2048 -days 365 \

-out [Link] -outform PEM -subj /CN=MyTestCA/ -nodes
# openssl x509 -in [Link] -out [Link] -outform DER

This is all that is needed to generate our test Certificate Authority. The root certificate is

in testca/[Link] and is also in testca/[Link]. These two files contain the same information,
but in different formats. Whilst the vast majority of the world is perfectly happy to use
the PEM format, Microsoft and Mono like to be different, and so use the DER format.

Having set up our Certificate Authority, we now need to generate keys and certificates for the
clients and the server. The Erlang client and the RabbitMQ broker are both able to use PEM files
directly. They will both be informed of three files: the root certificate, which is implicitly trusted,
the private key, which is used to prove ownership of the public certificate being presented, and the
public certificate itself, which identifies the peer.

For convenience, we provide to the Java and .Net clients, a PKCS #12 store, which contains both
the client's certificate and key. The PKCS store is usually password protected itself, and so that
password must also be provided.
The process for creating server and client certificates is very similar. The only difference is
the keyUsage field that is added when signing the certificate. First the server:

# cd ..
# ls
testca
# mkdir server
# cd server
# openssl genrsa -out [Link] 2048
# openssl req -new -key [Link] -out [Link] -outform PEM \
-subj /CN=$(hostname)/O=server/ -nodes
# cd ../testca
# openssl ca -config [Link] -in ../server/[Link] -out \
../server/[Link] -notext -batch -extensions server_ca_extensions
# cd ../server
# openssl pkcs12 -export -out keycert.p12 -in [Link] -inkey [Link] -passout
pass:MySecretPassword

And now the client:

# cd ..
# ls
server testca
# mkdir client
# cd client
# openssl genrsa -out [Link] 2048
# openssl req -new -key [Link] -out [Link] -outform PEM \
-subj /CN=$(hostname)/O=client/ -nodes
# cd ../testca
# openssl ca -config [Link] -in ../client/[Link] -out \
../client/[Link] -notext -batch -extensions client_ca_extensions
# cd ../client
# openssl pkcs12 -export -out keycert.p12 -in [Link] -inkey [Link] -passout
pass:MySecretPassword

Enabling SSL Support in RabbitMQ

To enable the SSL/TLS support in RabbitMQ, we need to provide to RabbitMQ the location of
the root certificate, the server's certificate file, and the server's key. We also need to tell it to listen
on a socket that is going to be used for SSL connections, and we need to tell it whether it should
ask for clients to present certificates, and if the client does present a certificate, whether we should
accept the certificate if we can't establish a chain of trust to it. These settings are all controlled by
two arguments to RabbitMQ:

 -rabbit ssl_listeners

This is a list of ports to listen on for SSL connections. To listen on a single network interface, add
something like {"[Link]", 5671} to the list.
 -rabbit ssl_options

This is a tuple list of new_ssl options. The complete list of ssl_options is available via
the new_ssl man page: i.e. erl -man new_ssl, but the most important
are cacertfile, certfile and keyfile.

The simplest way to set these options, is to edit the configuration file. An example of the config file
is below, which will start one ssl_listener on port 5671 on all interfaces on this hostname:

[
{rabbit, [
{ssl_listeners, [5671]},
{ssl_options, [{cacertfile,"/path/to/testca/[Link]"},
{certfile,"/path/to/server/[Link]"},
{keyfile,"/path/to/server/[Link]"},
{verify,verify_peer},
{fail_if_no_peer_cert,false}]}
]}
].

Note to Windows users: Backslashes ("\") in the configuration file are interpreted as escape
sequences - so for example to specify the path c:\[Link] for the CA certificate you would need
to enter {cacertfile, "c:\\[Link]"} or {cacertfile, "c:/[Link]"}.
When a web browser connects to an HTTPS web server, the server presents its public certificate,
the web browser attempts to establish a chain of trust between the root certificates the browser is
aware of and the server's certificate, and all being well, an encrypted communication channel is
established. Although not used normally by web browsers and web servers, SSL allows the server
to ask the client to present a certificate. In this way the server can verify that the client is who they
say they are.

This policy of whether or not the server asks for a certificate from the client, and whether or not
they demand that they are able to trust the certificate, is what
the verify and fail_if_no_peer_cert arguments control. Through
the {fail_if_no_peer_cert,false} option, we state that we're prepared to accept clients which don't
have a certificate to send us, but through the {verify,verify_peer} option, we state that if the client
does send us a certificate, we must be able to establish a chain of trust to it. Note that these values
can change across versions of ssl shipped with Erlang/OTP, so check your man page erl -man
new_ssl to ensure you have the proper values.

Note that if {verify, verify_none} is used, no certificate exchange takes place from the client to the
server and rabbitmqctl list_connectionswill output empty strings for the peer certificate info items.

After starting the broker, you should then see the following in the [Link]:

=INFO REPORT==== 9-Aug-2010::15:10:55 ===

started TCP Listener on [Link]:5672

=INFO REPORT==== 9-Aug-2010::15:10:55 ===

started SSL Listener on [Link]:5671

Also, take note of the last line, which shows that RabbitMQ server is up and running and listening
for ssl connections.

Trust the Client's Root CA

Currently, we're telling RabbitMQ to look at the testca/[Link] file. This contains just the
public certificate of our test Certificate Authority. We may have certificates being presented by
clients which have been signed by several different Certificate Authorities, and we wish RabbitMQ
to trust all of them. Therefore, we can simply append these certificates to one another and provide
the path to this new file as the cacerts argument to RabbitMQ:

# cat testca/[Link] >> all_cacerts.pem

# cat otherca/[Link] >> all_cacerts.pem

and so forth.

Key Managers, Trust Managers and Stores

There are three components to be aware of in the Java security framework: Key Manager, Trust
Manager and Key Store.

A Key Manager is used by a peer to manage its certificates. This means that in a session set-up, the
Key Manager will control which certificates to send to the remote peer.
A Trust Manager is used by a peer to manage remote certificates. This means that in a session set-
up, the Trust Manager will control which certificates are trusted from a remote peer.

A Key Store is a Java encapsulation of certificates. Java needs all certificates to either be converted
into a Java specific binary format or to be in the PKCS#12 format. These formats are managed
using the Key Store class. For the server certificate, we'll use the Java binary format, but for client
key/certificate pair, we'll use the PKCS#12 format.

Connecting without validating certificates

Our first example will show a simple client, connecting to a RabbitMQ server over
SSL without validating the server certificate, and without presenting any client certificate.

import [Link].*;
import [Link].*;

import [Link].*;

public class Example1

{
public static void main(String[] args) throws Exception
{

ConnectionFactory factory = new ConnectionFactory();

[Link]("localhost");
[Link](5671);

[Link]();
// Tells the library to setup the default Key and Trust managers for you
// which do not do any form of remote server trust verification

Connection conn = [Link]();

Channel channel = [Link]();

//non-durable, exclusive, auto-delete queue

[Link]("rabbitmq-java-test", false, true, true, null);
[Link]("", "rabbitmq-java-test", null, "Hello, World".getBytes());

GetResponse chResponse = [Link]("rabbitmq-java-test", false);

if(chResponse == null) {
[Link]("No message retrieved");
} else {
byte[] body = [Link]();
[Link]("Recieved: " + new String(body));
}
 [Link]();
[Link]();
}
}

This simple example is an echo test. It creates a channel rabbitmq-java-test and publishes to the

default direct exchange, then reads back what has been published and echoes it out. Note that we
use an exclusive, non-durable, auto-delete queue so we don't have to worry about manually
cleaning up after ourselves.

When using instructions for creating a certificate signing request (csr), two files are created:
-- certificate signing request file with extension req
-- key file, containing public and private server keys, with extension pem