The Channel instances are created by the Connection object; therefore, the logical location to place the channel creation logic is in RabbitMqManager, as follows:

public Channel createChannel()
{
    try
    {
        return connection == null ? null : connection.createChannel();
    }
    catch (final Exception e)
    {
        LOGGER.log(Level.SEVERE, "Failed to create channel", e);
        return null;
    }
}

Again, this is quite simple; if anything goes awry when creating a channel, the method returns null. This is in line with what CCM desires, to shield the application from any RabbitMQ-related failures. Instead of dealing with exceptions coming from the messaging subsystem, it will just have to deal with potential null values. In the same spirit, the disposal of channels is delegated to a method that takes care of potential exceptions as follows:

public void closeChannel(final Channel channel)
{
    // isOpen is not fully trustable!
    if ((channel == null) || (!channel.isOpen()))
    {
        return;
    }

    try
    {
        channel.close();
    }
    catch (final Exception e)
    {
        LOGGER.log(Level.SEVERE, "Failed to close channel: " + channel, e);
    }
}

Note that the isOpen method can't be fully trusted; another thread may close the channel after this check is done. So, the call to the close method could still fail because the channel might have closed already.

Realizing that the "open channel, do something with the channel, close channel" scenario may occur regularly in the code, CCM decides to support it by creating some code artifacts. It first creates an interface that defines what the contract for this pattern should be, as follows:

public interface ChannelCallable<T>
{
    String getDescription();

    T call(Channel channel) throws IOException;
}

Then, it adds a method to RabbitMqManager in order to execute such a ChannelCallable instance, as follows:

public <T> T call(final ChannelCallable<T> callable)
{
    final Channel channel = createChannel();
 
    if (channel != null)
    {
        try
        {
            return callable.call(channel);
        }
        catch (final Exception e)
        {
            LOGGER.log(Level.SEVERE, "Failed to run: " + callable.getDescription() + " on channel: " + channel, e);
        }
        finally
        {
            closeChannel(channel);
        }
    }

    return null;
}

Again, the invoker of this call method will be shielded from any error that could stem from the messaging layer; it will just receive null if something goes wrong. Notice how the ChannelCallable description is used in the log message. The guiding principle here is that you should always provide as much contextual information as possible when something goes wrong.

CCM is quite happy with its core infrastructure code. It is now able to connect to a RabbitMQ broker, open a channel, and issue a series of commands, all in a thread-safe and exception-safe manner. It's now time to build on this foundation!

If you remember the discussion about AMQP in Chapter 1, A Rabbit Springs to Life, messages are published to exchanges from where they get routed to queues, ready to be consumed. A routing strategy determines which queue (or queues) the message will be routed to. The routing strategy bases its decision on a routing key (a free-form string) and potentially on message meta-information. In the case of the user-to-user messaging system considered here, one message needs to be routed to the queue acting as the inbox of the addressee. Therefore, the exchange-routing strategy that needs to be used is the direct one, which matches the destination queue name with the routing key used when the message is produced, as illustrated in the following figure:

The direct exchange route messages to specific queues

To tie the messaging logic in its application, CCM will piggyback an existing polling mechanism that's already in place between the JavaScript frontend and the Java backend. This is not the most efficient approach, and in fact it will be reviewed as you'll soon find out, but it's the easiest way for it to get started and roll out the feature in the best time frame. The following figure shows how the frontend poll will be used to fetch messages from the user's inbox and a regular AJAX call will be used to send a new message. Messages themselves will be represented as JSON objects (refer to Appendix, Message Schemas, for the formal specification of these JSON messages).

They will contain meta-information such as timestamp, sender, and receiver IDs on top of the text contents of the message itself, as shown in the following diagram:

The frontend/backend interactions of CCM's main application

Let's follow the code CCM has created to roll this feature out and learn about the different concepts at the same time. It's created a UserMessageManager class to encapsulate all the operations related to this particular feature:

public class UserMessageManager
{
    static final String USER_INBOXES_EXCHANGE = "user-inboxes";

    @Inject
    RabbitMqManager rabbitMqManager;

    public void onApplicationStart()
    {
        rabbitMqManager.call(new ChannelCallable<DeclareOk>()
        {
            @Override
            public String getDescription()
            {
                return "Declaring direct exchange: " + USER_INBOXES_EXCHANGE;
            }

            @Override
            public DeclareOk call(final Channel channel) throws IOException
            {
                String exchange = USER_INBOXES_EXCHANGE;
                String type = "direct";
                // survive a server restart
                boolean durable = true;
                // keep it even if nobody is using it
                boolean autoDelete = false;
                // no special arguments
                Map<String, Object> arguments = null;

                return channel.exchangeDeclare(exchange, type, durable, autoDelete, arguments);
            }
        });
    }

After receiving the RabbitMqManager instance via the dependency injection, it's created an onApplicationStart method that, as its name suggests, gets called every time the application server starts. All this method does is declare the exchange where the user-to-user messages are published to. Why do we do this on start? This is because it's a fundamental requirement of the user-to-user messaging subsystem; if the exchange doesn't exist, attempts to publish messages to it will raise exceptions.

Notice that the method to create the exchange is called declare not create—this is to suggest that if the exchange already exists, it will do nothing; otherwise, it will actually create it. This is why it's safe to declare this every time the application starts. Also, it would be an overkill to do it when every message is sent, so the application start is the best time to do it.

Besides using the direct type, we also configure the durable, autoDelete, and arguments properties of the exchange. We do not want this exchange to go away after a restart of RabbitMQ, nor when it's not being used anymore; hence, the values we've used.

After creating the exchange, the next thing we want to do is to have the user inbox queue created and bound to the exchange. The following is how we do it:

public void onUserLogin(final long userId)
{
    final String queue = getUserInboxQueue(userId);

    rabbitMqManager.call(new ChannelCallable<BindOk>()
    {
        @Override
        public String getDescription()
        {
            return "Declaring user queue: " + queue + ", binding it to exchange: "
                   + USER_INBOXES_EXCHANGE;
        }
 
        @Override
        public BindOk call(final Channel channel) throws IOException
        {
            return declareUserMessageQueue(queue, channel);
        }
    });
}

private BindOk declareUserMessageQueue(final String queue, final Channel channel) throws IOException
{
    // survive a server restart
    boolean durable = true;
    // keep the queue
    boolean autoDelete = false;
    // can be consumed by another connection
    boolean exclusive = false;
    // no special arguments
    Map<String, Object> arguments = null;
    channel.queueDeclare(queue, durable, exclusive, autoDelete, arguments);

    // bind the addressee's queue to the direct exchange
    String routingKey = queue;
    return channel.queueBind(queue, USER_INBOXES_EXCHANGE, routingKey);
}

Every time a user logs in on the system, the application calls onUserLogin. After getting the addressee's queue name from getUserInboxQueue (merely "user-inbox." + userId), it then calls declareUserMessageQueue (you'll soon understand why the method is split in half). In this method, the queue is declared with an approach that's really similar to how it's done for an exchange, but with slightly different properties, as follows:

Then the queue is bound to the exchange using its own name as the routing key so that the direct routing strategy can route messages to it. When this is done, publishing messages to the user-inboxes exchange will actually deliver messages to the user queue whose name matches the published routing key.

Again, when the same properties are used, these operations are idempotent, so we can safely declare the queue and bind it to the exchange again and again, each time a user logs in.

static final String MESSAGE_CONTENT_TYPE = "application/vnd.ccm.pmsg.v1+json";
static final String MESSAGE_ENCODING = "UTF-8";

public String sendUserMessage(final long userId, final String jsonMessage)
{
    return rabbitMqManager.call(new ChannelCallable<String>()
    {
        @Override
        public String getDescription()
        {
            return "Sending message to user: " + userId;
        }

        @Override
        public String call(final Channel channel) throws IOException
        {
            String queue = getUserInboxQueue(userId);

            // it may not exist so declare it
            declareUserMessageQueue(queue, channel);

            String messageId = UUID.randomUUID().toString();

            BasicProperties props = new BasicProperties.Builder()
                .contentType(MESSAGE_CONTENT_TYPE)
                .contentEncoding(MESSAGE_ENCODING)
                .messageId(messageId)
                .deliveryMode(2)
                .build();

            String routingKey = queue;

            // publish the message to the direct exchange
            channel.basicPublish(USER_INBOXES_EXCHANGE, routingKey, props,
                jsonMessage.getBytes(MESSAGE_ENCODING));

            return messageId;
        }
    });
}

AMQP message structure

The following figure illustrates the structure of an AMQP message where you will recognize the four properties we've just used and discover a few more. Note that this figure uses the specification name of the fields; each language implementation renames them slightly so they can be valid names, for example, content-type becomes contentType in Java.

Structure of an AMQP message

Except the reserved one, all these properties are free to use and, unless otherwise specified, are ignored by the AMQP broker. In the case of RabbitMQ, the only field that is supported by the broker is the user-id field, which is validated to ensure it matches the name of the broker user that established the connection to the broker. Notice how the headers property allows you to add extra key-value pairs in case none of the standard properties fit your needs.

public List<String> fetchUserMessages(final long userId)
{
    return rabbitMqManager.call(new ChannelCallable<List<String>>()
    {
        @Override
        public String getDescription()
        {
            return "Fetching messages for user: " + userId;
        }
        @Override
        public List<String> call(final Channel channel) 
          throws IOException
        {
            List<String> messages = new ArrayList<>();

            String queue = getUserInboxQueue(userId);
            boolean autoAck = true;

            GetResponse getResponse;

            while ((getResponse = channel.basicGet(queue, autoAck)) != null)
            {
                final String contentEncoding = getResponse.getProps().getContentEncoding();
                messages.add(new String(getResponse.getBody(), contentEncoding));
            }

            return messages;
        }
    });
}

With the application running on a pair of servers connected to RabbitMQ, you can see the following established connections from the management console:

The management console provides connection information

As expected, one connection per application server has been established. Notice how the upstream and downstream network throughputs are clearly represented. What about channels? Because they get opened and closed very quickly, it's actually hard to see any from the management console. With our current architecture, they just don't stay long enough to be rendered on the user interface of the console. So let's look at the following exchanges:

The user-inbox direct exchange shows up in the management console

You can see the user exchange and the rate of messages coming in and out of it. Their being consumed as fast as they come in is a good sign as it means the current architecture is sufficient for our needs and messages are not piling up. However, what are all these other exchanges that we can see here? Clearly, we haven't created them by code, so they should be coming from somewhere. Indeed, the nameless exchange represented as (AMQP default) and all the exchanges whose names start with amq. are defined by the AMQP specification and, as such, must be provided by default by RabbitMQ. Now, what about queues? Let's have a look at them:

Each user-to-user inbox queue is visible in the management console

As expected, you see one queue per user and some nifty usage statistics. Notice how the ack column is empty. This should be no surprise to you if you remember what we've said about message acknowledgement. You're receiving messages while letting RabbitMQ know we won't be acknowledging them; thus, there's no activity related to acknowledging messages!

Confident about its architecture and implementation, CCM rolls out the user-to-user messaging subsystem to production. It's an immediate success. Users actually want a new feature added to it: the capacity to send a message to a group of users. Let's see how we're going to implement this new feature with RabbitMQ.