Category Archives: Flex

Decoupling Asynchronous Messaging With Spring JMS and Spring Integration

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.

Share/Save

JMX MBean Proxying for Export via Remoting to the Client

I wanted to expose mbeans related to application monitoring and services such as a JMS broker, users in session, etc and remote that data to the client which in my case was Flex.  First I checked out using annotations for automated registration and mbean creation but that wouldn’t help me export to a remoting format..without creating my own classes. Something for another post, but what I did was create a package with some classes, which I will explain here. You can certainly deliver this strategy more minimally but I’ve found that to come back and bite me later so I’ve added in some type safety to the proxy creation of the MBeans.

The Basic Components

There are a few important components you need to set this up as a framework

1. An MBeanConnection

2. A type-safe proxy generator

3. An applicable JMX service URL, Port, Hostname

To export mbeans to the client

1. ObjectName that can be connected to if they are services

2. Ports for those services where applicable (i.e. a broker)

3. Interfaces to proxy

I. The Goal: export mbean data by remoting to the client

I created a service which is mirrored as a Flex actionscript class for remoting which encapsulates the calls for various system monitoring data via jmx: This I registered for remoting in the framework of choice, in my case it was Flex. I have a custom annotation that handles this which I don’t include in this snippet.

public class ClientMBeanService implements IClientMBeanService { ..view the code… }

II. The Delegate

With the idea of each component to be exported create a Delegate class that implements the interface dictating the contract that it must implement a method to provide the port, the ObjectName and export the mbean:

public class BrokerDelegate implements IMBeanDelegate { ..view the code… }

All Delegates must implement

public interface IMBeanDelegate { ..view the code… }

III. For Remote and Local MBeanConnections

public class MBeanConnection { ..view the code… }

IV. The Cool Part: Get a Typesafe Proxy Intance of The MBean

public final class MBeanProxyInstance { ..view the code… }

V. Typesafing: I did not originally write this class, only modified it for Java 6, but the crux is:

public class TypeSafeMBeanProxy { ..view the code… }

Share/Save

Enterprise Messaging with JMS, Flex, Spring and ActiveMQ

I finally finished building an enterprise messaging system using Spring, JMS, and ActiveMQ integrated with Flex Messaging from Livecycle Data Services and a clients in both Flex and Java. The Flex-Java(Spring) integration is handled by Anvil for this project.

I plan to re-configure this in a simplified app using BlazeDS, Spring BlazeDS Integration with Spring, ActiveMQ, Flex, JMS and putting the source code here for a test drive in Tomcat, unless SpringSource beats me to the punch ;)

I probably won’t finish this for at least a month though since I’m still tuning the Spring/ActiveMQ configuration for enterprise use cases.

Share/Save

Spring BlazeDS Integration

I went through the reference docs and quickly set up a basic implementation using Livecycle Data Services and one thing quickly occurred to me just in reading the docs:

Here is an easy test drive from Christophe at Adobe.

I hope I am not taking the statement, “the FlexRemotingServiceExporter will be able to work transparently..” out of context but the sample below from the reference docs adds a layer of complexity to spring configuration. Example: I have 100 java services wrapping dao’s etc, configured in Spring, plus their flex handler services to pipe the remoting calls. Wrapping each service in a FlexRemotingServiceExporter adds to configuration mayhem. Could we simply declare the FlexRemotingServiceExporter, annotate the services as @RemoteService or new similar type of annotation that wires it to FlexRemotingServiceExporter as a wrapper, where it does its job allowing dynamic creation of remoting destinations? But by the sample below it would add a lot of overhead to massive enterprise apps such as the one I am currently working on. We are partnering with Adobe on our framework, and I had just proposed to write a Spring Extension for this when 2 weeks later read the press release, so I look forward to seeing this module develop.

From: Spring BlazeDS Integration Reference Guide

Code:
<!-- Expose the productService bean for BlazeDS remoting -->
<bean id="product" class="org.springframework.flex.messaging.remoting.FlexRemotingServiceExporter">
    <property name="messageBroker" ref="mySpringManagedMessageBroker"/>
    <property name="service" ref="productService"/>
</bean>

Would it help to set a parent a bean and save the implementors from adding the wrapper to each?

Code:
<bean name="fxExporter" abstract="true" "class="org.springframework.flex.messaging.remoting.FlexRemotingServiceExporter">
     <property name="messageBroker" ref="mySpringManagedMessageBroker"/>
</bean>

and then set
<bean id="product" parent="fxExporter" class="org.springframework.flex.messaging.remoting.FlexRemotingServiceExporter">
    <property name="service" ref="productService"/>
</bean>

I know this is just the first foundation release so this is just a suggestion (cutting down on configuration as you can already do dynamic destinations for java-flex at present). What i AM looking forward to is the JMS module with Flex Messaging. I’ve got this working now with Flex -Livecycle Data Services – AMQ – JMS, and hope this module will simplify implementations going forward.

Share/Save

SpringSource First Milestone of Spring BlazeDS Released

SpringSource released its first Milestone for Spring and BlazeDS, the free version of Adobe’s LCDS

“This first milestone is very much a foundational release, focusing on support for configuring and bootstrapping the BlazeDS MessageBroker (the central component that handles incoming messages from the Flex client) as a Spring-managed object, routing HTTP messages to it through the Spring DispatcherServlet infrastructure, and easily exporting Spring beans as destinations for direct Flex remoting. Future milestones leading up to the final 1.0 will build upon this foundation to provide deeper features such as Spring Security integration, messaging integration using Spring’s JMS support, an AMFView for use in conjunction with Spring 3.0′s REST support, and hopefully further things to address the needs of our community that we haven’t thought of yet.” – Jeremy Grelle, SpringSource

More info: http://www.springsource.org/node/904

Reference Docs http://static.springframework.org/spring-flex/docs/1.0.x/reference/html/index.html

I’ve been developing framework for messaging with Flex, Livecycle Data Services, Java and JMS and will give this first release a test run

Share/Save