QueueSender or MessageProducer? Which one I should choose?

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

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>

      <employeeId>23445</employeeId>

    <firstName>Ambarish</firstName>

      <lastName>Deshpande</lastName>

      <role>Lead Java</role>

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

  </entry>

 </map>

<map>

 <entry>

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

<com.thd.eai.dto.EmployeeDTO>

      <employeeId>23411</employeeId>

      <firstName>Paul</firstName>

      <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

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

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”

Making Virtual Topic Consumer “Selector Aware”

<virtualDestinations>

<virtualTopic name="MF.PRICE.CHANGE.TP01" prefix="VTCON.*." selectorAware="true"/>

</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

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.

<broker xmlns="http://activemq.apache.org/schema/core">

    <destinationInterceptors>

      <virtualDestinationInterceptor>

        <virtualDestinations>

          <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>

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

</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

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();
}
}