Always close a RabbitMQ connection

I see now that in the previous rabbit examples I have missed to ensure that a connection would always be closed at the end of the program. The result is that a rabbit process, under some specific circumstance, could unhappily hang until an interrupt terminates its sad life. Remember not to do the same mistake in your production code.

The typical (bad) code is something like:

// ...
try {
    Connection connection = factory.newConnection(); // 1
    Channel channel = connection.createChannel();

    // ...

    connection.close(); // 2 !! BAD !!
}
catch(IOException e) {
    // ...
}

1. A connection is created and open
2. The connection is closed

The issue is that in case of an exception after (1) but before (2), the cleanup connection code is not called, since the control jumps directly to the catch section.

The solution is pretty easy, we should add a finally clause, and clean the connection there:

// ...
try {
    Connection connection = factory.newConnection();
    // ...
    // 1
}
catch(IOException e) {
    // ...
}
finally { // 2
    try{
        if(connection != null) connection.close();
    } catch(IOException e) {} // 3
}

1. The connection cleanup is not done here anymore.
2. The finally section is always execute, whatever happens above.
3. For what said in (2), we should ensure the connection has been actually been instantiate, before calling its close() method. Besides, we have to try-catch it, since it could throws an IO exception.

For tiny test applications, this is not such an important remark. It is probably better stressing other points and leaving out this detail. But in real code, forgetting to adequately protect the connection cleanup procedure could lead to serious problems.

Go to the full post

Passive or active queues?

In the previous RabbitMQ examples I have written, I have always declared queues calling Channel.queueDeclare(). But it is also possible to do it calling Channel.queueDeclarePassive(). As you could easily guess, the first method is for an active queue declaration, and the second for a passive one. The point of this post is: what a passive queue is and how to choose between declaring a queue as passive or active.

If you play with queue setting, one day or another you will end up getting an exception, due to the fact that once a queue is created, you can't access it specifying a different setting:

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)
...    
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error;
 reason: {#method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - 
 parameters for queue 'ack' in vhost '/' not equivalent, class-id=50, method-id=10),
 null, "[B@19d3b3a"}
    at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:67)
...

In this case I tried to access a queue called "ack", declared in the default virtual host, passing the parameter "autodelete" set to true, where it was created with false.

When this happens, you could have a look to the RabbitMQ administrator, specifically in the Queues tab, to get all the available information on your queue.

But back to the main theme of the post.

If a queue is declared passively, RabbitMQ assumes it already exists, and we just want to establish a connection to it. What if you call Channel.queueDeclarePassive() and the queue is not there? I guess you already know the answer, you get a fat exception:

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)
...
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error;
 reason: {#method<channel.close>(reply-code=404, reply-text=NOT_FOUND -
 no queue 'ack' in vhost '/', class-id=50, method-id=10), null, "[B@140fee"}
    at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:67)
...

Close to the exception shown above, but here the reply text is a NOT_FOUND that shows clearly what the problem is. Rabbit was running happily assuming the queue was there, but actually it wasn't.

So, we use a passive queue declare when we can safely assume the queue is already there, and its non-existence would considered an exceptional, almost catastrophic, event.

The safer active queue declaration has the obvious downside of being more expensive, and requiring the user to provide the setting each time a queue declaration is issued. An hybrid approach (active on the server, passive on the client) could make sense if the application has a rigid protocol ensuring that a component would always start before the other(s).

Go to the full post

ConnectionFactory setting by URI

In the previous post we set properties on a RabbitMQ ConnectionFactory constructing an object and then calling all the relevant setter on it. Sometimes it is handier to directly create a ConnectionFactory object passing an URI in its constructor.

We can use both a standard Java URI object, as defined in the java.net package, or a String in the expected AMQP format. Let's say that we already have the URI as a string, and we can pass it to our rabbit application, we can use this information directly:

private ConnectionFactory getFactory(String uri) throws IOException { // 1
    ConnectionFactory factory = new ConnectionFactory();
    try {
        factory.setUri(uri); // 2
    } catch (URISyntaxException|NoSuchAlgorithmException|KeyManagementException e) {
        throw new IOException("Can't set connection factory URI", e); // 3
    }
    return factory;
}

1. We expect uri to be something like "amqp://user:password@127.0.0.1:6391/test", where "test" is the name of our virtual host. We are not forced to specify all parts in the AMQP URI, but the host name must be passed if at least one property among username, password, or port is given.
2. The passed URI is carefully checked, and three different exceptions could be thrown to acknowledge a specific error.
3. The error handling here is very loose, any passible exception is converted in an IOException and forwarded to the caller. The reason for this is just keeping the code simple, since I don't care much of having such detailed information on the reason for failure. This shouldn't be an optimal choice for production code.

Producer and consumer should be slightly changed to try-catch also on this method:

public void consumer(String uri) {
    try {
        ConnectionFactory factory = getFactory(uri);
        Connection connection = factory.newConnection();
        
        // ...

        connection.close();
    }
    catch (IOException|InterruptedException e) {
        e.printStackTrace();
    }
}

Go to the full post

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