Installing ActiveMQ

What you need to run ActiveMQ on your machine is a recent Java development kit, and ActiveMQ itself. You can find the latter in the official Apache ActiveMQ download page, and you can get more information on the process in the related Getting Started page.

The ActiveMQ broker is ready to run out of the box, on its default port 61616, simply running a batch file in its binary folder. In my case, I have installed ActiveMQ on Windows XP (I know, we are in 2012, but this ancient operating system is still alive and kicking) in a folder named C:\dev\apache-activemq-5.5.1, and I run its broker executing from a shell in that folder the command
bin\activemq
As a feedback I get some log information including a notification of the JVM used, and on the started ActiveMQ components and subsystems. For the sake of testing the installation, the two most interesting lines are:
INFO | Listening for connections at: (your machine):61616
...
INFO | ActiveMQ Console at http://0.0.0.0:8161/admin
Whenever you want to check if the connection is still alive, you can run netstat, as I did here:
netstat -na|find "61616"
 TCP    0.0.0.0:61616          0.0.0.0:0              LISTENING
Meaning: the port 61616 on localhost is listening for a TCP connection. All is running as expected.

Besides, we can access the ActiveMQ console from our favorite browser at the http://localhost:8161/admin address, and see there a number of administrative information on the broker.

Go to the full post

ActiveMQ 5.5 and SLF4J 1.6

Working on a simple Hello World example for ActiveMQ, I bumped in a conflict between this Apache Message-Oriented-Middleware (MOM) and the SLF4J logging system. It is nothing serious, but I guess it is worth to enter in some detail on this issue.

SLF4J is the used as logger by ActiveMQ. Unfortunately, still in version 5.5, the SLF4J bundled version is 1.5, that's a pity, because starting from SLF4J version 1.6 if you place no binder in your application classpath, the NOP logger is assumed. The effect is that all your log goes to a logical respective of /dev/null, as to say, it disappears in thin air. That's no fun, but better than the behavior of 1.5: crashing miserably with an error like this:
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for
 further details.
Exception in thread "main" java.lang.NoClassDefFoundError: 
    org/slf4j/impl/StaticLoggerBinder
        at org.slf4j.LoggerFactory.getSingleton(LoggerFactory.java:230)
...
So, you really have to plug in a binder. I already had at hand a 1.6 plugin, I tried to used it, but I got another error:
SLF4J: The requested version 1.6 by your slf4j binding is not compatible
 with [1.5.5, 1.5.6, 1.5.7, 1.5.8, 1.5.9, 1.5.10, 1.5.11]
No compatibily.

There are a couple of solution to this issue. Or you use a plugin for that specific version, or you put the SLF4J API jar of your choice in your classpath before the ActiveMQ full jar.

I opted for the second choice, and my classpath is now referring to these jar - in this order:
  • slf4j-api-1.6.4.jar
  • activemq-all-5.5.1.jar
  • slf4j-simple-1.6.4.jar
Actually, the SLF4J plugin could be in any place, but it is crucial that the API SLF4J jar is before the ActiveMQ one.

Go to the full post

Improved RPC by JSON

If the data exchange is meant to use only strings, the basic RPC support provided by RabbitMQ is more than adequate, and the StringRpcServer class makes it even easier. But what if we want to use other data types? The rabbit answer to this question is JSON-RPC. This RPC protocol based on JSON is implemented in a couple of classes, JsonRpcServer and JsonRpcClient, that take care of the low leve details and let us free to use RPC in a more natural way.

To get acquainted to it, I went through the JSON-RPC example provided in the test folder of the source RabbitMQ distribution, package com.rabbitmq.examples, files HelloJsonService.java, HelloJsonClient.java, and HelloJsonServer.java. I have reworked it and you can see my commented version in this post, full Java source code is available on github.

The service

The JSON-RPC interface between client and server is represented by a service, that exposes to the client which methods are available on the server:
public interface JsonSvc {
    String greeting(String name);
    int sum(List<Integer> values);
}
This interface is going to be implemented by a class, I called it MyJsonSvc, used internally by the RPC server.

The client

In the client, the service is used to provide access to the server functionality, as shown here:
// ...
JsonRpcClient client = new JsonRpcClient(channel, "", QUEUE_NAME, RPC_TIMEOUT_ONE_SECOND); // 1
JsonSvc service = (JsonSvc)client.createProxy(JsonSvc.class); // 2

System.out.println(service.greeting("Rabbit")); // 3

List<Integer> numbers = new ArrayList<>();
numbers.add(1);
numbers.add(2);
numbers.add(3);
System.out.println("1 + 2 + 3 = " + service.sum(numbers));
// ...
client.publish(null, null); // 3
1. A JSON-RPC client is created, specifying a channel, a queue, and also an optional timeout in milliseconds.
2. This is the core of the example, the rabbit JSON-RPC client creates a proxy for the passed interface, then we can use it in our client as shown below.
3. We could still use the client as a plain RpcClient client, bypassing the JSON layer. Here we are sending an empty message with no associated properties.

The server

Implementing the JSON-RPC capabilities in the server is straightforward:
// ...
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
JsonRpcServer server = new MyRpcServer(channel, QUEUE_NAME, JsonSvc.class, new MyJsonSvc()); // 1

System.out.println("JSON-RPC server is up.");
server.mainloop(); // 2
// ...
1. We create a JsonRpcServer object specifying a service on which it would operate, and the interface that it should use to access it.
2. Then we start looping on it.

You have surely noticed in the code here above that I didn't create a plain JsonRpcServer, but a mysterious MyRpcServer. I did that because I wrote the client to be a bit smarter than usual, calling not only the JsonSvc methods, but also sending a plain (empty) message. To give the server a way to manage it as expected (in this case an empty message is seen as a signal to terminate the server run), I have to add functionality to the JsonRpcServer, as we already have seen for StringRpcServer. Actually, all we need is just reimplementing handleCast(), so that we can react correctly to an empty message:
// ...
private class MyRpcServer extends JsonRpcServer {
    // ...
    @Override
    public void handleCast(byte[] requestBody)
    {
        if(requestBody.length == 0) {
            System.out.println("Empty message, terminating.");
            terminateMainloop();
        }
    }
}

Go to the full post

RPC for strings

Writing a RabbitMQ RPC (Remote Procedure Call) application is not difficult, still we have to deal with a few internal details that could easily be hidden in a utility class. Actually, Rabbit gives us two hierarchies designed for that, based on com.rabbitmq.client.RpcServer and com.rabbitmq.client.RpcClient.

RpcClient is flexible enough to be used directly in most cases, while the RpcSever has to be adapted to the effective application requirements. The example of this post is based on the HelloClient - HelloServer classes included in the source RabbitMQ distribution (you should find them in the test directory, in the com.rabbitmq.examples package). The Java source code for my variation is available on github.

The client sends a string message to the server, and receives back a response. It could also send an empty message, that would be interpreted by the server as a command to shutdown, and in this case no answer is expected.

We delegates to RpcClient all the low level details, and this is the slim resulting code:
// ...
RpcClient service = new RpcClient(channel, "", QUEUE_NAME); // 1

if(arg.equalsIgnoreCase("x")) { // 2
    System.out.println("Terminating server");
    service.publish(null, null);
}
else {
    String result = service.stringCall(arg); // 3
    System.out.println(result);
}
// ...
1. Instantiate an RPC client that is going to work on the specified channel and queue. The second parameter, not used here, is reserved for the exchange; in that case the third parameter would be used for the routing key.
2. If the user pass an x (upper or lowercase) we interpret it as a request to shut the server down. In this case we the raw RpcClient.publish() method, that expects two parameter, the properties associated to the message, and the message itself, as an array of bytes. In this case both of them are null.
3. Usually we rely on RpcClient.stringCall(), that wraps a complete exchange with the server. The passed message is sent, and when the reply arrives is passed back to the caller.

The server it a bit more complicated. Firstly we need to specialize the StringRpcServer, we could create an anonymous inner class on the fly, but here I wrote it as a plain inner class, aiming to readability:
private class MyRpcServer extends StringRpcServer {
    public MyRpcServer(Channel channel, String queueName) throws IOException {
        super(channel, queueName);
    }

    @Override
    public String handleStringCall(String request) { // 1
        System.out.println("Input: " + request);
        return "Hello, " + request + "!";
    }

    @Override
    public void handleCast(byte[] requestBody) // 2
    {
        if(requestBody.length == 0) {
            System.out.println("Empty message, terminating.");
            terminateMainloop(); // 3
        }
    }
}
1. The standard case, a string is received from the client, here we use it to generate the result, and we send it back to the caller.
2. Less commonly, we want to consume the message received from the client without sending back anything.
3. This is the method to call to terminate the looping on the RPC server, exactly what we need here.

The server itself instantiates its RPC server object and let it looping on the messages arriving from the clients:
// ...
channel.queueDeclare(QUEUE_NAME, false, false, false, null);

StringRpcServer server = new MyRpcServer(channel, QUEUE_NAME);
System.out.println("RPC server is up");
server.mainloop();
// ...
RpcServer.mainLoop() loops indefinitely on RpcServer.processRequest() that checks if both correlationId and replyTo are set among the request properties. If so, RpcServer.handleCall() is called, and its returned value is published back to specified queue. Otherwise, RpcServer.handleCast() is called.

Go to the full post

Fibonacci RPC client

In this RPC (remote procedure call) RabbitMQ Fibonacci application, once written the server component, we are almost done.

The client simply has to send a message to the server, and just sits there waiting for the answer. The variation is that I wrote such a powerful client that could even kill the server. To do that it just has to send an empty message. This is not a very safe behavior, but it is handy to show how to manage uncommon messages, at least in the RPC by messaging pattern, where the client doesn't wait for an answer.

Here is the most interesting part of the client code, you could also have a look to the source code for the complete Java class, that contains both server and client functionality:
// ...
if(arg.equalsIgnoreCase("x")) {
    System.out.println("Terminating Fibonacci server");
    channel.basicPublish("", RPC_QUEUE_NAME, null, null); // 1
}
else {
    String queueName = channel.queueDeclare().getQueue();
    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(queueName, true, consumer); // 2

    String id = UUID.randomUUID().toString(); // 3
    BasicProperties props = new BasicProperties.Builder().correlationId(id).replyTo(queueName).build(); // 4
    channel.basicPublish("", RPC_QUEUE_NAME, props, arg.getBytes()); // 5
    System.out.println("Reply to " + queueName + ", " + id);

    System.out.print("fibonacci(" + arg + ") = ");
    while (true) { // 6
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        if (delivery.getProperties().getCorrelationId().equals(id)) { // 7
            System.out.println(new String(delivery.getBody())); // 8
            break;
        }
    }
}
// ...
1. This is the killer empty message for the RPC Fibonacci server.
2. An implicit acknowledgment is enough in this case.
3. As explained talking about the server, a unique correlation id is used to ensure the matching between request and reply. Here we generate it using the UUID facility.
4. The correlation id and the name for the specific queue created exclusively for this client is passed to the server in the properties associated to the message.
5. Message and properties are sent.
6. We loop indefinitely, discarding all the messages that are not relevant.
7. That's it! The correlation id matches.
8. In the message body we find the reply to our question, and we can terminate.

Go to the full post

Fibonacci RPC server

As a first example of RabbitMQ RPC (remote procedure call) application, I have studied the Fibonacci generator provided in the RabbitMQ official tutorial. Here you can find my version of the rabbit Fibonacci server, the rabbit Fibonacci client is in the next post.

I put both client and server in the same class, and you could get the complete Java source code from github.

The server waits indefinitely on a queue, named "rpc_queue", for inputs to be used to calculate a Fibonacci number. It consumes just a message at time, and sends an explicit acknowledgment to the broker when it completes the elaboration. All of this is not very useful here, but it could be seen as a hint for a future redesign where more Fibonacci calculators are available to the huge population of clients requesting this awesome service.

The server accepts only input in the range MIN_INPUT .. MAX_INPUT, and interprets an empty message as a terminator. All the other possible messages are rejected.

Sending the result to the client is a bit tricky. We need the client to tell the server on which queue it is waiting for this answer, and this is not enough. Since a client could send more than one request, at least theoretically, we need to get and send back an identifier, that would correlate the request to the response. This is done putting these values in a properties object associated to the message.

This is the most interesting part of the resulting code:
// ...
try {
    // ...

    channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null); // 1
    channel.basicQos(1); // 2

    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(RPC_QUEUE_NAME, false, consumer); // 3

    System.out.println("RPC Fibonacci calculator is ready");

    boolean terminator = false;
    while (true) {
        String response = ""; // 4

        QueueingConsumer.Delivery delivery = consumer.nextDelivery(); // 5

        try {
            String message = new String(delivery.getBody());
            if(message.isEmpty()) { // 6
                System.out.println("Empty message, terminating");
                terminator = true;
                break;
            }

            int n = Integer.parseInt(message);
            System.out.println("Calculating fibonacci(" + message + ")");
            if(n < MIN_INPUT || n > MAX_INPUT)
                response = "N/A";
            else
                response += fibonacci(n); // 7
        }
        catch (Exception e){
            e.printStackTrace();
        }
        finally {
            if(!terminator) { // 8
                BasicProperties bp = delivery.getProperties(); // 9
                String corrId = bp.getCorrelationId(); // 10
                String replyQueue = bp.getReplyTo(); // 11
                System.out.println("Replying to " + replyQueue + ", " + corrId);
                
                BasicProperties replyProps = new BasicProperties.Builder().correlationId(corrId).build();
                channel.basicPublish("", replyQueue, replyProps, response.getBytes()); // 12
            }
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); // 13
        }
    }
}
catch  (Exception e) {
    e.printStackTrace();
}
finally {
    try{ if(connection != null) connection.close(); } catch(IOException e) {}
}
1. The queue used by this application only.
2. No prefetch is done, just one message is picked up at time.
3. Explicit acknowledgment when the output is generated.
4. It is handy to initialize the response to an empty string, so that we can concatenate the resulting Fibonacci number to it.
5. Wait a client's request.
6. Special case, the client asks the server to shutdown.
7. Call the actual Fibonacci generator.
8. Usually we send back a message to the caller, only in case of termination there is no need of sending back anything.
9. In the delivery we have an object where the properties of the message are stored.
10. We use the correlation id the identify uniquely a request coming from a client
11. In the reply to field we expect to see the queue where the client wants us to put the result.
12. This is the main line in the post. We publish the response to the queue passed by the client, specifying in the properties the correlation id of its request.
13. Only now we can say to the broker it could get rid of the message we read on (5).

Go to the full post

RabbitMQ message acknowledgment

To understand better how it works, could be useful compare the behavior of a very simple rabbit consumer when the acknowledgment flag is set to true or false in the Channel.basicConsume() call.

I have written a stripped down example where the producer puts a message in a queue and then quits. The consumer works with automatic or explicit acknowledgment accordingly to an input value is provided to it. Looking at the code, we can see how the changes are very localized. We specify that the QueueingConsumer for the channel/queue is in one mode or the other, and then, if an explicit ack is required, we give it when the job is done:
// ...
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(QUEUE_NAME, autoAck, consumer); // 1

QueueingConsumer.Delivery delivery = consumer.nextDelivery();
// ...

if(!autoAck) { // 2
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    // ...
}
1. Here we specify if an handshake is required (autoAck sets to false) or if we rely on the standard auto-acknowledging RabbitMQ policy.
2. When we are done with the message, and if we are not in auto-acknowledgment mode, we give our explicit acknowledgment to the rabbit broker.

The complete Java code is on github.

Go to the full post