Direct exchange on a custom RabbitMQ broker

The stress in this post is not on how to do a direct exchange with RabbitMQ, but on how to create a RabbitMQ producer-consumer application when the broker is not an off-the-shelf setup. In the previous post I have twisted a bit the broker rabbit configuration, setting a non-standard port, defining a virtual host and a user on it. Now I am going to write a trivial client-server RabbitMQ application for this setup.

This minimal application is contained in a single class, and it has a couple public methods, producer() and consumer(), that implement a very raw direct exchange where a single message is produced and consumed. They both call the private method getFactory() instead of creating a standard ConnectionFactory object. And this the interesting part of the application:

static private final String HOST = "127.0.0.1"; // 1
static private final int PORT = 6391; // 2
static private final String V_HOST = "test"; // 3
static private final String USER = "user"; // 4
static private final String PASSWORD = "password";

//...

private ConnectionFactory getFactory() {
    ConnectionFactory factory = new ConnectionFactory(); // 5
    factory.setHost(HOST); // 6
    factory.setPort(PORT);
    factory.setVirtualHost(V_HOST);
    factory.setUsername(USER);
    factory.setPassword(PASSWORD);

    System.out.println("Starting on " + factory.getHost() + ':' + factory.getPort() + " ["
            + factory.getVirtualHost() + "] for " + factory.getUsername());
    return factory;
}

1. This is a puny testing application running on the same machine of the broker, so localhost is used.
2. Default port for RabbitMQ is 5672, but here we are using a different one.
3. We don't use the default virtual host ("/") but our specific "test" one.
4. The default user is "guest" (with password "guest"), here we are using instead our user expressly created for our v-host.
5. A default connection factory works fine for a default setting of the RabbitMQ broker. For make it work for our customized broker, we have to explicitly set all the changed properties.
6. Actually, the host is localhost, as set by default in the ConnectionFactory ctor, but let's assume we'll change it soon, as it usually happens in real life.

Obviously all those properties should be be easily configured, stored on mass memory and fetched by the application at startup, or passed as argument from the command line, but this is not the point of this post.

The rest of the class code is not very interesting, but if you want to see the full Java class, you can get it from github.

Go to the full post

RabbitMQ broker setup

As we have seen, it is very easy start working with RabbitMQ. If we are happy with the default setting we can be up and running in a matter of minutes. It is still easy, but requires a bit of work to change the rabbit broker setup to use a different port and restrict the access to authorized users.

From the point of view of a broker client, there are a few things that could go wrong when trying to connect. As usual, these problems are converted in exceptions, that your client should make available to the user is some form. Here is the three of them I bumped in while doing some testing.

The broker is down:

java.net.ConnectException: Connection refused: connect
    at java.net.TwoStacksPlainSocketImpl.socketConnect(Native Method)
    at java.net.AbstractPlainSocketImpl.doConnect(Unknown Source)
    at java.net.AbstractPlainSocketImpl.connectToAddress(Unknown Source)
    at java.net.AbstractPlainSocketImpl.connect(Unknown Source)
    at java.net.PlainSocketImpl.connect(Unknown Source)
    at java.net.SocksSocketImpl.connect(Unknown Source)
    at java.net.Socket.connect(Unknown Source)
...

The user password is wrong:

com.rabbitmq.client.PossibleAuthenticationFailureException: Possibly caused by authentication failure
    at com.rabbitmq.client.impl.AMQConnection.start(AMQConnection.java:342)
    at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:516)
    at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:533)
...

The user is unknown:

java.io.IOException
    at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:106)
    at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:102)
    at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:124)
    at com.rabbitmq.client.impl.AMQConnection.start(AMQConnection.java:375)
...
Caused by: com.rabbitmq.client.ShutdownSignalException: connection error;
reason: java.io.EOFException
    at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:67)
...

There is a rabbit plugin designed to provide administrator capabilities, but before using it, we have to enable it through this command:

rabbitmq-plugins enable rabbitmq_management

On Windows you should get this feedback:

The following plugins have been enabled:
  mochiweb
  webmachine
  rabbitmq_mochiweb
  amqp_client
  rabbitmq_management_agent
  rabbitmq_management

Plugin configuration has changed. Restart RabbitMQ for changes to take effect.
If you have RabbitMQ running as a service then you must reinstall by running
  rabbitmq-service.bat stop
  rabbitmq-service.bat install
  rabbitmq-service.bat start

As you could expect, if you don't need it anymore, you can get rid of it:

rabbitmq-plugins disable rabbitmq_management

Once you have installed the plugin, and restarted the broker, you can access the rabbit administrator via your preferred browser, by an user with administration rights, by default guest with password guest:

http://localhost:15672/

[edit]
When I wrote the post, the default port was 55672, with version 3.0 it has be changed to 15672
[/edit]

We can create a virtual host, so to keep insulated all the message streams relative to a specific environment. Just go to the "Virtual Hosts" tab, click on "Add a new virtual host", specify a meaningful name (shame on me, I called it "test") and click on the "Add virtual host" button. So easy just that, and we have a new virtual host.

Still a virtual host with no user associated is not very useful and, as we see in the "All virtual hosts" section, a newly created virtual host has "No users" associated.

So we access the "Add / update a user" section in the "Users" tab, and create a new user. Insert a reasonable user-password couple (again, I didn't pay attention to my good suggestion, and I entered a lousy "user"-"password") and choose the priviledge for the new user among Management, Monitoring, and Administrator. You can also not enter any tag, meaning it is just a simple user with no access to the management plugin. This make perfectly sense in my case, and so I went for it.

We still have to associate the new user to the new virtual host, so we get back to the "Virtual Hosts" tab, select your v-host from the "All virtual hosts" table, click on the "Set permission" section, select the newly created user from the drop-down list and then click on "Set permession".

That's it. Now we have a brand new v-host that works only for the newly specified user.

But before using them in a client, let's mess it up a little bit more. Say that we don't want to use the standard rabbit port but a custom 6391. We can tell to rabbit which port to use setting the environment variable RABBITMQ_NODE_PORT to the required value. So, in Windows we could write a tiny command shell like this one:

@echo off
setlocal
set RABBITMQ_NODE_PORT=6391
rabbitmq-server.bat
endlocal

Running the standard rabbitmq-server.bat the rabbit broker would run on its standard port 15672 (or 55672 for versions before 3.0), running our script it would refer to our 6391 port.

Having changed the configuration as described, now we want to write a very simple direct exchange rabbit application that uses it. It is quite easy, but I went out of time, and I am forced to show you that in the next post.

Go to the full post

Simple Coherence observer

Once you have put some elements in a Oracle Coherence named cache, you are often interested in checking out if someone else is doing anything in there, inserting, updating, or even deleting from it.

What you could want is observing what happens to your cache, probably something like using the observer pattern.

Doing that with Coherence is pretty easy, we need to define a class extending AbstractMapListener, to specify what we want actually do in case of insertion, editing or deletion on the cache; than we call addMapListener() on our cache passing an instance of such class, and basically we are done.

A very simple example of a map listener could be:

public class MyCacheListener extends AbstractMapListener {
    @Override
    public void entryInserted(MapEvent event) {
        System.out.println("*** Inserted: " + event);
    }

    @Override
    public void entryUpdated(MapEvent event) {
        System.out.println("*** Updated: " + event);
    }

    @Override
    public void entryDeleted(MapEvent event) {
        System.out.println("*** Deleted: " + event);
    }
}

It is not a fancy implementation, it just dumps to standard output the event signaling the item affected in the cache, but starting from this we can create a more useful functionality.

The class that works with the cache would probably implement a method like this:

public void observe() {
    cache.addMapListener(new MyCacheListener());
}

And this is a piece of code that would start observing on a Coherence cache:

CacheObserver cohCli = new CacheObserver(); // 1
cohCli.observe();

cohCli.checkPut("key", "value"); // 2
cohCli.getCheck("key", "value");
cohCli.checkPut("key", "change");
cohCli.checkRemove("key");

// ...

1. In my test code, CacheObserver extends the SimpleCache class seen in the previous post adding the observe() method seen here above.
2. Any time a change is done in the cache, the observer would dump the event generated to the screen.

Go to the full post

Basic Coherence functionality

Once you have setup Oracle Coherence in your development environment, and you have tested a simple Hello World application, you are ready to write something moderately more interesting.

In the following example I am using just a pair of Coherence classes. They still have in their package name a "Tangosol" reference, from the company that now, like many other ones, is part of Oracle:

import com.tangosol.net.CacheFactory;
import com.tangosol.net.NamedCache;

public class SimpleCache {
    protected NamedCache cache;
    protected static final String CACHE_NAME = "MyCache";

    public SimpleCache() { // 1
        CacheFactory.ensureCluster();
        cache = CacheFactory.getCache(CACHE_NAME);
    }

    public void dtor() { // 2
        CacheFactory.shutdown();
    }

    public void checkRemove(String key) {  // 3
        System.out.println("+++ removing " + key);
        if(cache.containsKey(key)) {
            cache.remove(key);
        }
        else {
            System.out.println("+++ " + key + " is not in cache");
        }
    }

    public void checkPut(String key, String value) { // 4
        System.out.println("+++ putting " + key);
        if(cache.containsKey(key)) {
            System.out.println("+++ " + key + " already in cache");
        }
        cache.put(key, value);
    }

    public void getCheck(String key, String expValue) { // 5
        System.out.println("+++ getting " + key);
        String output = (String)cache.get(key);
        if(output == null)
            System.out.println("+++ " + key + " is not in the cache");
        else if(output.compareTo(expValue) != 0)
            System.out.println("+++ Unexpected: " + expValue + " != " + output);
    }

    public void full(String key, String value) { // 6
        checkPut(key, value);
        getCheck(key, value);
        checkRemove(key);
    }

    public static void main(String[] args) {
        SimpleCache cohCli = new SimpleCache();

        if(args.length == 0)
            cohCli.full("key", "value"); // 7
        else {
            if(args[0].matches(".*[Pp].*")) // 8
                cohCli.checkPut("key", "value");
            if(args[0].matches(".*[Gg].*"))
                cohCli.getCheck("key", "value");
            if(args[0].matches(".*[Rr].*"))
                cohCli.checkRemove("key");
        }
        cohCli.dtor();
    }
}

1. In the class constructor we get the cache from the factory.
2. From the name of this method you could correctly guess I am more a C++ that a Java guy. If you wonder, dtor is a common short name for destructor, there is not such a beast in Java, but the name should be thought as an hint to the user programmer - call it when you are done with an instance of this class.
3. Before removing an element in the cache, ensure it actually is in it.
4. If an element with the specified key is already in the cache, issue a warning before storing the new value.
5. Get an element, and check that its value matches the expectation.
6. Utility method, combines (3), (4), and (5) in an unique call.
7. If no parameter is passed to the Java application, a full test is performed, putting, getting, and finally removing a key-value in the cache.
8. The passed parameter is expected to be a string stating what we want to do. A "pgr" is a synonim for a full test; if I want just put I should specify just "p" (or "P") and so on.

Go to the full post

Filtering RabbitMQ messages by patterns

The RabbitMQ direct exchange type is surely an improvement over the fanout exchange, but when we need a higher degree of flexibility, there is a more powerful exchange type, topic, that could come in our help.

The topic routing key should follow a specific pattern, a list of words separated by dots. The consumer could use a couple of wildcards: the star (*), that is considered a synonim for any single word, and the hash (#) for any number of dot-separated words.

So "one.*" would match "one.two", but not "one.two.three", that would be instead a match for "one.#".

The producer here generates a few messages for routing keys that follows the rule quality.color.animal, and then sends an empty message with the control routing key to terminate the clients execution:

private static final String EXCHANGE_NAME = "logDirect";

private enum Quality { Quick, Lazy, Quiet }
private enum Color { Blue, Red, Yellow }
private enum Animal { Elephant, Rabbit, Fox }
private static final String RK_CONTROL = "Control";

private void producer() {
    // ...
    try {
        // ...
        channel.exchangeDeclare(EXCHANGE_NAME, "topic"); // 1

        for(Quality q: Quality.values())
            for(Color c: Color.values())
                for(Animal a: Animal.values()) {
                    String key = q.name() + '.' + c.name() + '.' + a.name(); // 2
                    channel.basicPublish(EXCHANGE_NAME, key, null, "Hello".getBytes());
                    System.out.println("Sent message using routing key " + key);
                }
        channel.basicPublish(EXCHANGE_NAME, RK_CONTROL, null, null);
        // ...

1. The exchange is declared as a topic.
2. For example, Quick.Red.Fox is one of the keys generated in this triple for loop.

The consumer accepts in input a topic, that will be used in the binding between the client and the temporary queue created locally:

private void consumer(String topic) {
    // ...
    try {
        // ...
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();

        channel.queueBind(queueName, EXCHANGE_NAME, RK_CONTROL);
        channel.queueBind(queueName, EXCHANGE_NAME, topic);
        // ...

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            byte[] body = delivery.getBody();
            String key = delivery.getEnvelope().getRoutingKey();
            if(key.compareTo(RK_CONTROL) == 0 && body.length == 0) {
                System.out.println("Control terminator detected.");
                break;
            }
        // ...

You could try to run this producer-consumer couple passing each time a different input string to the consumer, checking for the result. The client should be up before the server starts, otherwise all its messages will be lost.

Try launching it with a parameter like Quiet.#, and you should see as output all nine messages generated by the server that have a key starting with Quiet; an input pattern like *.Red.* will result in the nine messages having Red as second word in the key, including the Quick.Red.Fox; a wrong input like Lazy.* won't give back any message, since it would match only with two-word keys having Lazy at the beginning, but currently our server generates only three word keys.

The full Java class source code is available on github. This post is based on fifth installment of the official RabbitMQ tutorial.

Go to the full post

RabbitMQ direct exchange

We have seen how to use a fanout exchange, now it is time to see a direct exchange at work. We are going to implement a router messaging pattern, where a producer emits messages specifying a routing key, and the consumers accepts all and only the messages that are associated to a specific routing key. In this sense we can say that in this scenario a consumer receives messages selectively.

The producer sends messages with a few routing keys symbolizing a different severity for the associated message. Moreover a special routing key is defined for control messages that could be used to internal management:

private static final String EXCHANGE_NAME = "logsDirect"; // 1
private enum Severity { Debug, Info, Warning, Error }; // 2
private static final String RK_CONTROL = "Control";

private void producer() {
    // ...
    
    try {
        // ...
        channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 3

        for(Severity s: Severity.values()) {
            channel.basicPublish(EXCHANGE_NAME, s.name(), null, "Hello".getBytes()); // 4
            System.out.println("Sending message with severity " + s.name());
        }
        channel.basicPublish(EXCHANGE_NAME, RK_CONTROL, null, null); // 5
        // ...

1. A new exchanges is used.
2. This application uses five streams of messages, four different severity log message levels, and a stream of control messages.
3. The exchange type is "direct". This means that a message goes to the queues whose binding key exactly matches the routing key of the message. Direct exchange becomes equivalent to fanout when all the queues uses the same routing key.
4. Publish to an exchange specifying the routing key.
5. Sending an empty message for the control routing key.

The consumer subscribes to the routing key passed by the user, besides, each consumer gets the control messages:

private void consumer(String[] subscriptions) { // 1
    // ...
    try {
        // ...
        channel.exchangeDeclare(EXCHANGE_NAME, "direct"); // 2
        String queueName = channel.queueDeclare().getQueue(); // 3

        channel.queueBind(queueName, EXCHANGE_NAME, RK_CONTROL); // 4
        for(String subscription: subscriptions) { // 5
            boolean matching = false;
            for(Severity sev: Severity.values()) {
                if(subscription.equalsIgnoreCase(sev.name())) {
                    channel.queueBind(queueName, EXCHANGE_NAME, sev.name()); // 6
                    System.out.println("Subscribing to " + sev.name() + " messages");
                    matching = true;
                    break;
                }
            }
            if(!matching) // 7
                System.out.println(subscription + " severity not available");
        }

        // ...
        
        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            String key = delivery.getEnvelope().getRoutingKey(); // 8
            byte[] body = delivery.getBody();
            if(key.compareTo(RK_CONTROL) == 0 && body.length == 0) { // 9
                System.out.println("Control terminator detected.");
                break;
            }
        // ...

1. Each passed string, if matching with the specified Severity enum, is used to subscribe to a specified routing key.
2. The exchange is declared of the same producer type.
3. A temporary queue is created, see previous post for details.
4. All consumers subscribe to the control message flow.
5. Check all the strings passed by the user, to set the custom subscriptions.
6. Bind the queue used by this consumer to this routing key.
7. The string passed by the user was not recognized, let's issue a warning.
8. Extract the routing key, as set by the producer, from the delivered message.
9. A control empty message is here conventionally considered as a terminator.

The full Java class code is available on github. This post is based on fourth installment of the official RabbitMQ tutorial.

Go to the full post

Pub-Sub with RabbitMQ

The publish-subscribe messaging pattern is used when we want each consumer to get all the messages published by the producer. We have many clients acting as subscriber to a single publisher.

Till the previous post, we used a simplified version of the rabbit messaging schema. There, the producer was connected directly to the queue that was accessed by the consumers. Usually we add an extra layer, called "exchange", that helps managing easier more complex models.

There are four exchange types: direct, topic, headers, fanout. Here we are interested in the last one, the fanout, that is used to broadcast all the messages received by the exchange to each queue known by it.

The main change in the producer is that we declare and use an exchange instead of a queue:

private static final String EXCHANGE_NAME = "logs";

public void producer() {
    // ...
    try {
        // ...
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); // 1

        System.out.println("Sending a message");
        channel.basicPublish(EXCHANGE_NAME, "", null, "a message".getBytes()); // 2
        System.out.println("Sending an empty message");
        channel.basicPublish(EXCHANGE_NAME, "", null, null); // 3
        // ...
}

1. In this way we create a fanout exchange. After this, we can see that exchange on the rabbit server, using the command

rabbitmqctl list_exchanges

If we are not using an exchange, we should fall back to the default nameless ("") one. We have already seen it in the previous examples.
2. Publishing a message to an exchange is not much a difference from publishing to a queue. As you should remember, publishing to the default exchange is just a matter of specifying a queue instead:

channel.basicPublish("", "hello", null, message.getBytes());

3. I stick to the convention of considering an empty message as a terminator for the subscribers.

On the client side, it is interesting to see how a temporary queue is created, one for each consumer, and used:

public void consumer() {
    // ...

    try {
        // ...
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        String myQueue = channel.queueDeclare().getQueue(); // 1
        channel.queueBind(myQueue, EXCHANGE_NAME, ""); // 2

        System.out.println("Consumer waiting for messages.");
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(myQueue, true, consumer);

        while (true) {
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();
            byte[] body = delivery.getBody();
            if(body.length == 0){
                System.out.println("Empty message received: terminating.");
                break;
            }

            String message = new String(body);
            System.out.println("Received: " + message);
        }

        // ...

1. We delegate to the channel the task of creating a non-durable, exclusive, autodelete queue, generating internally a random unique name, that is returned so that we can use it below.
2. This is how we bind a queue to an exchange. To see the currently active bindings, we can use this command:

rabbitmqctl list_bindings

The complete code for this example is available on github. The post is based on the third part of the RabbitMQ official tutorial.

Go to the full post