Showing posts with label JMS. Show all posts
Showing posts with label JMS. Show all posts

JMS Request-Reply

Implementing the Request-Reply messaging pattern with ActiveMQ is not complicated. The client component sends a request to the server on the dedicated queue, it attaches to it, as properties, the destination queue where it expects to find the reply, and a correlation id to uniquely identify the request-reply couple. The server uses these information to generate a consistent reply.

The complete Java source code is on github, here I comment just what I think are the most interesting part of the code.

This it the client, after establishing a connection to the broker:
Destination destination = session.createQueue(REQUEST_QUEUE_NAME); // 1
MessageProducer producer = session.createProducer(destination);
Destination dRep = session.createTemporaryQueue(); // 2
TextMessage message = session.createTextMessage(arg);
message.setJMSCorrelationID(UUID.randomUUID().toString()); // 3
message.setJMSReplyTo(dRep); // 4
producer.send(message);
// ...
MessageConsumer consumer = session.createConsumer(dRep); // 5
Message reply = consumer.receive(); // 6
// ...
1. The queue where the requests are sent.
2. A second queue, for the replies. We could have used a "normal" queue, but it is more common to create a cheaper temporary queue for this task.
3. We could keep track by hand of the unique message ids generated by the application, or we could delegate to Java the generation of an unique id, as I did here.
4. The queue to be used as destination for the reply is stored in the JMSReplyTo message property.
5. A consumer is created on the reply queue.
6. And the client patiently waits for an answer from the server.

The interesting part of the server is where it replies to the client:
if(message instanceof TextMessage) {
    TextMessage answer = session.createTextMessage(); // 1
    answer.setText("Reply to " + ((TextMessage) message).getText());
    answer.setJMSCorrelationID(message.getJMSCorrelationID()); // 2
    MessageProducer producer = session.createProducer(null); // 3
    producer.send(message.getJMSReplyTo(), answer);
}
1. The message to be sent as answer.
2. The correlation id is sent back to the client.
3. A producer with no associated queue is created, we are going to explicitly set the destination in its send() method using the JMSReplyTo message property.

Go to the full post

Selecting messages by custom properties

A JMS message comes with a number of standard properties, providing information on the associated message. For example, each message comes with a message id that could be fetched calling getJMSMessageID(). Besides, we can use custom properties to enrich a message in any way we think make sense for our specific case. An interesting aspect of custom properties is that we can use them to discriminate on which message a consumer should receive.

Think to an application where the producer generates messages for the best offers we have in our hardware store. The consumer could be interested in all or only in a subset of them. For this reason, it could be useful to put in the message payload just a generic description, and use custom properties to store structured information, as the product name and its price.

In this way the consumer could easily specify if it wants to get all the messages, or create a selector on the custom properties. The rules to create a selector are coming by simplification from the SQL conditional expressions.

The complete Java source code of this example is on github, it is written for ActiveMQ, so you should have this MOM installed and working on your development environment.

The producer delegates the most interesting part of its job to the send() method, that creates a message, fills it with specific data, and then send it to its associated queue:
TextMessage message = session.createTextMessage();
message.setText(txt);
message.setStringProperty(PROP_DES, des); // 1
message.setDoubleProperty(PROP_PRICE, price); // 2
producer.send(message);
1. Create a String property named as defined in the first parameter ("Description"), and containing the value stored in the second one ("Pins" and "Nails" in my test example).
2. The PROP_PRICE (actually, "Price") property is of type double.
In this test it is handy using a MessageListener, as discussed in a previous post, so that all the relevant messages are consumed by its onMessage() method:
logger.info("{}: {}", PROP_DES, message.getStringProperty(PROP_DES)); // 1
logger.info("{}: {}", PROP_PRICE, message.getDoubleProperty(PROP_PRICE)); // 2
logger.info("Message id: {}", message.getJMSMessageID()); // 3
logger.info("Message: {} ", ((TextMessage) message).getText());
1. The PROP_DES property is retrieved from the message, as the String that it is.
2. PROP_PRICE is a double, but we could have managed it as a String, calling getStringProperty(). ActiveMQ knows how to convert the internal double value to a String, so this cast works fine. If we tried to do the other way round, extracting a double from (1), we would have got a NumberFormatException.
3. There is not much use for this line here, but it is just to show a standard JMS property at work.

We need just one fundamental step, telling the JMS Session that the consumer has to get only some message:
switch (arg) {
    case "Nails":
    case "Pins":
        filter = PROP_DES + " = '" + arg + "'"; // 1
        consumer = session.createConsumer(destination, filter);
        break;
    case "Cheap":
        filter = PROP_PRICE + " < 15"; // 2
        consumer = session.createConsumer(destination, filter);
        break;
    default:
        consumer = session.createConsumer(destination); // 3
        break;
}
1. The string "Description = Nails" (or "Description = Pins") is passed in the next line to the session as a message selector for the consumer that has to be created. Only the messages true under that condition are delivered to this consumer.
2. Same as above, but the condition is now "Price < 15".
3. A "normal" consumer is created, with no associated selector.

Go to the full post

Asynchronous Hello ActiveMQ consumer

It is simple to modify the hello ActiveMQ example to change the consumer from syncronous to asyncronous.

If we want to asynchronously consume a message, we have to create a class that implements the JMS MessageListener interface, something like:
private class MyMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        if(message instanceof TextMessage) {
            try {
                logger.info("Received {} ", ((TextMessage) message).getText());
            } catch (JMSException e) {
                logger.error("Can't extract text from received message", e);
            }
        }
        else
            logger.info("Unexpected non-text message received.");
    }
}
Then the consumer code changes only slightly:
MessageConsumer consumer = session.createConsumer(destination);
consumer.setMessageListener(new MyMessageListener()); // 1
try{ Thread.sleep(1000); } catch(InterruptedException e) {} // 2
1. Associate to the consumer an instance of our listener.
2. Usually we should implement a more sophisticated way of keeping alive the consumer till all the relevant messages are consumed. But for this simple example sleeping for a second should be enough.

The Java source code for the hello producer-consumer example, both in the synchronous and asynchronous flavor, is on github.

Go to the full post

Hello ActiveMQ

Writing an hello application with ZeroMQ is very easy, RabbitMQ requires a bigger effort, and ActiveMQ is not complicated, but probably the less immediate in the company. This escalation in complication is reflected also in the system's responsiveness (ZeroMQ is blazing fast and, compared to it, ActiveMQ could seem slow) but is well repaid by a higher level of offered services.

It is impossible to say in abstract which Message Queue structure is "better". It is more a matter of selecting the adequate one for specific environment requirements. A reason to decide for ActiveMQ is often that it implements the JMS specification.

In any case, say that we have already decided, and we have installed ActiveMQ on on our machine. Now it is time to write an hello application.

ActiveMQ is bundle with SLF4J as logger. I have talked of the trickiness in their relation in a previous post, so let's assume this is not an issue anymore. If you don't even know what SLF4J is, you could have a look at another post where I have written some notes on its installation.

Producer

The following code is all you need to establish a connection to the ActiveMQ broker and put a text message on a queue:
ConnectionFactory factory = new ActiveMQConnectionFactory(); // 1
Connection connection = null;
try {
    connection = factory.createConnection(); // 2
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 3
    Destination destination = session.createQueue(QUEUE_NAME); // 4
    MessageProducer producer = session.createProducer(destination); // 5
    TextMessage message = session.createTextMessage(arg); // 6
    producer.send(message); // 7
    logger.info("Sending: {}", message.getText()); // 8
} catch(JMSException ex) {
    ex.printStackTrace();
} finally { try { if(connection != null) connection.close(); } catch (JMSException e) {} } // 9
1. Since no address is specified, ActiveMQ assumes we want to connect to default broker, on localhost:61616.
2. A new connection to the browser is created and, in the next line, started.
3. On a connection we create a session, that could be transactional. Here we don't need it, so we pass "false" as first parameter. The second parameter specify if we want to acknowledge explicitly the message consume or not. Here we let ActiveMQ to take care of this.
4. Where the message is going to be sent. If the queue specified does not exist, ActiveMQ creates it for us.
5. The producer is an object created by the session for a specified destination that takes care of the message move from this client to the broker.
6. A text message is created from a String (in this case "arg" is the variable containing it).
7. The producer sends the message.
8. Remember that we are using SLF4J as a logger, if you wander about the strange syntax in the string, you may be interested in the post I have written on the efficiency reason for it.
9. Whatever happens in the try block, we want the connection to be closed.

Consumer

Consuming a message on ActiveMQ is not much different than producing it:
MessageConsumer consumer = session.createConsumer(destination); // 1
Message message = consumer.receive(1000); // 2
if(message == null)
    logger.info("No pending message on {} queue.", QUEUE_NAME);
else if(message instanceof TextMessage) // 3
    logger.info("Received: {}", ((TextMessage)message).getText());
else // 4
    logger.info("Unexpected non-text message consumed.");
1. First difference, we create a consumer on the session.
2. A message is consumed. The passed parameter is a timeout. If no message is waiting for us on the queue, and nothing arrives in a second, the consumer returns a null.
3. In the producer we created a text message, so we expect it to be of the same type here. If this is what we have, we work with it.
4. If we get here we are perplexed: who put a non-textual message on our queue?

You you want to play with the complete source Java code, you can find it on github.

Go to the full post

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