More details here.

Release posts are about all I have time for these days…been heads down for a month getting an amqp/rabbitmq thing ready for release.

I’ve been a linux developer for years so I was all thumbs having to look up everything when setting up a new dev environment on a mac. Here’s some tips for setting up a local RabbitMQ server to play with (i.e. not using macports locally, not the EC2 install etc):

Install and build Erlang

First, Check for the latest Erlang GA Release and note the version.

1. cd to the dir you wish to install into
2. wget http://erlang.org/download/otp_src_{version}.tar.gz
3. tar xzvf otp_src_{version}.tar.gz
4. cd otp_src_{version}
5. NOTE: run ./configure –help to see what opts you want to run with
6. ./configure
7. make
8. sudo make install
9. cd ../

Install RabbitMQ Server

First check the latest version of RabbitMQ Server and note the filename.

1. wget http://www.rabbitmq.com/releases/rabbitmq-server/v1.7.2/rabbitmq-server-generic-unix-1.7.2.tar.gz
2. tar xzvf rabbitmq-server-generic-unix-1.7.2.tar.gz
3. Complete the install and config from http://www.rabbitmq.com/install.html#generic-unix, making sure the logs, clustering, data and config dirs are set up
4. Start RabbitMQ: run rabbit../sbin/rabbitmq-server

If you need a few things first

1. First, if you don’t have it, download Apple XCode so you can compile Erlang with gcc. FYI the XCode download could not take longer, start it and go build an enterprise app in ROO, deploy it to the cloud, and when you’re done it may have completed. If I knew more about it I’d just download the particular compiler needed which is some form of C compiler. If I find out I’ll update this for a simpler install. if you know, send me a tweet: @hollyedelson
2. While that’s downloading, if you don’t have it, get the latest version of wget from http://ftp.gnu.org/pub/gnu/wget
3. On the command line:
   1. tar -xzf wget-{version}.tar.gz
   2. cd wget{version}
   3. ./configure
   4. make
   5. sudo make install
4. Now run the steps from Version A above.

Find Out More

rabbitmq.com erlang.org

If I had more time I would add a few new posts on esper and rabbitmq for messaging in the cloud but I just got home from 2 weeks of coast-to-coast consulting and technology presentations. Pretty exhausting. I hope to have time to post on these and a few other fun technologies. I’ve got some code samples and tests I might post with that, plus some cools stuff on esper queries with Spring Integration.

Pure Master Slave Simplified Topology

I am not actually going to go into Broker topologies, there are many great resources for that such as this by Bruce Snyder: http://www.slideshare.net/bruce.snyder/messaging-with-activemq-presentation or http://activemq.apache.org/topologies.html, all great stuff. This example uses a store and forward topology, or, distributed queues, and incorporates basic authentication: My use case was to handle down JMS Servers. What I needed to do was implement failover as well as master slave strategies and a topology for message redundancy in case of hardware failure, etc. The client could not have any message loss. With failover, you can see how ActiveMQ switches from the main broker to the second, third, etc on failure. I have a case of four JMS servers in production, each server it is on is load balanced. There are just a few configurations to add or modify in order to set up JMS Failover with Master/Slave for your broker topology. Here is a basic configuration. For this use case, all JMS servers are configured as standalone versus embedded.

I. Client URI

You will need to add the Failover protocol, either with a basic URI pattern or a composite. In this use case, there are load balanced servers in Production and multiple Development and QA environments which require different configurations for master/slave and failover.

In your application’s properties file for messaging add a modified version of this with your mappings:
activemq.broker.uri=failover://(tcp://localhost:61616,tcp://slaveh2:61616,tcp://master2:61616,tcp://slave2:61616,network:static://(tcp://localhost:61616,tcp://master2:61616,tcp://slave2:61616))?randomize=false

Note: I set connections as locked down (static) communication configurations vs multicast or dynamic discovery so that I know exactly what servers can communicate with each other and how. Also this is assuming you have one set per environment to account for mapping the appropriate IP’s in development, qa, production, dr, etc.

Note: Do not configure networkConnections for master slave, they are handled on the slave with the following configuration:
<masterConnector remoteURI= "tcp://masterhost:61616" userName="wooty" password="woo"/>

II. Spring Configuration

<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
destroy-method="stop">
<property name="connectionFactory">
<bean class="org.apache.activemq.spring.ActiveMQConnectionFactory">
<constructor-arg value="${amq.broker.uri}"/>
<property name="userName" value="${activemq.username}"/>
<property name="password" value="${activemq.password}"/>
</bean>
</property
</bean>

III. Broker Configuration

Master

<broker brokerName="{hostname}" waitForSlave="true" xmlns="http://activemq.apache.org/schema/core" dataDirectory="${activemq.base}/data">
<networkConnectors>
<!-- passed in by the client broker URI so you can easily manager per environment: sweet -->
</networkConnectors>
<transportConnectors>
<!-- TCP uses the OpenWire marshaling protocol to convert messages to stream of bytes (and back) -->
<transportConnector name="tcp" uri="tcp://localhost:61616?trace=true" />
<transportConnector name="nio" uri="nio://localhost:61618?trace=true" />
<!-- <transportConnector name="ssl" uri="ssl://localhost:61617"/>
<transportConnector name="http" uri="http://localhost:61613"/
<transportConnector name="https" uri="https://localhost:61222"/> -->
<transportConnectors>
</transportConnectors>

<!-- Basic security and credentials -->
<plugins>
<simpleAuthenticationPlugin>
<users>
<authenticationUser username="system" password="manager" groups="admin, publishers,consumers"/>
</users>
</simpleAuthenticationPlugin>
</plugins>
..more configuration
</broker>

Slave: for ActiveMQ 4.1 or later which also allows for authentication as show below

<broker brokerName="{hostname}Slave" deleteAllMessagesOnStartup="true" xmlns="http://activemq.apache.org/schema/core">
<transportConnectors>
<transportConnector uri="tcp://localhost:61616"/>
</transportConnectors>
<services>
<masterConnector remoteURI= "tcp://masterhost:62001" userName="wooty" password="woo"/
</services>
<!-- Basic security and credentials -->
<plugins>
<simpleAuthenticationPlugin>
<users>
<authenticationUser username="system" password="manager" groups="admin, publishers,consumers"/>
</users>
</simpleAuthenticationPlugin>
</plugins>
</broker>

The importance of decoupling in applications is vital but it is not easy to do it well, I am constantly working to improve my strategies. Even more important is the role of Messaging in an enterprise context and in the design. I think Message-driven architecture is perhaps the most important to integrate into applications in terms of its scope of applicability and how it lends to scalability. In this vein, Spring Integration becomes a highly intriguing element worthy of study, testing, and integration. I barely touch on Spring Integration here, particulary in its standard usage but simply use a few classes programmatically to decouple JMS implementation from its clients.

Messaging and Decoupling

Messaging is everywhere and so subtle we are not aware of it, it just happens all around us, all the time. Consider this: in Genetics on a cellular level, which is nongranular in the scope of genetics itself, the major elements in a cell communicate, but how? They are not connected by any physical construct save the mutual environment they are in so how do they do it? They message each other with signals, receptors and a means to translate those messages, which contain instructions.

In Gene expression, DNA/RNA within eukaryote cells (think systems within systems within systems…elements moving in space and time under specific, ever changing constraints and environmental fluctuations (put that in your Agile timebox!) ) communicate by transmitting messages, intercepting, translating and even performing message amplification. There are specialized elements, mRNA specifically, Messenger RNA, which are translated by signal recognition particles… Cool, right? But this happens all around us, outside, in space, everywhere. And it is all decoupled, and therein lies the beauty of messaging.

So what about our applications? Here is one very simple, isolated application of using Spring Integration ( a low-level usage, not very sophisticated ) to decouple your client and server messaging code:

So I wrote a little java package that integrates Spring JMS, ActiveMQ, Spring Integration and Flex Messaging for the front end which hooks into either BlazeDS or Livecycle Data Services. I had a bunch of constraints to solve for such as all Destinations had to come from the database, there were lifecycle issues as far as timing of element initializations with the IoC bean creation and Flex Messaging elements being created and having what they needed such as the Flex Messaging adaptors which I resolved by flex java bootstrapping. In another post I will go into the JMS package further. For the topic here let’s focus on the JMS-Spring JMS-Spring Integration bridge. The image to the left shows the layout of my jms package to facilitate the mapping. In this system, messages come in from 2 areas: the client and java services on the server. Complicated systems will have many more but let’s talk about the 2 that most would have.  The client sends messages to the server that are both user messages and operational messages by the system. Java services send messages when certain business rules and criteria are triggered, passing business data and any message-aware object could be sending messages. Sending on the server To insure proper usage I created an interface that a service must implement to send to ActiveMQ, called JMSClientSupport. Note all code in this post is simplified. Here, I actually have it returning a validation message if errors occurred so that a business service developer could implement handling per requirements. A Business Entity public class Foo implements Serializable {...} A Service public class FooServiceImpl implements BusinessService, FooService, SMTPClientSupport, JMSClientSupport { public void insert(Foo foo) { ..//do some important business stuff publish(foo); } public void publish(Object object) { if ((someBusinessValidation((Foo) object)) { jmsService.send(destinationId, object); } } } public interface JMSClientSupport { void publish (Object object); }

Sending from the Client

You could have any client sending messages of any nature to the server. In this case I am using Flex. Messages of type <T> are wrapped on the client as an IMessage {AsyncMessage,CommandMessage etc}. When these messages make their way through Blaze or Livecycle, I have it wired to hit this java adapter which is represented in a Hash per FlexDestination for 1:1 FlexDestination : JMS Destination by Flex.

For this example I am implementing the JMS MessageListener to show tight coupling as well as decoupling:

public class FlexMessagingAdapter extends MessagingAdapter implements MessageListener {
// Invoked by Flex when a message comes in from the client to this adapter's Destination
public Object invoke(Message message) {
// a custom interceptor that extracts partition Destination info like ActiveMQ message group or subtopics like STOCKS.NASDAQ for more specific routing
String partition = new DestinationPartitionInterceptor(message).intercept();
jmsService.send(destination, new IntegrationMessageCreator (message, partition));
}
return null;
}

// Decoupled: Invoked when a Message is received from the Spring Integration channel
public void handleMessage(org.springframework.integration.core.Message<?> message) {....}

// Sets the Spring Integration MessageChannel for sending and receiving messages
public void setMessageChannel(MessageChannel messageChannel) {
this.messageChannel = messageChannel;
}

// Tightly coupled with JMS by the MessageListener.onMessage() method
public void onMessage(javax.jms.Message jmsMessage) {
flex.messaging.messages.Message message = new IntegrationMessageCreator(jmsMessage).createMessage(getDestination().getId(), getSubscribers());
if (getMessageService().getMessageBroker().getChannelIds().contains("streaming-amf")) {
MessageBroker broker = MessageBroker.getMessageBroker(null);
broker.routeMessageToService(message, null);
} else {
getMessageService().pushMessageToClients(message, true);
}
}}

The Transformer: Where Messages Intersect

I have a second post that show an even more decoupled messaging strategy with Spring Integration but this is purely a a basic idea using Flex Messaging, Spring Integration, Spring JMS and ActiveMQ. I will post the more broad strategy next

Step 1: Client messages are transformed here by extending the JMS MessageCreator. In this class I pull out the data from any Object type but specifically Flex Message and a JMSMessage types.

public class IntegrationMessageCreator implements MessageCreator {
// a few constructors here to handle multiple message types: JMSMessage, Flex Message, Object message, etc

private MessageBuilder createBuilder() {
MessageBuilder builder = null;
if (this.object != null) {
builder = MessageBuilder.withPayload(object);
} else if (this.flexMessage != null && flexMessage.getBody() != null) {
builder = MessageBuilder.withPayload(flexMessage.getBody()).copyHeaders(flexMessage.getHeaders());
}
// ActiveMQ Message Groups
if (this.partition != null) builder.setHeader(MessageConstants.Headers.JMSXGROUPID, partition);

return builder;
}

// to JMS
public javax.jms.Message createMessage(Session session) throws JMSException {
return new IntegrationMessageConverter().toMessage(createBuilder().build(), session);
}

// To Flex
public flex.messaging.messages.Message createMessage(String destinationId, int subscribers) {
Message integrationMessage = (Message) new IntegrationMessageConverter().fromMessage(this.jmsMessage);

flex.messaging.messages.Message flexMessage = new AsyncMessage();
flexMessage.setBody(integrationMessage.getPayload());
flexMessage.setDestination(destinationId);
flexMessage.setHeaders(integrationMessage.getHeaders());
// …and other good jms to flex data

return flexMessage;
}
}

The Converter


import org.springframework.integration.jms.HeaderMappingMessageConverter;
import org.springframework.integration.core.Message;
import javax.jms.Session;

public class IntegrationMessageConverter extends HeaderMappingMessageConverter {

// Converts from a JMS Message to an Integration Message. You should do a try catch but I cut it out for brevity
public Object fromMessage(javax.jms.Message jmsMessage) throws Exception {
return (Message) super.fromMessage(jmsMessage);
}

// Converts from an Integration Message to a JMS Message. You should do a try catch but I cut it out for brevity
public javax.jms.Message toMessage(Object object, Session session) throws Exception {
return jmsMessage = super.toMessage(object, session);
}
}

JMS Asynchronous Reception

In my jmsConfig.xml I configured one Spring MessageListenerAdapter which I have wired with a message delegate, a POJO and its overloaded handleMessage method name:

<bean id=”messageListenerAdapter”>
<property name=”delegate” ref=”defaultMessageDelegate”/>
<property name=”defaultListenerMethod” value=”handleMessage”/>
<property name=”messageConverter” ref=”simpleMessageConverter”/>
</bean>

As the application loads and all Spring beans are initialized, I initialize all of my JMS Destinations. As I do this, I also initialize a MessageListenerAdapter for each Destination. I have a stateless JMSService, which is called by another service, MessagingGateway, to initialize each Destination and which calls PollingListenerContainerFactory to create child MessageListenerAdaptors for each Destination. The adapters are configured based on an abstract parent configuration:

<bean id=”abstractListenerContainer” abstract=”true” destroy-method=”destroy”>
<property name=”connectionFactory” ref=”pooledConnectionFactory”/>
<property name=”transactionManager” ref=”jmsTransActionManager”/>
<property name=”cacheLevel” value=”3″/>
<property name=”taskExecutor” ref=”taskExecutor”/>
<property name=”autoStartup” value=”true”/>
</bean>

Snippet from PollingListenerContainerFactory:

/**
* Gets the parent from the IoC to reduce runtime config and
* resources to create children.
* <p/>
* DefaultMessageListenerContainer is Responsible for all threading
* of message reception and dispatches into the listener for processing.
* Supports dynamic scaling for a higher during peakloads.
*
* @param destination
* @param messageListener
* @return
*/
public static DefaultMessageListenerContainer createMessageListenerContainer(Destination destination, MessageListener messageListener) {

ChildBeanDefinition childBeanDefinition = new ChildBeanDefinition(“abstractListenerContainer”, configureListenerContainer(destination, messageListener));

String beanID = IdGeneratorUtil.getStringId();
ConfigurableListableBeanFactory beanFactory = ApplicationContextAware.getConfigurableListableBeanFactory();
((DefaultListableBeanFactory) beanFactory).registerBeanDefinition(beanID, childBeanDefinition);

DefaultMessageListenerContainer container = (DefaultMessageListenerContainer) ApplicationContextAware.getBean(beanID);
container.setDestination(destination);
return container;
}

/**
* Configures the child listener, based on the parent in jmsConfig.xml.
* <p>Configures Queue or Topic consumers:
* Queue: stick with 1 consumer for low-volume queues: default is
* Topic: there’s no need for more than one concurrent consumer
* Durable Subscription: Only 1 concurrent consumer supported
* <p/>
* props.addPropertyValue(“messageSelector”, “”); sets a message selector for this listener
*
* @param destination
* @param messageListener
* @return
*/
private static MutablePropertyValues configureListenerContainer(Destination destination, MessageListener messageListener) {
MutablePropertyValues props = new MutablePropertyValues();
props.addPropertyValue(“destination”, destination);
props.addPropertyValue(“messageListener”, messageListener);

// Enable throttling on peak loads
if (destination instanceof Queue) {
props.addPropertyValue(“maxConcurrentConsumers”, 50); // modify to needs
}
// Override default setting Point-to-Point (Queues)
if (destination instanceof Topic) {
props.addPropertyValue(“pubSubDomain”, true);
}

return props;
}

this is overkill to this topic but its cool stuff. So we now have a JMS listener for asynchronous JMS reception as well as flex, but now let’s look at the message delegate we wired into the MessageListenerAdapter:

public interface MessageDelegate {

void handleMessage(String message);

void handleMessage(Map message);

void handleMessage(Serializable message);
}

Pretty simple, right? It’s a POJO with absolutely no JMS code whatsoever for asynchronous message reception. How does it work? Spring abstracts the JMS code and calls it behind the scenes, if you look at the source code for Spring’s SimpleMessageConverter, it does the fromMessage() toMessage() handling for you and throws the resultant “message” into the appropriate overloaded method above. Now this is great for simple message abstraction but the above with JMS-Flex and Spring Integration is an example of more complicated handling. With clients you often need to translate and transfer the data from message type 1 to message type 2. In the adapter code above, you would use the handleMessage() method to get the message from Spring Integration and into the message type of your choice, here, a Flex message.

I’ve just started staging my new project, SuMQ, and appreciate patience as this will take me a while to get the code standardized and on google.

SuMQ is a light-weight enterprise messaging framework built in Java, leveraging Spring, JMS, and ActiveMQ. It plugs into Flex Messaging via BlazeDS for the client. This can also be configured for other clients aside from Flex.

The sample will be ready for clustered BlazeDS instances and load balanced Application Servers.

code.google.com/p/sumq