Implementing Jms to Jms Bridge using ActiveMQ

• Download jms - 1.63 KB

Introduction

A while back, I was assigned to work on a project on which I needed to integrate with another JMS system. As I searched the web for some detailed tutorials, I didn't find any detailed tutorial that could help me. Eventually I figured out all the pieces I had the design and implement in order to get such JMS-to-JMS bridge integration. I decided to summarize the things I learned in this article, not only for my own record keeping purpose, but also helping others who needed similar information. I assumed audiences know the technologies such as spring, and jms.

The intention of this article is to provide a detailed tutorial on the following topics:

• Software required for this article.

• How to publish a Message to queue and consume it.

• How to bridge one Open-JMS to another ActiveMQ-JMS.

I hope readers will enjoy this article! If you have any questions of comments, please leave it in the FAQ section.

Background

ActiveMQ is an open source, Apache 2.0 licensed Message Broker and JMS 1.1 implementation and Enterprise Integration Patterns provider which integrates seamlessly into Geronimo, light weight containers and any Java application. Apache ActiveMQ is fast, supports many Cross Language Clients and Protocols, comes with easy to use Enterprise Integration Patterns and many advanced features while fully supporting JMS 1.1 and J2EE 1.4.

ActiveMQ provides bridging functionality to other JMS providers that implement the JMS 1.0.2 and above specification. A JMS bridge can be co-located with an ActiveMQ broker or run remotely. In order to support JMS 1.0.2 there is separation between Queues and Topics.

Features:

• Supports a variety of Cross Language Clients and Protocols from Java, C, C++, C#, Ruby, Perl, Python, PHP

• OpenWire for high performance clients in Java, C, C++, C#

• Stomp support so that clients can be written easily in C, Ruby, Perl, Python, PHP, ActionScript/Flash, Smalltalk to talk to ActiveMQ as well as any other popular Message Broker

• full support for the Enterprise Integration Patterns both in the JMS client and the Message Broker

• Supports many advanced features such as Message Groups, Virtual Destinations, Wildcards and Composite Destinations

• Fully supports JMS 1.1 and J2EE 1.4 with support for transient, persistent, transactional and XA messaging

• Spring Support so that ActiveMQ can be easily embedded into Spring applications and configured using Spring's XML configuration mechanism

• Tested inside popular J2EE servers such as Geronimo, JBoss 4, GlassFish and WebLogic

• Includes JCA 1.5 resource adaptors for inbound & outbound messaging so that ActiveMQ should auto-deploy in any J2EE 1.4 compliant server

• Supports pluggable transport protocols such as in-VM, TCP, SSL, NIO, UDP, multicast, JGroups and JXTA transports

• Supports very fast persistence using JDBC along with a high performance journal

• Designed for high performance clustering, client-server, peer based communication

• REST API to provide technology agnostic and language neutral web based API to messaging

• Ajax to support web streaming support to web browsers using pure DHTML, allowing web browsers to be part of the messaging fabric

• CXF and Axis Support so that ActiveMQ can be easily dropped into either of these web service stacks to provide reliable messaging

• Can be used as an in memory JMS provider, ideal for unit testing JMS Can be used as an in memory JMS provider, ideal for unit testing JMS

OpenJMS is an open source implementation of Sun Microsystems's Java Message Service API 1.1 Specification

Features:

• Point-to-Point and publish-subscribe messaging models

• Guaranteed delivery of messages

• Synchronous and asynchronous message delivery

• Persistence using JDBC

• Local transactions

• Message filtering using SQL92-like selectors

• Authentication

• Administration GUI

• XML-based configuration files

• In-memory and database garbage collection

• Automatic client disconnection detection

• Applet support

• Integrates with Servlet containers such as Jakarta Tomcat

• Support for TCP, RMI, HTTP and SSL protocol stacks

• Support for large numbers of destinations and subscribers

Install and Configure the Required Software

This tutorial requires setup and configuration of the following software packages:

• JDK 1.5 (or higher).

• Open Jms 0.7.7 download link< http://openjms.sourceforge.net/downloads.html

• Apache ActiveMQ 5.1 download link http://activemq.apache.org/activemq-510-release.html

I also recommend the use of Eclipse IDE. But, for this tutorial, all work can be done using a simple code editor (like UltraEdit32, Crimson Editor, or Notepad), Command Prompt For my own convenience, this tutorial was created on Windows XP. And it should be relative easy to be migrated to other platforms.

<h3install jdk="" <="" h3="">

Install JDK on Windows XP is super easy, just download the MSI installer, and then install it either default to "Program Files" or directly to "C:\". After installing JDK, it is recommended to configure the system variables:

1. Create system environment "JAVA_HOME" and point it to the JDK base directory (i.e. C:\Program Files\Java\jdk-1.5.0_17 or C:\jdk-1.5.0_17).

After configuring the system variable, open a Command Prompt and type "java -version". The output will indicate the version of JDK installed in the system. This should help verify the success of JDK installation. After the verification, close the Command Prompt.

Install OpenJMS and ActiveMQ

Install OpenJMS is also easy, download the binary executable archive file from the Open JMS website ( http://openjms.sourceforge.net/downloads.html). Then unzip openjms-0.7.7-beta-1.zip the archive file to "C:\". This will unpack the archive file to "C:\ ". "C:\ openjms-0.7.7-beta-1" will be the base directory of OpenJMS

Install ActiveMQ is also easy, download the binary executable archive file from the Apache ActiveMQ website (http://activemq.apache.org/activemq-510-release.html). Then unzip apache-activemq-5.1.0-bin.zip the archive file to "C:\". This will unpack the archive file to "C:\ ". "C:\ apache-activemq-5.1.0 " will be the base directory of ActiveMQ

Create a Simple Java class for Publishing a Message

import java.util.Hashtable;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSender;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class SimpleMessageSender {

	private Context jndiContext;
	
	private QueueConnectionFactory factory;
	
	private QueueConnection queueConnection;
	
	private static final String URL = "tcp://localhost:3035";
	
	private static final String CONNECTION_FACTORY = "ConnectionFactory";
	
	private static final String QUEUE_NAME = "jmstojmsBridgeQueue";
	
	/**
	 * Constructor initialize queue connection factory and connection.
	 * @throws NamingException 
	 * @throws JMSException 
	 */
	public SimpleMessageSender() throws NamingException, JMSException
	{
		Hashtable environment = new Hashtable();
		environment.put(Context.INITIAL_CONTEXT_FACTORY, org.exolab.jms.jndi.InitialContextFactory.class.getName());
		environment.put(Context.PROVIDER_URL, URL);
		jndiContext = new InitialContext(environment);
		factory = (QueueConnectionFactory)jndiContext.lookup(CONNECTION_FACTORY);
		queueConnection = factory.createQueueConnection();
	}
	/**
	 * @param message
	 */
	public void sendMessage(String message){
		 
		try {
		
			QueueSession queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
			Queue queue = queueSession.createQueue(QUEUE_NAME);
			QueueSender sender = queueSession.createSender(queue);
			TextMessage txtMessage = queueSession.createTextMessage("OPEN_JMS :"+message);
			sender.send(txtMessage);
			System.out.println("Message has been send successfully");
			close();
		} catch (JMSException e) {
			e.printStackTrace();
		} catch (NamingException e) {
			
			e.printStackTrace();
		}
	}
	private void close() throws JMSException, NamingException
	{
	   if(queueConnection!=null){
		   queueConnection.close();
	   }
	   if(jndiContext!=null){
		   jndiContext.close();
	   }
	   factory=null;
	}
	
	
	public static void main(String args[]) throws Exception{
		if (args.length<1){
			System.out.println("Usage : java SimpleMessageSender <message>");
			return;
		}
		SimpleMessageSender simpleMsgSender = new SimpleMessageSender();
		simpleMsgSender.sendMessage(args[0]);
		
	}
}

Create a Simple Java class for Consuming a Message

<p>
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.QueueConnection;
import javax.jms.QueueReceiver;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;

public class SimpleMessageListener  implements MessageListener{
   
	private static final String QUEUE_NAME = "jmstojmsBridgeQueue";
	
	public void onMessage(Message message) {
		
		if (message!=null && message instanceof TextMessage){
			TextMessage txtMessage = (TextMessage)message;
			try {
				System.out.println("Message :: "+txtMessage.getText());
			} catch (JMSException e) {
				
				e.printStackTrace();
			}
		}
	}

 public SimpleMessageListener() throws JMSException {
	 
	 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
	 QueueConnection queueConnection = factory.createQueueConnection();
	 queueConnection.start();
	 QueueSession queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
	 Queue queue  = queueSession.createQueue(QUEUE_NAME);
	 QueueReceiver receiver = queueSession.createReceiver(queue);
	 receiver.setMessageListener(this); 
	
 }
 public static void main(String args[])throws Exception
 {
	 new SimpleMessageListener();
	 System.out.println("Waiting for Message");
 }
}

Before compile the code you have to set the required jar files to classpath

c:\jms>set CLASSPATH=.;<OPEN_JMS_INSTALL_DIR>\lib\jms-1.1.jar; <OPEN_JMS_INSTALL_DIR>\lib\openjms-0.7.7-beta-1.jar;%CLASSPATH%

c:\jms>javac SimpleMessageSender.java

D:\jms>set CLASSPATH=<ACTIVEMQ_INSTALL_DIR>\activemq-all-5.1.0.jar;%CLASSPATH%

c:\jms>javac SimpleMessageListener.java

Adding the JMS Bridge Connector

Open the <ACTIVEMQ_INSTALLED_DIR>/conf/activmq.xml file.

Add the following code after the </transportConnectors> element.

<p><jmsBridgeConnectors>
        <jmsQueueConnector name="OpenJMSBridge-Inbound"
           jndiOutboundTemplate="#remoteJndi"
           outboundQueueConnectionFactoryName="ConnectionFactory"
           localQueueConnectionFactory="#localFactory">
          <inboundQueueBridges>
            <inboundQueueBridge inboundQueueName="jmstojmsBridgeQueue"/>
          </inboundQueueBridges>
          </jmsQueueConnector>
</jmsBridgeConnectors>

Add the following code after the  </broker> element

 <bean id="remoteJndi" class="org.springframework.jndi.JndiTemplate">
      <property name="environment">
        <props>
          <prop     key="java.naming.factory.initial">org.exolab.jms.jndi.InitialContextFactory</prop>
          <prop key="java.naming.provider.url">tcp://localhost:3035</prop>
        </props>
      </property>
    </bean>
  <bean id="localFactory"
    class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://localhost:61616" />
  </bean>

Open the <OPENJMS_INSTALLED_DIR>/config/Openjms.xml file.

Comment it out the following code.

<p>

<!-- Connector scheme="rmi">
      <ConnectionFactories>
        <QueueConnectionFactory name="JmsQueueConnectionFactory" />
        <TopicConnectionFactory name="JmsTopicConnectionFactory" />
        <ConnectionFactory name="RMIConnectionFactory"/>
      </ConnectionFactories>
    </Connector -->
</p>

NOTE: If you don’t want to comment the above code, you should modify the JMX listening port in ActiveMQ because by default activemq is listening 1099 port number. The same port number is also listening openjms rmi connector.

Steps to Run the Sample Code

Step1: Copy Open-JMS jar files to <ActiveMQ-INSTALLED_DIR>/lib directory.

Step2: Start Open-JMS using the following batch file.

<openjms_install_dir>\bin>start startup.bat

Step3: Start ActiveMQ-JMS using the following batch file.

<activemq_install_dir>\bin\start activemq.bat

Step4: run the SimpleMessageSender class using the following:

java SimpleMessageSender “This is a Sample Message”

SimpleMessageSender output on console

Step5:

Open another cmd window and set classpath and java path and run SimpleMessageListener class.

java SimpleMessageListener

SimpleMessageListener output on console:

This is all the Java code you need to write to publish and consume java message. Let's examine the activemq.xml jmsBridgeConnectors elements:

jmsBridgeConnectors : is the root element for jms bridge connectors.

jmsQueueConnector: A Bridge to other JMS Queue providers.

jndiOutboundTemplate: used for locating the Connection Factory for the ActiveMQ Connection if the localTopicConnection or localTopicConnectionFactory is not set.

outboundQueueConnectionFactoryName: used to initialize the foreign JMS Connection if localQueueConnection is not set.

localQueueConnectionFactory: used to initialize the ActiveMQ JMS Connection if localQueueConnection is not set.

inboundQueueBridge: Create an Inbound Queue Bridge.

inboundQueueName:The foreign queue name to receive from. If more information about these elements please visit the following link ActiveMQ.

History

version1.0