Lingo with Jencks (JCA Container)

When implementing the server side with Lingo you probably want high performance JMS subscription to use a thread pool and to support parallel processing of inbound messages as well as pooling JMS connections, sessions and handling transactions and XA.
You can do this efficiently in JMS by using a JCA Container like Jencks.

How to configure:
<!-- If embedded JMS broker otherwise ignore this tag-->
<bean id="broker" class="org.apache.activemq.xbean.BrokerFactoryBean" singleton="true">
  <property name="config" value="classpath:broker.xml" />
</bean>

<bean id="server" class="org.logicblaze.lingo.jms.JmsServiceExporterMessageListener" depends-on="broker">
  <property name="service" ref="helloImpl" />
  <property name="serviceInterface" value="test.mdp.Hello" />
  <property name="connectionFactory" ref="jmsFactory" />
</bean>

<bean id="helloImpl" class="test.mdp.HelloBean" singleton="true" />

<bean id="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL" value="tcp://localhost:61616" />
</bean>

<bean id="jencks" class="org.jencks.JCAContainer">

  <!-- lets use the default configuration of work manager and transaction manager-->
  <property name="bootstrapContext">
    <bean class="org.jencks.factory.BootstrapContextFactoryBean">
      <property name="threadPoolSize" value="25" />
    </bean>
  </property>

  <!-- the JCA Resource Adapter -->
  <property name="resourceAdapter">
    <bean id="activeMQResourceAdapter" class="org.apache.activemq.ra.ActiveMQResourceAdapter">
      <property name="serverUrl" value="tcp://localhost:61616" />
    </bean>
  </property>
</bean>

<!-- an inbound message connector using a stateless, thread safe MessageListener -->
<bean id="inboundA" class="org.jencks.JCAConnector">

  <property name="jcaContainer" ref="jencks" />

  <!-- subscription details -->
  <property name="activationSpec">
    <bean class="org.apache.activemq.ra.ActiveMQActivationSpec">
      <property name="destination" value="MyQueue" />
      <property name="destinationType" value="javax.jms.Queue" />
    </bean>
  </property>

  <property name="ref" value="server" />
</bean>

<bean id="client" class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">
  <property name="serviceInterface" value="test.mdp.Hello" />
  <property name="connectionFactory" ref="jmsFactory" />
  <property name="destination" ref="exampleDestination" />

  <!-- lets add some custom headers to the messages -->
  <property name="messageProperties">
    <map>
      <entry key="MyHeader">
        <value>Hello World</value>
      </entry>
    </map>
  </property>
</bean>

<bean id="exampleDestination" class="org.apache.activemq.command.ActiveMQQueue">
  <constructor-arg index="0" value="MyQueue" />
</bean>

Jencks Container Approach to MDP

Jencks is logically equivalent to the ActiveMQ JCA Container and migrating to it is pretty simple. The benefits of Jencks are that it supports full XA recovery and works well with Geronimo's TransactionManager and WorkManager.
So it is Spring and POJO enhancements to the Geronimo JCA container to implement Message Driven POJOs.

The JCA (Java Connector Architecture) provides the most efficient way of thread pooling, resource pooling, transaction handling and consumption on JMS and other Resource Adapters. So, if you are consuming JMS messages, the JCA container provides the most efficient way of pooling the JMS sessions and connections, pooling threads and processing messages in parallel as well as transaction and exception handling.

You can download the latest release of Jencks container from http://docs.codehaus.org/display/JCA/Download

So here are the steps to proceed:
1. Create a spring configuration file (service.xml) and insert the following lines:
<bean id="jencks" class="org.jencks.JCAContainer">
  <property name="transactionManager" ref="transactionManager"/>
  <property name="threadPoolSize" value="25"/>

  <property name="resourceAdapter">
    <bean id="activeMQResourceAdapter" class="org.apache.activemq.ra.ActiveMQResourceAdapter">
      <property name="serverUrl" value="tcp://localhost:51616"/>
    </bean>
  </property>
</bean>

The above lines configure the JCA container. You can configure the thread pool size, transaction manager here.
You can also associate whichever JCA Resource Adapter you wish, such as to use another JMS provider. Since I am going to use ActiveMQ, have used ActiveMQResourceAdapter.

Insert below lines in service.xml to register Transaction Manager:
<bean id="transactionManager" class="org.jencks.factory.TransactionManagerFactoryBean"/>

Now you have to define ActiveMQ broker using
<bean id="broker" class="org.apache.activemq.xbean.BrokerFactoryBean">
  <property name="config" value="file:///c:/activemq/conf/mybroker.xml" />
</bean>

2. Now register your MDP with this container. Insert the following in service.xml
<bean id="connA" class="org.jencks.JCAConnector">
  <property name="jcaContainer" ref="jencks" />
  <property name="activationSpec">
    <bean class="org.apache.activemq.ra.ActiveMQActivationSpec">
      <property name="destination" value="MyQueue"/>
      <property name="destinationType" value="javax.jms.Queue"/>
    </bean>
  </property>
  <property name="ref" value="HelloBean"/>
</bean>

<bean id="HelloBean" class="test.mdp.HelloBean" singleton="true"/>

For Topics, just change the ActiveMQActivationSpec.
<bean class="org.apache.activemq.ra.ActiveMQActivationSpec">
  <property name="destination" value="MyTopiv"/>
  <property name="destinationType" value="javax.jms.Topic"/>
</bean>

The above lines of configuration create stateful, pooled MessageListener.

The activation specification is dependent on the JCA Resource Adapter being used. If you are using ActiveMQ as your resource adapter then you can configure are all the various properties here, giving you full control over every aspect of the JMS subscription, pooling and delivery mode. For ActiveMQ adapter you can find the properties available at http://activemq.apache.org/activation-spec-properties.html

The POJO which implements the JMS MessageListener interface is referenced via the ref property in the connector. The HelloBean in this example can be any POJO created by Spring. If you wish you can use a spring singleton, or can add your own Spring interceptors around it - though note that Jencks already takes care of security and transaction handling for you automatically.

Note: If you are using XA transactions on inbound messages and want to support recovery then you must specify an id or name in the Spring XML configuration for the inbound connector as shown above.

3. Required JARs are:


4. Some key points about Jencks Container:
a. concurrency of message processing: One of the main points of using JCA is to allow you to process inbound messages concurrently in a thread pool using a pool of JMS connections and sessions. There are 2 main configuration points you can use in Jencks

--the thread pool size defines the maximum number of concurrent messages that can be processed in Jencks across any subscription in that JCAContainer.

--for some ActivationSpec implementations you can further configure the subscription to define the concurrency model. For example in ActiveMQ you can specify the maxSessions and maxMessagesPerSessions properties to configure how many concurrent messages can be processed for that subscription.

Using both approaches you can define how many threads are used for all the subscriptions and also how many of those threads can be used up by each subscription. This means you can segment your subscriptions to ensure one subscription doesn't overload other subscriptions.

You can also create multiple JCAContainer instances to allow you to have completely separate thread pools for different kinds of subscriptions if you wish.

b: To use XA for an inbound connection in the JCA container you'll need to specify a transaction manager to the inbound connector. Here's an example

<bean id="inboundConnectorA" class="org.jencks.JCAConnector">

  <property name="jcaContainer" ref="jencks"/>
  <property name="activationSpec">
    <bean class="org.apache.activemq.ra.ActiveMQActivationSpec">
      <property name="destination" value="test.spring.inboundConnectorA"/>
      <property name="destinationType" value="javax.jms.Topic"/>
    </bean>
  </property>

  <!-- use XA transactions -->
  <property name="transactionManager" ref="transactionManager"/>

  <property name="ref" value="HelloBean"/>
</bean>
You also need to provide a transaction manager implementation to use.

<bean id="transactionManager" class="org.jencks.factory.TransactionManagerFactoryBean"/>

Outbound Messaging:
Jencks not only supports Inbound JMS messaging but also outbound messaging.

Outbound messaging allows you to pool and reuse JMS sessions and connections when sending messages. In addition JCA takes care of associating inbound and outbound messaging with the same local transaction or XA.

This is very useful if you are inside either a Servlet or EJB type environemnt where you want to send messages from different threads, but don't want to create & close a connection, session, producer for each send.

JMS is designed for high performance message exchange. It makes heavy use of asynchronous message exchange to boost performance compared to synchronous request/response type systems.However creating and closing connections, sessions, producers is an expensive operation - since each operation will often require a synchronous request/response with the message broker (e.g. for security checks etc). So where possible JMS resources like connections, sessions, producers, consumers should be setup once on startup and reused/pooled. Then a send or receive is a blazingly fast operation.

<beans>
  <!-- ActiveMQ -->

  <bean id="broker" class="org.apache.activemq.xbean.BrokerFactoryBean">
    <property name="config" value="classpath:org/jencks/samples/outbound/broker.xml"/>
  </bean>

  <bean id="jmsResourceAdapter" class="org.apache.activemq.ra.ActiveMQResourceAdapter">
    <property name="serverUrl" value="tcp://localhost:51616"/>
  </bean>

  <bean id="jmsQueue" class="org.springframework.jndi.JndiObjectFactoryBean">
  <property name="jndiName" value="queue"/>
  <property name="jndiEnvironment">
    <props>
      <prop key="java.naming.factory.initial">org.apache.activemq.jndi.ActiveMQInitialContextFactory</prop>
      <prop key="java.naming.provider.url">tcp://localhost:51616</prop>
      <prop key="queue.queue">MyQueue</prop>
    </props>
  </property>
  </bean>

  <!-- Transaction manager-->

  <bean id="transactionManager" class="org.jencks.factory.TransactionManagerFactoryBean"/>

  <!-- Connection Manager -->

  <bean id="connectionManager" class="org.jencks.factory.ConnectionManagerFactoryBean">
    <property name="transactionManager" ref="transactionManager"/>
  </bean>

  <!-- JMS Managed Connection Factory -->

  <bean id="jmsManagedConnectionFactory" class="org.apache.activemq.ra.ActiveMQManagedConnectionFactory">
    <property name="resourceAdapter" ref="jmsResourceAdapter"/>
  </bean>

  <!-- JMS Connection Factory -->

  <bean id="jmsConnectionFactory" class="org.jencks.factory.ConnectionFactoryFactoryBean">
    <property name="managedConnectionFactory" ref="jmsManagedConnectionFactory"/>
    <property name="connectionManager" ref="connectionManager"/>
  </bean>

</beans>

Important to nate that the inbound and outbound config MUST match up.

Lingo approach to MDP

Lingo is a POJO based remoting and messaging framework. So the steps are:
1. Create a class (POJO) which will act as a MDP.

package test.mdp;
public interface Hello {
public void greet(String name);
}

package test.mdp;
public class HelloImpl implements Hello {
public void greet(String name) {
System.out.println("Hello " + name);
}
}

Here we have created an interface Hello which is implemented by HelloImpl. The point to notice here is that out POJO HelloImpl does not implement any external listener.

2. Convert your POJO to MDP:
To do this we are using Lingo's JMS exporter. Spring supports exporter.
Create a spring context file service.xml. Add the following contents:

<bean id="server" class="org.logicblaze.lingo.jms.JmsServiceExporter">
<property name="service">
<bean class="test.mdp.HelloImpl"/>
</property>
<property name="serviceInterface" value="test.mdp.Hello"/>
<property name="connectionFactory">
<ref bean="connectionFactory" />
</property>
<property name="destination">
<ref bean="destination" />
</property>
</bean>

Assuming that ActiveMQ will be running on the localhost, listening on port 61616

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
</bean>

<bean id="destination" class="org.apache.activemq.message.ActiveMQQueue">
<constructor-arg index="0" value="MyQueue" />
</bean>

3. Start the MDP:

public class StartHello {
public static void main(String[] args) {
new FileSystemXmlApplicationContext("service.xml");
}
}

Make sure ActiveMQ is running, and run StartHello. Your MDP will be added as a new client to ActiveMQ and will be ready to service requests.

4. Send a message:

Create a spring context say client.xml. Add the following contents:
<bean id="client" class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">
<property name="serviceInterface" value="test.mdp.Hello"/>
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="destination" />
</bean>

<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
</bean>

<bean id="destination" class="org.apache.activemq.message.ActiveMQQueue">
<constructor-arg index="0" value="MyQueue" />
</bean>

Now create your HelloClient,

public class MDPojoClient {
public static void main(String[] args) {
ApplicationContext ctx = new FileSystemXmlApplicationContext("client.xml");
Hello client = (Hello)ctx.getBean("client");
client.greet("Tommy");
}
}
Here greet() is called synchronously.

4.b. both server and client can be merged into one as
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">

<beans>

<!-- client side proxy-->
<bean id="client" class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">
<property name="serviceInterface" value="test.mdp.Hello"/>
<property name="connectionFactory" ref="jmsFactory"/>
<property name="destination" ref="destination"/>
</bean>

<!-- the server side -->
<bean id="server" class="org.logicblaze.lingo.jms.JmsServiceExporter">
<property name="service" ref="helloImpl"/>
<property name="serviceInterface" value="test.mdp.Hello"/>
<property name="connectionFactory" ref="jmsFactory"/>
<property name="destination" ref="destination"/>
</bean>

<!-- the actual implementation of the service - which is only made public for testing purposes -->
<bean id="helloImpl" class="test.mdp.HelloImpl" singleton="true"/>


<!-- JMS ConnectionFactory to use -->
<bean id="jmsFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>

<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="MyQueue"/>
</bean>
</beans>


5. Required Jars:

spring.jar
log4j-.jar
commons-logging-.jar
activemq-core-.jar
concurrent-.jar
geronimo-spec-jms-.jar
geronimo-spec-j2ee-management-.jar
lingo-1.1.jar (http://logicblaze.com)

6. How to make our MDP asynchronous:

Lingo decides whether a method is synchronous or asynchronous by consulting a Metadata Strategy. The default metadata strategy is SimpleMetadataStrategy. By default, SimpleMetadataStrategy assumes that all calls are synchronous. But by configuring LingoRemoteInvocationFactory with your JmsProxyFactoryBean, you can change the behavior. You just need to register LingoRemoteInvocationFactory bean in client.xml.

<bean id="invocationFactory" class="org.logicblaze.lingo.LingoRemoteInvocationFactory">
<constructor-arg>
<bean class="org.logicblaze.lingo.SimpleMetadataStrategy">
<constructor-arg value="true" />
</bean>
</constructor-arg>
</bean>

Here Spring will instantiate LingoRemoteInvocationFactory using a single argument constructor, passing a SimpleMetadataStrategy instance. The SimpleMetadataStrategy itself is declared as an inner-bean. Its "oneWayForVoidMethods" property is set to "true" to indicate that void methods are one-way and should not block.

Now change you client.xml, change the following

<bean id="client" class="org.logicblaze.lingo.jms.JmsProxyFactoryBean">
<property name="serviceInterface" value="com.habuma.mdpojo.lingo.HelloService"/>
<property name="connectionFactory" ref="connectionFactory" />
<property name="destination" ref="destination" />
<property name="remoteInvocationFactory" ref="invocationFactory" />
</bean>

To test its working, you can put a Thread.sleep(10000) in the greet() method.

Note:
1. If the method being invoked returns something, the client must wait for the return value.So it wont be asynchronous.
2. Lingo will work fine with Websphere MQ.

Lingo with WebSphere MQ:
1. Removed ActiveMQ and Geronimo jars, and add following jars from Websphere MQ
com.ibm.mq.jar
com.ibm.mqbind.jar
com.ibm.mqjms.jar
com.ibm.mqpcf.jar
mqcontext.jar
fscontext.jar
jms.jar
jndi.jar
providerutil.jar

2. Change your service.xml and client.xml for below
<bean id="connectionFactory" class="com.ibm.mq.jms.MQQueueConnectionFactory">
</bean>

<bean id="Queue" class="com.ibm.mq.jms.MQQueue">
<constructor-arg index="0" value="queue://QMGR/QUEUE" />
</bean>

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate102">
<property name="connectionFactory">
<ref local="connectionFactory"/>
</property>
</bean>

OR if you want to use JNDI the use the below tags

<bean id="jndiTemplate" class="org.springframework.jndi.JndiTemplate">
<property name="environment">
<props>
<prop key="java.naming.factory.initial">
com.ibm.mq.jms.context.WMQInitialContextFactory
</prop>
<prop key="java.naming.provider.url">
localhost:1414/SYSTEM.DEF.SVRCONN
</prop>
</props>
</property>
</bean>

<bean id="connectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiTemplate">
<ref bean="jndiTemplate"/>
</property>
<property name="jndiName">
<value>MyCFJNDI
</property>
</bean>

<bean id="destination" class="org.springframework.jndi.JndiObjectFactoryBean">
<property name="jndiTemplate">
<ref bean="jndiTemplate"/>
</property>
<property name="jndiName">
<value>MyQueueJNDI</value>
</property>
</bean>