What is rhq-msg?
Before I get into the news about the new rhq-msg repository, let me first take a step back and summarize what rhq-msg is and what I'm trying to accomplish with it.
rhq-msg is a simple messaging API built on top of ActiveMQ and JMS. However, I wanted to isolate the user of the API from as much ActiveMQ and JMS specific classes and code as possible. I wanted a simpler API that provides basic messaging functionality without requiring the user from having to create and manage lots of little specific classes (like JMS Connections, Sessions, Destinations, Consumers, Producers, etc) and without having to know very many specific ActiveMQ details.
I also wanted to ensure that non-Java clients and servers can interact with rhq-msg clients and servers. So the messages that are sent and received from the rhq-msg API are JSON-encoded and thus can be sent to and received by other non-Java endpoints so long as they can handle JSON-encoded messages (ActiveMQ supports non-Java clients, I just wanted to make sure the messages rhq-msg handles can easily flow to/from those non-Java clients as well).
So the point is I have small, simple, easier-to-use messaging API that can talk to non-Java endpoints, but yet still retain all the nice functionality (like guaranteed delivery and things like that) that JMS and ActiveMQ provides.
New Repository for rhq-msg
My past few blog posts were about the prototyping work I'm doing with respect to rhq-msg and rhq-audit.
I just split out rhq-msg and put it in its own rhq-msg repository since it really does belong as a separate project.
I also added a nifty feature to it - you can now use it for request-response workflows. Before, the rhq-msg API really only supported fire-and-forget async messaging. You had a message and you sent it to an endpoint asynchronously and that was that. The problem is sometimes you want a response back, and many times you want to wait for that response to come back in an RPC-like fashion (as opposed to accepting the response asynchrously). I call this a request-response workflow.
Well, the API now supports this. In fact, it supports receiving the response both synchronously (via a Java Future implementation) and asynchronously (via a message listener implementation). More on this below.
A Quick Overview of the API
The main purpose of rhq-msg is to provide a simpler API to do messaging (simpler than, say, JMS). If I can't describe how to send and receive messages via rhq-msg in a couple paragraphs of a blog, then I think I failed :) So let me give a quick overview of what the API calls would look like if you want to send and receive messages. I will cover sending fire-and-forget messages, request-response messaging, and listening for incoming messages.
The main rhq-msg API is found in the rhq-msg-common module.* Common Code For Both Producers and Consumers
Before I talk about how to send and receive messages, let me introduce the few classes both senders (aka producers) and receivers (aka consumers) use.
Each message that flows through rhq-msg is a BasicMessage. You can subclass that to create your own message types, but all messages must derive from BasicMessage. BasicMessage provides the JSON functionality needed to serialize the messages over the wire.
ConnectionContextFactory connects to your rhq-msg broker and creates contexts for both producers and consumers that are then passed to the MessageProcessor so it knows where to send/receive its messages from.
MessageProcessor provides the API to send and listen for messages.
OK, with that out of the way, let's talk about how to send and listen for messages. For these examples, let's assume you have a broker running on the local machine (127.0.0.1) listening to port 17173, you have a BasicMessage you want to send (or you want to listen for a BasicMessage) and the messages are to be found on a queue named "Foo".
* Sending Fire-and-Forget Messages
Fire-and-forget means you send a message off to an endpoint and forget about it - you don't need or expect a response back. You assume the message broker will deliver the message and the recipient will process the message properly.
Endpoint endpoint = new Endpoint(Endpoint.Type.QUEUE, "Foo");
ConnectionContextFactory factory = new ConnectionContextFactory("tcp://localhost:17173");
ProducerConnectionContext context = factory.createProducerConnectionContext(endpoint);
MessageProcessor processor = new MessageProcessor();
processor.send(context, basicMessage);
factory.close();
Here you create your factory so it connects to your broker. Using the factory, create a producer context. With that new producer context and your basic message, just tell a message processor to send it. That's it. Message is away. You can keep your factory around to create additional contexts which share the connection. But once you are done sending messages, close the factory so it cleans up all resources and closes its connection to the broker. Note that every class you see is a rhq-msg object. No ActiveMQ or JMS classes are used to code this up (though, obviously, under the covers, ActiveMQ and JMS is used heavily).
* Sending Request-Response Messages
Many times when you send a message request, you want a response back. In this example I will show how you can have an RPC-like request-response workflow. I will assume that the remote endpoint will process my BasicMessage and send back to me its own BasicMessage response. Note that, as I mentioned earlier, I could use custom message types (I don't have to use BasicMessage) but those custom message types must always derive from BasicMessage.
Endpoint endpoint = new Endpoint(Endpoint.Type.QUEUE, "Foo");
ConnectionContextFactory factory = new ConnectionContextFactory("tcp://localhost:17173");
ProducerConnectionContext context = factory.createProducerConnectionContext(endpoint);
MessageProcessor processor = new MessageProcessor();
Future<BasicMessage> future = processor.sendRPC(context, basicMessage, BasicMessage.class);
BasicMessage response = future.get(30, TimeUnit.SECONDS);
The first several lines are identical with the fire-and-forget example above. The difference starts with the API call made on the processor - here you call sendRPC() passing in the same context and your outgoing basicMessage request object as before, but you also now pass in the message type of the expected response message. In return, you get a Future object which you can use to retrieve the response when it is received. In this example, it blocks for 30 seconds waiting for the response.
There is another API I won't talk about in detail here, but suffice it to say there is another request-response API you can call on the MessageProcessor (as opposed to sendRPC()) and that is "sendAndListen()". sendAndListen() also allows you to send a request and listen for a response, but this API allows you to give your own message listener so it can wait for and receive the response, rather than go through a Future object. It seems more intuitive and easier to use Future, but in case you want to write your own listener object and listen for the response that way, this is doable. I'll explain the listener API below - it would be the same thing you would need to pass to sendAndListen().
* Listening for Incoming Messages
This last example shows how you implement consumers via rhq-msg API. You implement these via listeners. These listeners are your server-side code - they listen for and accept incoming messages from producers and process those messages. There are two general types: those that do not send responses back, and those that do. I'll cover both in that order.
To listen for "fire-and-forget" messages (that is, messages that are sent that do not require a response sent back to the sender) you implement a BasicMessageListener and hand that listener off to the message processor:
Endpoint endpoint = new Endpoint(Endpoint.Type.QUEUE, "Foo");
ConnectionContextFactory factory = new ConnectionContextFactory("tcp://localhost:17173");
ConsumerConnectionContext context = factory.createConsumerConnectionContext(endpoint);
MessageProcessor processor = new MessageProcessor();
processor.listen(context, listener);
where "listener" is a subclass implementation of BasicMessageListener. An example is:
public class MyCustomListener extends BasicMessageListener<BasicMessage> {
@Override
protected void onBasicMessage(BasicMessage receivedMessage) {
// Process the received message.
}
}
Notice that we still create a ConnectionContextFactory (because we still need a connection to the message broker) but we ask that factory to create for us a consumer context this time. We call "listen()" on the message processor, passing to it that consumer context and our custom listener. That listener is now listening for messages to arrive on the "Foo" queue and will process them.
What about request-response processing? If your listener needs to send data back to the sender, it needs to implement a RPCBasicMessageListener. Other than that difference, the main code is still the same:
Endpoint endpoint = new Endpoint(Endpoint.Type.QUEUE, "Foo");
ConnectionContextFactory factory = new ConnectionContextFactory("tcp://localhost:17173");
ConsumerConnectionContext context = factory.createConsumerConnectionContext(endpoint);
MessageProcessor processor = new MessageProcessor();
processor.listen(context, listener);
but this time "listener" is a subclass implementation of RPCBasicMessageListener. An example is:
public class MyCustomRPCListener extends RPCBasicMessageListener<BasicMessage, AnotherMessage> {
@Override
protected AnotherMessage onBasicMessage(BasicMessage receivedMessage) {
// Process the received message.
AnotherMessage response = ...create your response message object...;
return response;
}
}
Note the onBasicMessage() method now can return a non-void type - the response message type. This return type (the specific response message type) is declared in the generic type definition found in your class definition. The first generic type is that of the expected incoming message type, the second generic type is that of the response message type.
How to Build rhq-msg?
rhq-msg is composed of a set of Maven modules. From the parent root module directory, just run "mvn install" and everything will build, tests will run, and artifacts will be packaged.
There are also Eclipse project files that allow you to work with rhq-msg via M2E (the Eclipse Maven plugin).
How to Run a rhq-msg Broker
If you want to run your own code that utilizes the rhq-msg API, you will need to run a rhq-msg broker (which is really just an ActiveMQ broker). You can run a rhq-msg broker easily in a couple ways. You can run the EmbeddedBroker from the Maven command line or you can install the EmbeddedBroker in a WildFly 8 installation. This EmbeddedBroker code is found in the rhq-msg-broker Maven module.
* Running EmbeddedBroker from the command line
Starting a broker as a standalone process is relatively painless. You can use Maven to do it:
mvn -Prun-test-broker install
Right now the rhq-msg broker is not packaged with all its dependencies, but if you do it yourself (that is, get all the dependency jars and start Java with the appropriate classpath) you can run straight from the Java command line:
java -cp <the-appropriate-classpath> -jar rhq-msg-broker-0.1.jar -c broker-config.xml
where "broker-config.xml" is an actual ActiveMQ XML configuration file (you can also pass in a simpler ActiveMQ .properties configuration file if you wish). Examples are test-broker.xml and test-broker.properties.
* Installing EmbeddedBroker in WildFly 8
One of the artifacts within rhq-msg is a WildFly Extension Module that provides a broker. Once installed in WildFly, you will have an rhq-msg broker subsystem running within WildFly itself. It is a handy way to have your own broker running in your own WildFly instance. So this means you can have, for example, an rhq-msg client within a web application running in WildFly and provide your own broker for that client running in the same WildFly instance. Technically, this is useful for any ActiveMQ or JMS client - since this broker is nothing more than an ActiveMQ broker and can serve any client from anywhere, not just those using the rhq-msg API.
First, download and unzip the WildFly 8 app server (I have not tested on WildFly 9+). I'll use <wildfly-install-dir> to indicate where you installed WildFly.
Now install the custom rhq-msg broker WildFly Extension Module into WildFly. You can use the new Maven plugin "wildfly-extension-maven-plugin" to do this - the rhq-msg-broker-wf-extension Maven module has integrated that Maven plugin. I talk about this in a previous blog post. So, simply running this Maven command will install your rhq-msg broker WildFly Extension Module for you:
mvn -Dorg.rhq.msg.broker.wildfly.home=<wildfly-install-dir> wildfly-extension:deploy
Since this is nothing more than a normal subsystem like all the other subsystems in WildFly, you can use the JBoss CLI to look at its configuration. To poke around, run the JBoss CLI in GUI mode ("jboss-cli.sh --gui") and look for the "org.rhq.msg.broker" subsystem.
In Closing
This is a prototype and was developed just to see how a messaging API around JMS and ActiveMQ can be made simpler and easier to use. I am sure I am missing some pieces of functionality (one such missing piece is security - this currently doesn't handle logging into a secured broker). If you see anything missing, let me know. Feel free to suggest corrections or enhancements. All the code is in github, so play around with it and see how useful it is.
