Working With Queues

ActiveMQ Connection with TomEE JNDI

2022-01-07T03:45:00.003-08:00

1. Create Dynamic Web Project in Eclipse

2. Use below configuration in Tomee.xml in conf:

<Resource id="amqRA" type="ActiveMQResourceAdapter">
      BrokerXmlConfig = broker:(tcp://localhost:61616)?useJmx=false
      ServerUrl = vm://localhost
      StartupTimeout = 10 seconds
</Resource>

<Resource id="amqCF" type="jakarta.jms.ConnectionFactory">
      ResourceAdapter = amqRA
      TransactionSupport = xa
      PoolMaxSize = 10
      PoolMinSize = 0
      ConnectionMaxWaitTime = 5 seconds
      ConnectionMaxIdleTime = 15 Minutes
</Resource>

3. Java Lookup code in Servlet:

package com.teg;

import java.io.IOException;
import java.io.PrintWriter;

import javax.naming.InitialContext;
import javax.naming.NamingException;

import jakarta.jms.Connection;
import jakarta.jms.ConnectionFactory;
import jakarta.jms.JMSException;
import jakarta.servlet.ServletConfig;
import jakarta.servlet.ServletException;
import jakarta.servlet.http.HttpServlet;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;

/**
* Servlet implementation class JndiTest
*/
public class JndiTest extends HttpServlet {
   private static final long serialVersionUID = 1L;
   private ConnectionFactory cf = null;
    /**
     * Default constructor.
     */
    public JndiTest() {
        // TODO Auto-generated constructor stub
    }

   /**
   * @see Servlet#init(ServletConfig)
   */
   public void init(ServletConfig config) throws ServletException {
        System.out.println("Application Deployed in TomEE");
        try {
            InitialContext initContext = new InitialContext();
            ConnectionFactory cf = (ConnectionFactory)
                    initContext.lookup("openejb:Resource/amqCF");
            System.out.println("Connection Factory:: " + cf);
            Connection connection = cf.createConnection();
            connection.start();
            System.out.println("Connection can be tried here");
            connection.close();
        } catch (NamingException | JMSException e) {
            System.out.println(e);
            e.printStackTrace();
        }
   }

   /**
   * @see HttpServlet#doGet(HttpServletRequest request, HttpServletResponse response)
   */
   protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        // TODO Auto-generated method stub
        PrintWriter out = response.getWriter();
        out.println("Served at: doGet");
        try {
            Connection connection = cf.createConnection();
            connection.start();
            out.println("Connection can be tried here");
            connection.close();
        } catch (JMSException e) {
            out.println(e);
            e.printStackTrace();
        }
        out.println("Serviced at: doGet");
   }

   /**
   * @see HttpServlet#doPost(HttpServletRequest request, HttpServletResponse response)
   */
   protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        // TODO Auto-generated method stub
        doGet(request, response);
   }

}

4. Export WAR file in Put it in Tomee Webapps folder

- Please delete existing wars, folders in Webapps and Work folder before running new project.

Simple ActiveMQ Message Producer Example

2016-12-14T19:09:00.002-08:00

While Working with JMS, starting with a simple example in local is always best. ActiveMQ provides best way to test your code locally by giving vm://localhost:61616 connector.

Below example will help you to start with JMS Producer.

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

public class SenderTest {

public static void main(String[] args) {
Connection conn = null;
try {
// 1 - Connection Factory
ConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost:61616");
// 2 - Create a Connection
conn = factory.createConnection();
conn.start();
// 3 - Create a Session
Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
// 3A - Create a Message
Message message = session.createTextMessage("Hello World in JMS!!");
// 3B - Create a Destination / Queue / Topic

Destination destination = session.createQueue("TEST.QUEUE");
// You can create specific Instances
Queue queue = session.createQueue("OTHER.QUEUE");
Topic topic = session.createTopic("TEST.TOPIC");

// Step 4
MessageProducer producer = session.createProducer(destination);
producer.send(message);
// close the connection
session.close();
conn.close();

} catch (JMSException e) {
e.printStackTrace();
}
}

}

SpringJMS - Step by Step Guide to create Consumer (Super Simple)

2015-10-31T01:35:00.000-07:00

Step 1 - You will need same Jar files which we discussed in Producer Example

Spring Core
Spring Context
Spring JMS
activemq-all
commons-logging
slf4j

Step 2: Create applicationContext.xml in classpath

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:p="http://www.springframework.org/schema/p"

xmlns:context="http://www.springframework.org/schema/context"

xmlns:jms="http://www.springframework.org/schema/jms"

xmlns:amq="http://activemq.apache.org/schema/core"

xsi:schemaLocation="http://www.springframework.org/schema/beans

http://www.springframework.org/schema/beans/spring-beans.xsd

http://www.springframework.org/schema/context

http://www.springframework.org/schema/context/spring-context.xsd

http://www.springframework.org/schema/jms

http://www.springframework.org/schema/jms/spring-jms.xsd

http://activemq.apache.org/schema/core

http://activemq.apache.org/schema/core/activemq-core.xsd">

<context:component-scan base-package="com.itech" />

<context:annotation-config/>

<bean id="activemqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<constructor-arg name="brokerURL" value="vm://localhost:61616" />

</bean>

<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">

<constructor-arg ref="activemqConnectionFactory" />

</bean>

<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

<constructor-arg index="0" value="THIS.IS.TEST.QUEUE" />

</bean>

<jms:listener-container connection-factory="connectionFactory">

<jms:listener destination="THIS.IS.TEST.QUEUE" ref="messageListenerTest" />

</jms:listener-container>

</beans>

Step 3:

You need a Consumer Class which implements MessageListener interface:

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

import org.springframework.stereotype.Component;

@Component("messageListenerTest")

public class MessageListenerTest implements MessageListener {

@Override

public void onMessage(Message message) {

TextMessage textMessage = (TextMessage) message;

try {

System.out.println(textMessage.getText());

} catch (JMSException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

That's it, Now try sending a message with Producer example, and message will be listened by this MessageListener implementation.

Super Simple Example of Spring JMS - Producer

2015-10-29T22:04:00.000-07:00

Step 1: You need Jar files:

A. Spring core, spring-context
B. Spring JMS
C. activemq-all
D. slf4j , log4j etc.

Step 2: Create a Java Project in Eclipse and add "applicationContext.xml" File directly under "src" folder:

applicationContext.xml :

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd">

<context:component-scan base-package="com.itech" />
<context:annotation-config/>


<bean id="activemqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

<constructor-arg name="brokerURL" value="vm://localhost:61616" />
</bean>


<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg ref="activemqConnectionFactory" />
</bean>


<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">

<constructor-arg index="0" value="THIS.IS.TEST.QUEUE" />
</bean>


<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory" />
<property name="defaultDestination" ref="queueDestination" />
</bean>

</beans>

Have a look at XML, it has almost everything you need to start with: - ConnectionFactory, Destination and Spring Provides JmsTemplate which we will be using to send the message.

Step 3: Create Test Class:

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

public class ProducerTest {

public static void main(String[] args) {
// Load Application Context
ApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");

// Get JmsTemplate Bean from context
JmsTemplate template = (JmsTemplate) ctx.getBean("jmsTemplate");

// Create Message
MessageCreator message = new MessageCreator() {
@Override
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage("This is test message");
}
};

// Send the message
template.send(message);

// Close the context
((ClassPathXmlApplicationContext)ctx).close();
// You are Done!!
}

}

That's it. You are Done! You have written a complete code in SpringJMS for sending a message.

Thanks!

MDB (Message Driven Beans) Vs. SpringJMS - Point by Point Comparison

2015-10-29T10:46:00.000-07:00

EJB’s Message Driven Beans (MDB) Vs. SpringJMS

MDB and SpringJMS can be compared with multiple features:

1. Learning Curve

Most important feature of SpringJMS is, object can be easily configured in XML format making is extra easy to manage. Some features like concurrentConsumer can be managed with XML without touching Java code.

MDB on the other hand involves expert level understanding of EJB and has more LOC to write for same operation. So Overall learning Curve for SpringJMS is less.

2. Development Time

Spring implementation uses the JmsTemplate helper class to simplify working with the JMS API to send the message. Also, auto starting consumers with application context and ease of handling properties with xml configuration, reduces development time, in turn production cost of application.

MDB though robust, needs good amount of efforts in comparison.

3. Container Dependency

MDB requires EJB container. Container is mainly responsible for instantiation and handling other life cycle events for MDB. EJB 3.0 also has additional packaging and deployment requirements that
Spring does not. Also message listeners must be EJB components.

Spring is container of its own. SpringJMS do not require any external container to execute. Any POJO can potentially be activated in response to incoming messages

4. Multi-Queue Configuration:

MDBs are static for one queue. One MDB cannot be used for consuming from two or multiple queues. On the contrary, Spring provides the flexibility to customize the number of queues to listen to and dynamically configure the destinations as well.

5. Security

For MDB, security is managed by the EJB container by configuring a security identity for the MBM

Spring UserCredentialsConnectionFactoryAdapter provides a mechanism to specify user credentials for every JMS connection to secure the JMS connection.

6. Dependency management/configurability/intermediation

Spring offers much more mature and configurable support for dependency management than EJB 3.0 with its dependency injection and IOC framework

7. Availability

The Enterprise server controls JMS connection pooling and failover and reconnecting to the message broker when the connection is lost has to be supported and configured at the enterprise server level .

Individual applications and modules cannot control or override JMS availability

Spring DMLC automatically re-establishes connections if the message broker becomes unavailable

8. Performance

The EJB MDB container manages the JMS Connection Pool and needs to be configured to optimize the concurrent processing of messages

Spring DMLC allows you to scale the number message consumers using the concurrentConsumers property and the maxConcurrentConsumers property. It provides various levels of caching of the JMS resources (connections and sessions) and JMS consumers for increased performance

9. Transaction

Can use the default container managed transaction to allow the EJB container to handle transaction demarcation or specify local transactions to group message sends and receives

Spring DMLC provides support for local JMS transactions as well as an external transaction manager around message reception and listener execution

10. Message Convertors

Spring provides message converters which can automate the conversion of Java objects to message content

MDB provides no such features.

11. Future Orientation

Spring has vibrant open source community and considered as very stable but futuristic design. Many newer MOMs and EIPs (Enterprise Integration Pattern) like Camel directly uses Spring for messaging implementation.

EJBs on the contrary and seen lesser in market.

Summary:

Considering current era of IT development, adopting SpringJMS over MDB is always a better choice.

Using MQXAQueueConnectionFactory in Tomcat

2013-08-06T08:53:00.000-07:00

Using Atomikos / XA transaction with your connection is not a bad idea sometimes. Below Connection Factory resource will give a nice idea about tomcat JNDI entry for

MQXAQueueConnectionFactory.

<Resource

name="jms/MQCF"

auth="Container"

type="com.atomikos.jms.AtomikosConnectionFactoryBean"

factory="com.atomikos.tomcat.EnhancedTomcatAtomikosBeanFactory"

uniqueResourceName="MQPutQCFXA_MQSeries_XA_RMI"

xaConnectionFactoryClassName="com.ibm.mq.jms.MQXAQueueConnectionFactory"

xaProperties.hostName="YOUR_HOST"

xaProperties.port="XXXX"

xaProperties.channel=" CHANNEL "

xaProperties.queueManager=" YOUR _QMGR"

maxPoolSize="3"

minPoolSize="1" />

I see a common exception while connecting with XA Connection factory:

Caused by: com.ibm.mq.jmqi.JmqiException: CC=2;RC=2495;AMQ8568: The native JNI library 'mqjbnd' was not found. [3=mqjbnd]
    at com.ibm.mq.jmqi.local.LocalMQ$1.run(LocalMQ.java:252)
    at java.security.AccessController.doPrivileged(Native Method)
    at com.ibm.mq.jmqi.local.LocalMQ.initialise_inner(LocalMQ.java:216)

This occurs when you are trying to connect to local binding. You can change that to client connection by making transport type as ‘1’.

xaProperties.transportType="1"

Just adding this property in Resource entry should resolve the issue. Please post questions if any.

I will add ActiveMQ XA Connection factory example soon.

ActiveMQ Composite Destinations

2012-10-12T07:23:00.002-07:00

ActiveMQ Composite Destinations

Composite queue allows one-to-many relationship. Send a message to One Queue (A) and it will forward the message to Queue B, Queue C and Topic D.

To map Composite destination you need to add following section in virtualDestinations tag.

</forwardTo>

</compositeQueue>

</virtualDestinations>

</virtualDestinationInterceptor>

</destinationInterceptors>

By default, subscribers cannot consume messages directly from a composite queue or topic - it is a logical construct only. Given the configuration above, subscribers can only consume messages from TEST.QUEUE.B, TEST.QUEUE.C and TEST.TOPIC.D; but NOT TEST.QUEUE.A.

Using filtered destinations

You may wish to create a virtual destination which forwards messages to multiple destinations but applying a selector first to decide if the message really does have to go to a particular destination.

</forwardTo>

</compositeQueue>

</virtualDestinations>

</virtualDestinationInterceptor>

</destinationInterceptors>

In the case above, message from TEST.QUEUE.A will be forwarded to filtered destinations with selector criteria. Result is same as having selector based consumer on TEST.QUEUE.B

Important Aspects of Composite Destinations:

- Listener Queues (under forwardTo Tag) will not be created automatically. If queues are not present, message will be LOST.

- If you want to store message in composite queue, make forwardOnly flag as ‘false’. It’s true by default.

<compositeQueue name="TEST.QUEUE.A" forwardOnly="false">

      <forwardTo>

</forwardTo>

</compositeQueue>

- Composite queue can send the message to Virtual Topic, which will be further consumed by multiple Virtual Destinations.

- Virtual Topic Selector Aware Consumers will not pull the message if consumer is not active; On the contrary, Filtered Composite destinations pull the message despite of consumer’s active status. This is very helpful in real world scenarios.

- Important: Composite Destinations Supports wildcards. This is also valid for Filtered Destinations.

</forwardTo>

</compositeQueue>

</virtualDestinations>

</virtualDestinationInterceptor>

</destinationInterceptors>

In the settings above, message will be forwarded to all the queues starting with “COMPOSITE.WILD”. Also, the message with filtered destinations will be forwarded to all the queues starting with ‘TEST.B’.

WTX with ActiveMQ Using JMS Adapter

2012-08-30T08:06:00.000-07:00

WebSphere Transformation Extender's interaction with Activemq. Not hard but little tricks are important. As usual, if you google this solution, you will not find a single site or blog stating exact steps. So, here it is. Post your comments for questions.

Step 1: Setting jndi.propertise

Place provided “jndi.properties” file in <WTX_INSTALL>/java/lib directory or defined CLASSPATH. Sample jndi.properties can be downloaded from here.

http://activemq.apache.org/jndi-support.html

Step 2: Include Jars

Include following Jars in local accessible folder.

activemq-all-5.5.1-fuse-07-11.jar

geronimo-jms_1.1_spec-1.1.1.jar

log4j-1.2.16.jar

slf4j-api-1.6.4.jar

slf4j-log4j12-1.6.4.jar

spring-jms-3.0.6.RELEASE.jar

Step 3: Add External Jars in dtx.ini

Include above mentioned jars in External jar section dtx.ini file. (replace home/aad/tmp with your folder path from root)

jar1=/home/aad/tmp/activemq-all-5.5.1-fuse-07-11.jar

jar2=/ home / aad /tmp/geronimo-jms_1.1_spec-1.1.1.jar

jar3=/ home / aad /tmp/log4j-1.2.16.jar

jar4=/ home / aad /tmp/slf4j-api-1.6.4.jar

jar5=/ home / aad /tmp/slf4j-log4j12-1.6.4.jar

jar6=/ home / aad /tmp/spring-jms-3.0.6.RELEASE.jar

Step 4: Command

-T -HDR -ICTXFURL tcp://<brokerURL>:<port>

-ICTXF org.apache.activemq.jndi.ActiveMQInitialContextFactory

-CFN "connectionFactory"

-QN " testQueue"

Where

connectionFactory – JNDI Connection Factory Name

testQueue – JNDI Queue Name

WebSphere MQ With Tomcat : JNDI Property Names

2012-08-08T11:14:00.000-07:00

WebSphere MQ JNDI Property Names:

I searched across the web and I didn’t find a single site stating exact JNDI MQ option for setting SSLCipherSuite value. I made this entire list of properties which can be useful for providing wide range of options to MQConnectionFactoty.

WebSphere MQ (IBM MQ) has lot of properties which can be made available from JNDI (Context.XML) entries. IBM MQ uses very specific Names which need to be set in Context XML Resource Entry.

Sample Resource Entry in Context is as below:

<Resource

CHAN="<Channel Name>"

HOST="<HostName>"

PORT="<portNumber>"

QMGR="<queueManagerName>"

auth="Container"

description="MQ JMS Queue Connection Factory"

factory="com.ibm.mq.jms.MQQueueConnectionFactoryFactory"

name="jms/ConnectionFactory"

type="com.ibm.mq.jms.MQQueueConnectionFactory"/>

Below is the list of properties and their meaning.

Property Name	Used For	Short Description
CHAN	Channel Name	1 through 20 ASCII characters
HOST	Host Name	A valid TCP/IP host name
PORT	Port Number	A valid TCP/IP port numbe
QMGR	Queue Manager Name	Queue Manager Name
VER	Version	String
DESC	Description	String
TRAN	Transport Type	BINDINGS, CLIENT, DIRECT, QUEUED
CID	Client ID	JMS client identifier
CCS	CCS ID	coded character set identifier
MNS	Map Style Name	String
RCX	Receive Exit	String
RCXI	Receive Exit Init	String
SCX	Security Exit	String
SCXI	Security Exit Init	String
SDX	Send Exit	String
SDXI	Send Exit Init	String
SCPHS	SSL Cipher Suite value	Cipher suite to use for SSL connection
SPEER	SSL Peer Name	Distinguished name is used to check the identifying certificate presented by the server at connection time
SCRL	SSL Certificate Store	list of zero or more Certificate Revocation List
HC	HDR Component list	String
MC	Message Comp list	String
CT	Connection Tag	String
CTO	Connection Options	String
SCC	Send Check Count	String
SCALD	Shared Conversation Allowed	String
SRC	SSL Reset Count	String
SFIPS	SSL Fips Required	String
SPAG	Synch Points All Gets	String
UCP	Use Connection Pooling	Optional set of connection pool settings
PINT	Polling Interval	The interval, in milliseconds, between scans of all receivers during asynchronous message delivery
PVER	Provider Version	String
WCFMT	Wild Card Format	String
MBS	Message Batch Size	String
FIQ	Fail If Quiesce	Whether applications return from a method call if the queue manager has entered a controlled failure.
LA	Local Address	Specifies the local server address
RINT	Rescan Interval	the interval in milliseconds between which a topic is scanned to look for messages that have been added to a topic out of order
TCM	Target Client Matching	String
AEX	Asynch Exception	String
CCDTRUL	CCDT URL	String

These properties are very useful while using websphere MQ with Tomcat. This will enable user to get rid of provider specific code in App.

This list doesn’t end here. I will add more properties, as and when I get to know them, I will also add the descriptions as I find them.

More questions? Post a comment. I will try to answer as soon as possible.

ActiveMQ Mirror Queue: Working Example.

2012-08-07T08:51:00.001-07:00

If you reached here, you might have already checked few very confusing paragraphs about Mirror queues in ActiveMQ.

Mirror Queues are disabled by default. To enable them, you have to following settings in activemq.xml

<virtualTopic name="MIRROR.TEST.>" prefix="MIR.>." selectorAware="false"/>

</virtualDestinations>

</virtualDestinationInterceptor>

</destinationInterceptors>

Out of above: “<mirroredQueue copyMessage = "true" postfix="" prefix="MIRROR."/>” is actually responsible for enabling Mirror Queues.

How ActiveMQ Handles Mirror Queues:

1. When you enable Mirror Queues, with say prefix as “MIRROR.”, for every queue in the broker one Topic will be created with MIRROR. Prefix.

2. Now to receive a message from this Topic, Virtual Destinations can be used. More about Virtual Destination can be found here.

Working Example:

Let’s consider above configuration for prefix for Mirror queues and Virtual Destinations.

1. Create a queue: TEST.QUEUE

2. With above settings, Mirror Topic is automatically created as : MIRROR.TEST.QUEUE

3. Create Virtual Topic Consumer : MIR.COPY.MIRROR.TEST.QUEUE

a. MIR – Defined Virtual Topic Interceptor

b. COPY – Name of the Consumer

c. MIRROR.TEST.QUEUE – is a Topic Name

4. Send a test message to TEST.QUEUE, same message will appear in MIR.COPY.MIRROR.TEST.QUEUE.

This is complete setting of Mirror Queues.

Problems:

1. Mirror Topics will be created for Every Queue in the broker

2. As broker has to track each message from the queue and copy it to topic, further to virtual destination, broker may face performance problem depending on load.

QueueSender or MessageProducer? Which one I should choose?

2012-07-27T14:07:00.004-07:00

QueueSender or MessageProducer?

Well the answer is: - MessageProducer. There is nothing new in QueueSender or TopicPublisher which is not available in MessageProducer. But converse is true. MessageProducer has few more features than QueueSender and TopicPublisher. Creating a MessageProducer provides the same features as creating a TopicPublisher. or QueueSender.

Also, there is an added benefit, if your code is sending messages to Queue as well as a Topic, you can dynamically pass destination to your MessageProducer creator and make the code consistent.

This analogy can be used with QueueSession and TopicSession objects. Using a generic Session is always recommended.

But don’t be panic if you have existing code with QueueSender and QueueReceiver, for futher development though, please use their parents J

Did you still have a doubt? Post a comment. I will try to answer it as soon as possible.

Converting CSV to XML With Camel Bindy

2012-07-20T07:19:00.001-07:00

Converting CSV file to XML using Apache Camel.

Why Apache Camel?

Because Camel Rocks :). Camel has huge transformation libraries which makes code easier and simple to understand and execute. Also Camel is free. It’s an open-source tool with huge user base and nice support.

What’s the Corporate Problem?

Converting CSV to XML is common problem across IT industry where CSV might be the input imported by user and Web-Service might be expecting XML formatted data to run the business logic. Here we will do this conversion in two steps, and both the steps are useful for applicable applications.

Step 1. Converting CSV file to POJO objects (Beans)

Step 2. Converting POJO list to XML using xstream (Open Source easy to use Jar file)

If you are just interested in POJO list, please stop at first step.

Can you run through any example?

Sure thing. Let’s consider we have Employee.csv file having data like below.

employeeId,firstName,lastName,role

23445,Ambarish,Deshpande,Java Lead

23411,Paul,Bulson,Team Member

First we need a bean (DTO) class to map CSV fields. Getters / Setters are optional here.

@XmlRootElement

@XmlAccessorType(XmlAccessType.FIELD)

@CsvRecord(separator = ",", skipFirstLine = true)

public class EmployeeDTO implements Serializable{

@XmlAttribute

@DataField(pos = 1)

private int employeeId;

@XmlAttribute

@DataField(pos = 2)

private String firstName;

@XmlAttribute

@DataField(pos = 3)

private String lastName;

@XmlAttribute

@DataField(pos = 4)

private String role;

}

Annotations?

All @XML* annotations are coming from javax.xml.bind.annotation.* package.

For @CSVRecord and @DataField you have import camel-bindy.jar. I have listed all the dependencies in Appendix section at last.

Now, for actual conversion, camel route must be created.

public class ConvertorRoute implements RoutesBuilder{

@Override

public void addRoutesToCamelContext(CamelContext context) throws Exception {

context.addRoutes(new RouteBuilder() {

public void configure() {

try {

DataFormat bindy = new BindyCsvDataFormat("com.package.dto");

from("file://TEST?fileName=Employee.csv").

unmarshal(bindy).

marshal().

xstream().

to("file://TESTOUT?fileName=employee.xml");

} catch (Exception e) {

e.printStackTrace();

}

});

}

public static void main(String[] args) {

try{

CamelContext context = new DefaultCamelContext();

ConvertorRoute route = new ConvertorRoute();

route.addRoutesToCamelContext(context);

context.start();

Thread.sleep(5000);

context.stop();

}catch(Exception exe){

exe.printStackTrace();

}

Steps in Route:

1. Register Bean package (where DTO class resides) with BindyCSVDataFormat.

DataFormat bindy = new BindyCsvDataFormat("com.package.dto");

2. Pick Up the file from TEST folder. (Root//TEST)

3. Unmarshal the csv file with register Bindy. This step creates a List of EmployeeDTO. If intended output is achieved here. Please use this output further.

4. Marshal().xstream() will convert this DTO List to XML and finally

5. Output file will be written on TESTOUT folder as employee.xml file.

Employee.XML file:

<?xml version="1.0" ?>

<list>

<map>

<entry>

<string>com.thd.eai.dto.EmployeeDTO</string>

<com.thd.eai.dto.EmployeeDTO>

<firstName>Ambarish</firstName>

<lastName>Deshpande</lastName>

</com.thd.eai.dto.EmployeeDTO>

</entry>

</map>

<map>

<entry>

<string>com.thd.eai.dto.EmployeeDTO</string>

<com.thd.eai.dto.EmployeeDTO>

<lastName>Bulson</lastName>

<role>Team Member</role>

</com.thd.eai.dto.EmployeeDTO>

</entry>

</map>

</list>

Appendix and References:

List of dependencies:

camel-bindy-2.9.0.jar,

camel-core-2.9.0.jar,

camel-csv-2.9.0.jar,

camel-xstream-2.9.0.jar,
slf4j-api-1.6.4.jar,

slf4j-log4j12-1.6.4.jar,

solr-commons-csv-1.3.0.jar,

xstream-1.3.1

Important Links:

Camel Bindy: http://camel.apache.org/bindy.html

Camel XStream: http://camel.apache.org/xstream.html

Camel CSV: http://camel.apache.org/csv.html

About XStream: http://xstream.codehaus.org/

ActiveMQ Message Sequencing : Analysis Part I

2012-06-29T12:02:00.000-07:00

Message Sequencing POC:

In most of the corporate production environments, messages are not sent and received from the same broker. Message might end up from multiple hops and clusters within infrastructure before it actually hits the business consumer. I did a small analysis with ActiveMQ out of the box configuration to use sequencing. Below are the results.

Test Case 1:

Messages are sent to queue / Topic and received in same broker (even from Virtual Destination queue). Single producer and exclusive consumer are used.

Result:

1. Messages are always received in Sequence in normal running consumer

2. Message maintains their sequence even if consumer thread restarts in between

3. Message Priority is IGNORED and sequenced is maintained

4. Results matches the documentation provided here: http://activemq.apache.org/exclusive-consumer.html

Test Case 2:

Messages are going through multiple hops before get consumed from exclusive consumer.

Result:

1. Message loses sequence

2. Priority Set on the messages resets to default (‘4’) regardless of producing value.

Summary:

· If producer and consumer maintained in same broker and consumers are made exclusive; message sequencing is guaranteed.

· If message is making multiple hops before reaching to consumer, sequencing is NOT guaranteed even if Exclusive consumers are used.

In Part II, we will use JMXGroupId concept for message sequencing and find the right results.

Virtual Topic SelectorAware Consumers

2012-06-19T06:29:00.002-07:00

Making a Consumer selectorAware comes with a lackuna.

If selectorAware property is r’true’, by default, consumer queue will not pull the message from the Topic, unless there is atleast one ‘Active Consumer Application’ listening to consumer queue. Even if there is no selector defined on the consumer queue.

On the contrary, if the selectorAware is false, all the messages from the Topic will be posted to consumer queue regardless of consuming application availability.

So, its actually a design decision to use this flag wisely for proper implementation.

Making Virtual Topic Consumer “Selector Aware”

2012-05-29T13:29:00.000-07:00

Making Virtual Topic Consumer “Selector Aware”

</virtualDestinations>

Making Virtual topics selector aware is as easy as flipping the flag to “true”. For this example we created two consumers, one - “VTCON.GEN.MF.PRICE.CHANGE.TP01” listens to all the messages for ‘MF.PRICE.CHANGE.TP01’ topic and second – “VTCON.MKT.MF.PRICE.CHANGE.TP01" has a Selector of “Market = 100” on it.

Now, let’s send two messages, one with “market” string property with value 100 and another with value 200. In a result we should get one message in VTCON.MKT.MF.PRICE.CHANGE.TP01 as it has a Selector of “Market = 100” on it and two messages in “VTCON.GEN.MF.PRICE.CHANGE.TP01”

Java Code will look like below:

package test.consumer;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
public class VirtualTopicSelectorAwareConsumer {

static Connection connection;
static Session session;

public static void main(String[] args) {
ActiveMQConnectionFactory connectionFactory =
new ActiveMQConnectionFactory("tcp://yourserver.com:port");

try {
connection = connectionFactory.createConnection();
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
// Consumer Listens to message with Market String property as 100
String queueName = "VTCON.MKT.MF.PRICE.CHANGE.TP01";
Destination mkt = session.createQueue(queueName);
String selector100 = "Market = '100'";
MessageConsumer consumer100 = session.createConsumer(mkt, selector100);
Consumer100 listener = new VirtualTopicSelectorAwareConsumer().new Consumer100();
consumer100.setMessageListener(listener);
session.commit();
// Consumer listens to all the message (Without a Selector)
String genralConsumer = "VTCON.GEN.MF.PRICE.CHANGE.TP01";
Destination genral = session.createQueue(genralConsumer);
MessageConsumer genConsumer = session.createConsumer(genral);
ConsumerNormal normal = new VirtualTopicSelectorAwareConsumer().new ConsumerNormal();
genConsumer.setMessageListener(normal);
session.commit();

Topic topic = session.createTopic("MF.PRICE.CHANGE.TP01");
MessageProducer producer = session.createProducer(topic);
Message mktMessage = session.createTextMessage("Message for Market 100");
mktMessage.setStringProperty("Market", "100");
producer.send(mktMessage);

Message mktMessage2 = session.createTextMessage("Message for Market 200");
mktMessage.setStringProperty("Market", "200");
producer.send(mktMessage2);
producer.close();
session.commit();

} catch (JMSException e) {
e.printStackTrace();
}

}
/**
* ConsumerNormal Listens All the messages and Log it
*/
private class ConsumerNormal implements MessageListener{

@Override
public void onMessage(Message arg0) {
try {
  System.out.println(((TextMessage)arg0).getText());
  session.commit();
} catch (JMSException e) {
  e.printStackTrace();
}
}
}

/**
* Consumer100 only listens to Market=100 messages
*/
private class Consumer100 implements MessageListener{
@Override
public void onMessage(Message arg0) {
try {
System.out.println(((TextMessage)arg0).getText());
session.commit();
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}

ActiveMQ Virtual Topics or Virtual Destinations

2012-05-18T07:25:00.000-07:00

Creating Virtual Topics

Virtual Topics helps with following prospective:

1. Load Balancing of messages

2. Fast Failover of Subscriber

3. Re-using same connection Factory for different Producers and Consumers. (Durable Subscribers needs a Unique JMS Client Id and same cannot be reused for any other Producer or consumer)

Steps to Define a Virtual Topic:

1. Topic has to be created on Broker with Name: VirtualTopic.<TopicName>, However “VirtualTopic” is Out-of-the-box prefix for using this functionality.

2. To make the change in prefix, following changes are needed.

<virtualTopic name=">" prefix="VTCON.*." selectorAware="false"/>

</virtualDestinations>

</virtualDestinationInterceptor>

</destinationInterceptors>

</broker>

In above settings two things are achieved.

A. All Topics are made as Virtual Topic with name=">". However name field also accepts wildcard to apply different Virtual Topics policies. To change this Virtual Topic for Single Topic, just replace the “>” with actual Topic name.

</virtualDestinations>

B. Prefix for Consumer will be “VTCON”. Now Virtual Topic consumers will be created with name like “VTCON.BASETBL.TEST.TP01”

C. Settings are also made to NOT allow VirtualTopic Consumers as selectorAware. This will forward every message published to topic to consumers with “VTCON” prefix. If selectorAware is true, consumers will be able to add selector and only selected messages will be passed to consumers.

CSV to XML transformation using Camel

2012-05-15T12:52:00.003-07:00

Lots of Projects I know needs this transformation.
CSV to XML, where you might be reading user uploaded CSV file or it can be Pipe delimited flat file and make a XML out of it.
Below is the code doing this using camel and Xstream. I am exploring more to get XML nodes to populate as headers, but not yet successful. I will update that as soon as I get the solution. till then.. this is the good code u can use.

package com.XYZ;

import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;

import com.thoughtworks.xstream.XStream;
import com.thoughtworks.xstream.io.xml.DomDriver;

public class FileCSVtoXMLConversion {
private static XStream xStream;

public static void main(String args[]) throws Exception {
// create CamelContext
CamelContext context = new DefaultCamelContext();
xStream = new XStream(new DomDriver());
// add our route to the CamelContext
context.addRoutes(new RouteBuilder() {
public void configure() {
from("file://Some/TEST?fileName=ArticleTest.csv").
unmarshal().
csv().
process(new Processor() {
  @Override
  public void process(Exchange exchange) throws Exception {
    String xmlConverted = xStream.toXML(exchange.getIn().getBody());
    exchange.getIn().setBody(xmlConverted);
  }
}
).to("file://Some/TESTOUT?fileName=ArticleTest.XML");
}
});

// start the route and let it do its work
context.start();
Thread.sleep(10000);

// stop the CamelContext
context.stop();
}
}

Sending a Message to Queue (JMS Way)

2012-05-14T13:42:00.001-07:00

There are thousand confusing ways to send a message to queue and every method looks correct and perfect for production implemetation.
Below is simple method which is used most of the time and also JMS API Complient. Benefit of using JMS Complient provider is: your code will remain unchanged even if provider changes. But yes, you have to make sure about you exception handling mechanism and your queue access.

Steps are explained below:
1. Get the ConnectionFactory from JNDI
2. Create Connection from Connection Factory
3. Get the destination from JNDI or Create Destination
3. Create a Session from Connection
4. Create a Message Producer from Session using Destination
5. Create a Message from Session
6. Send the message using message producer.
7. Close the mess.
8. Check the queue depth - Not in Java (using queue browsing tools)

package com.xyz;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class SampleMessageSender

{

private String jndiQCFName = null;
private ConnectionFactory factory = null;

public SampleMessageSender(String jndiQCFName){
this.jndiQCFName = jndiQCFName;
}

public void sendMessage(String jndiDestination, String messagePayload)

throws NamingException

{
InitialContext context = null;
MessageProducer producer = null;
Connection connection = null;
Session session = null;
try

{
context = new InitialContext();
factory = (ConnectionFactory) context.lookup("java:comp/env/"+jndiQCFName);
Destination destination = (Destination) context.lookup("java:comp/env/"+jndiDestination);
connection = factory.createConnection();
session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
// Create a Producer for this Destination
producer = session.createProducer(destination);
// Create a Message from Payload
Message message = session.createTextMessage(messagePayload);
// Send the message
producer.send(message);
}

catch(JMSException exe)

{
exe.printStackTrace();
// Do logging
if(exe.getLinkedException() != null){
// Linked exception may have provider specific details. Log it.
exe.getLinkedException().printStackTrace();
}
}

finally{
context.close();
try {
producer.close();
session.close();
connection.close();
} catch (JMSException e) {
// Nothing can be done here
e.printStackTrace();
}
}
}
}

2012-01-31T10:57:00.000-08:00

MQ Queue Manager: if you can directly access MQQueueManager object in your working
Environment, following is the way to create the Object directly.

MQEnvironment.hostname = "yourhost.yourcompany.com";

MQEnvironment.port = CLIENT_PORT; // Mention int port here

MQEnvironment.channel = CLIENT_CHANNEL; // Menion Server connection channel here

/* Mention queueManagerName here */
MQQueueManagerqueueManager = new MQQueueManager(queueManagerName);

Thats it. hostname, port , channel and queueManagerName can connect you to your queue Manager.

And yes. dont forget to import this:

import com.ibm.mq.MQEnvironment;
import com.ibm.mq.MQQueueManager;

Handling Big Messages with MQ (MQSeries)

2012-01-19T11:06:00.000-08:00

Its very hard to handle big messages, especially more than 3-4 MB. You will receive a JMSException after certain limit. But MQ gave us a concept of grouping the messages if required. Only problem is, if you have a Single huge file, how you can devide it, send it and receive it in proper sequence.

Below code exactly does the same thing. Hope it will help you. Please let me know sugestions and changes if required.

package com.xyz;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import javax.jms.Connection;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.log4j.Logger;
import com.ibm.mq.jms.JMSC;
import com.ibm.mq.jms.MQConnectionFactory;
import com.ibm.mq.jms.MQQueue;
public class MQMessageSegments {
private static Logger log = Logger.getLogger(MQMessageSegments.class);

static MQDirectConnector connector = null;
static Connection connection = null;
static Session session = null;
static Queue sendToQueue = null;
static int ch;
static Map<String, FileOutputStream> map = new HashMap<String, FileOutputStream>();
private static MQConnectionFactory factory = null;
private static String hostName = "localhost.homedepot.com";
private static int port = 1414;
private static String queueManagerName = "COM.AAD.LCL.TST";
private static String channelName = "SYSTEM.DEF.SVRCONN";
private static String localQueueName = "AAD.LOCAL.QUEUE" ;

static Connection conn2 = null;

static Session session2 = null;
public static void main(String args[]){

  try {
   log.debug("Connecting to Local Queue Manager - Start");
    factory = new MQConnectionFactory();
    factory.setHostName(hostName);
    factory.setPort(port);
    factory.setQueueManager(queueManagerName);
    factory.setChannel(channelName);
    connection = factory.createConnection();
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    MQQueue destination = new MQQueue(localQueueName);
    destination.setTargetClient(JMSC.MQJMS_CLIENT_NONJMS_MQ);

    MessageProducer producer = session.createProducer(destination);
   log.debug("Connecting to Local Queue Manager - End");
   /************* Here Starts Sending Part ********/
   log.debug("Creating Messages and Sending it across - Start");
    StringBuffer fileBuffer = new StringBuffer();
    String groupId = "ID:" + new BigInteger(24 * 8, new Random()).toString(10);
    long highst = 1000000; // Number of Characters, approximately 1.1 MB
    int i = 0;
    int sequence = 1;
    File file = new File("C:/access1.log");
    FileInputStream fin = new FileInputStream(file);
    while ((ch = fin.read()) != -1){
     fileBuffer.append((char) ch);
     i++;
     if(i >= highst){
      TextMessage message = session.createTextMessage();
      message.setStringProperty("JMSXGroupID", groupId);
      message.setIntProperty("JMSXGroupSeq", sequence);
      message.setText(fileBuffer.toString());
      producer.send(message);
      log.debug("sequence Number "+ sequence + " Processed");
      sequence++;
      i = 0;
      fileBuffer = null;
      fileBuffer = new StringBuffer();
     }
    }
   if(fileBuffer != null &&
     fileBuffer.toString().length() > 0){
    TextMessage message = session.createTextMessage();
    message.setStringProperty("JMSXGroupID", groupId);
    message.setIntProperty("JMSXGroupSeq", sequence);
    message.setBooleanProperty("JMS_IBM_Last_Msg_In_Group", true);
    message.setText(fileBuffer.toString());
    producer.send(message);
    log.debug("sequence Number "+ sequence + " Processed as Last Mgs in Grp" );
   }
   connection.close();
   session.close();
   log.debug("Creating Messages and Sending it across - End");
   /************* Here Ends Sending Part ********/
   Thread.sleep(10000); // Just waiting for sometime before receiving starts
   /************* Here Starts Receiving Logic Part ********/
   log.debug("Receiving Logic - Start");
   conn2 = factory.createConnection();
   conn2.start();
   session2 = conn2.createSession(true, Session.AUTO_ACKNOWLEDGE);
   // First trick is to have selector in Consumer to receive last message only
   MessageConsumer consumer = session2.createConsumer(destination, "JMS_IBM_Last_Msg_In_Group=TRUE");
   GroupMessageConsumer grpConsumer = new GroupMessageConsumer();
   consumer.setMessageListener(grpConsumer);
   conn2.setExceptionListener(grpConsumer);
   Thread.sleep(10000); // Just waiting for sometime so that Consumer can listen the message
   log.debug("Receiving Logic - End");
  }catch(Exception exe){
   exe.printStackTrace();
  }finally{
   try {

    connection.close();
    conn2.close();
   } catch (JMSException e) {
    e.printStackTrace();
   }

  }
}

private static class GroupMessageConsumer implements MessageListener, ExceptionListener{
  @Override
  public void onMessage(Message lastMessage) {
   log.debug("lastMessage received - Start");
   FileOutputStream fos = null;
   Session consumerSession = null;
   Connection consumerConnection = null;
   try {
    consumerConnection = factory.createConnection();
    consumerConnection.start();
    consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    MQQueue destination = new MQQueue(localQueueName);
    int groupSize = lastMessage.getIntProperty("JMSXGroupSeq");
    String groupId = lastMessage.getStringProperty("JMSXGroupID");

    boolean failed = false;
    File outFile = new File("C:/outListen.txt");
    fos = new FileOutputStream(outFile);
    log.debug("Reading other messages in Sequence - Start");
    for (int i = 1; (i < groupSize) && !failed; i++) {
     MessageConsumer consumer = consumerSession.createConsumer(destination,
           "JMSXGroupID='" + groupId + "'AND JMSXGroupSeq=" + i);
           TextMessage message = (TextMessage) consumer.receiveNoWait();
           log.debug("Reading Sequence "+ i);
           if (message != null) {
           fos.write((message.getText()).getBytes());
          } else {
            failed = true;
          }
           consumer.close();
    }
    log.debug("Reading Last Message in group ");
    fos.write(((TextMessage) lastMessage).getText().getBytes());
    log.debug("Reading other messages in Sequence - End");
    fos.close();
    if (failed) {
     log.debug("Message Not Processed");
       } else {
          log.debug("Message Processed");
       }
   } catch (JMSException e) {
    e.printStackTrace();
   } catch (FileNotFoundException e) {
    e.printStackTrace();
   } catch (IOException e) {
    e.printStackTrace();
   } finally {
    try {
     session2.commit();
     consumerSession.close();
     consumerConnection.close();
    } catch (JMSException e) {
     e.printStackTrace();
    }
   }


  }
  @Override
  public void onException(JMSException e) {
   e.printStackTrace();

  }

}
}

Streaming Messaging in ActiveMQ

2012-01-18T06:50:00.000-08:00

Many examples on the Web floating around with a copied code snap from apache site. But here is the code which actually sends a Stream Message and More.. it also receives it with MessageListener.

Please let me know suggestions, improvements and questions if you have. thanks.

package com.xyz;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.command.ActiveMQBytesMessage;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.util.ByteSequence;
import org.apache.log4j.Logger;
import com.xyz.AMQConnectorException;
public class ReadFileAndSendStream {
static ActiveMQDirectConnector connector = null;
static ActiveMQConnection connection = null;
static Session session = null;
static Destination sendToQueue = null;
static int ch;
static Map<String, FileOutputStream> map = new HashMap<String, FileOutputStream>();
private static Logger log = Logger.getLogger(ReadFileAndSendStream.class);
public static void main(String args[]){
  try {
   connector = new ActiveMQDirectConnector("tcp://Server.company.com:<PORT>");
   connection =   connector.getConnectionDetails().getConnection();

   session = connector.getConnectionDetails().getSession();
   // Reading a File and sending the stream to ActiveMQ queue
   sendToQueue = session.createQueue("AAD.STREAM.TEST");

   /********* Sending a huge huge huge message to ActiveMQ Queue **************/
   log.debug("Start time to Send :: " + System.currentTimeMillis());
   File file = new File("C:/access1.log");
   DataOutputStream out2 = new DataOutputStream(connection.createOutputStream(sendToQueue));
      FileInputStream fin = null;
      try {
        fin = new FileInputStream(file);
        while ((ch = fin.read()) != -1)
         out2.write((char) ch);
        fin.close();
      } catch (Exception e) {
      e.printStackTrace();
      }
   out2.close();
   session.commit();
   Thread.sleep(10000);
   log.debug("End time to Send :: " + System.currentTimeMillis());

   /********* I should write a consumer ***********/

   log.debug("Start time to read :: " + System.currentTimeMillis());


   MessageConsumer consumer = session.createConsumer(sendToQueue);
   MyConsumer myConsumer = new MyConsumer();
   consumer.setMessageListener(myConsumer);
   connection.setExceptionListener(myConsumer);

   log.debug("End time to read :: " + System.currentTimeMillis());

   /********* Reading a huge huge huge message from ActiveMQ Queue

   DataInputStream in = new DataInputStream(connection.createInputStream(sendToQueue));
   FileWriter fstream = new FileWriter("C:/out.txt");
   BufferedWriter out = new BufferedWriter(fstream);
    try {
         while ((ch = in.read()) != -1)
       out.write((char) ch);
         in.close();
       } catch (Exception e) {
       e.printStackTrace();
       }
     out.close();
     **************/
  } catch (AMQConnectorException e) {
   e.printStackTrace();
  }catch (JMSException e) {
   e.printStackTrace();
  } catch (IOException e) {
   e.printStackTrace();
  } catch (InterruptedException e) {
   e.printStackTrace();
  }

}

private static class MyConsumer implements MessageListener, ExceptionListener{
  @Override
   public void onMessage(Message message) {
    String messageId;
    FileOutputStream fos = null;
    if (message instanceof BytesMessage) {
    try {
     messageId = ((ActiveMQBytesMessage)message).getGroupID();

     if(!map.containsKey(messageId)) {
      log.debug("This is the message ID:" + messageId);
      File outFile = new File("C:/outListen.txt");
      fos = new FileOutputStream(outFile);
      map.put(messageId, fos);
     } else {
      fos = (FileOutputStream) map.get(messageId);
     }
     ByteSequence bs = ((ActiveMQBytesMessage)message).getContent();
     fos.write(bs.getData());

     log.debug("getGroupSequence" +((ActiveMQBytesMessage)message).getGroupSequence());
    }catch (IOException e) {
     e.printStackTrace();
    } finally {

    }
    }else
    {
     try {
      log.debug("This is the message type:" + message.getJMSType() + "\n\n" + message);
      fos = (FileOutputStream) map.remove(((ActiveMQMessage)message).getGroupID() );
      if (fos != null)
      {
       log.debug("Closing File");
       fos.close();
      }
      } catch (JMSException e) {
       e.printStackTrace();
      } catch (IOException e) {
       e.printStackTrace();
      } finally {
       try {
        session.commit();
       } catch (JMSException e) {
        e.printStackTrace();
       }
      }
    }
   }
   @Override
   public void onException(JMSException ex) {
    System.out.println(ex);
    ex.printStackTrace();

   }
}
}

Standalone MQ Queue Consumer Code

2012-01-17T10:39:00.000-08:00

This is small piece of code shows how to connect to MQ with help of MQQueueConnectionFactory and without using JNDI or any context lookups.
Code will create own factory instance and assign the values.
This is easy when you are using Java standalone application without using any application server.

This class also has a consumer example for implementation.

package com.xyz;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.QueueConnection;
import javax.jms.QueueSession;
import javax.jms.Session;
import com.ibm.mq.jms.JMSC;
import com.ibm.mq.jms.MQQueueConnectionFactory;

public class MQDirectQueueConsumer2 {

private String localQueueName;
private String hostName;
private int port;
private String queueManagerName;
private String channelName;

private int sessionProperty = 0;
private QueueConnection mqConnection = null;
private QueueSession mqSession = null;

private MQQueueConnectionFactory mqQCF = null;

/**
* MQDirectQueueConsumer2 constructor. Also calls connectToMQ() which returns QueueConnection
* Object for further action.
* @param hostName
* @param port
* @param queueManagerName
* @param channelName
* @param localQueueName
* @throws Exception
*/
public MQDirectQueueConsumer2(String hostName,
   int port,
   String queueManagerName,
   String channelName,
   String localQueueName) throws Exception {
  this.hostName = hostName;
  this.port = port;
  this.queueManagerName = queueManagerName;
  this.channelName = channelName;
  this.localQueueName = localQueueName;
  connectToMQ();

}
/**
* connectToMQ() creates a connection with queue manager.
* Make sure you have all necessary information to connect to QMGR
* @return QueueConnection
* @throws Exception
*/
private QueueConnection connectToMQ() throws Exception {
  try {
   mqQCF = new MQQueueConnectionFactory();
   mqQCF.setHostName(hostName);
   mqQCF.setPort(port);
   mqQCF.setQueueManager(queueManagerName);
   mqQCF.setTransportType(JMSC.MQJMS_TP_CLIENT_MQ_TCPIP);
   mqQCF.setChannel(channelName);

   mqConnection = mqQCF.createQueueConnection();
   if(sessionProperty == 0){
    mqSession = mqConnection.createQueueSession(true, Session.AUTO_ACKNOWLEDGE);
   }else{
    mqSession = mqConnection.createQueueSession(true, sessionProperty);
   }
   mqConnection.start();
  } catch (JMSException e) {
   e.printStackTrace();
   throw new Exception("Exception in connectToMQ()",e);
  }
  return mqConnection;
}

/**
* startConsumer() registers and starts the consumer for localQueue passed to constructor
* @throws Exception
*/
public void startConsumer() throws Exception{
  if(mqConnection == null){
   connectToMQ();
  }
  try{
    Destination queueToConsume    = mqSession.createQueue(localQueueName);
    MessageConsumer consumer = mqSession.createConsumer(queueToConsume);
    DirectQueueConsumer queueConsumer = new DirectQueueConsumer();
    mqConnection.setExceptionListener(queueConsumer);
    consumer.setMessageListener(queueConsumer);

  }catch(Exception exe){
   throw new Exception("Exception in initialize()"+exe);
  }
}

/**
* Inner Static private class DirectQueueConsumer is
* a MessageListener and ExceptionListener for localQueue
* received message will be passed to onMessage() method provided.
* @author THD
*
*/
private static class DirectQueueConsumer implements MessageListener, ExceptionListener
{
  @Override
  public void onMessage(Message arg0) {
   System.out.println("Received Message :"+ arg0);
   // TODO Application Logic to handle the message

  }
  @Override
  public void onException(JMSException arg0) {
   // TODO Application Logic to handle the message

  }
}
}