Maven support in IntelliJ IDEA (screencast)

The screencast demonstrates Maven integration features available in IntelliJ IDEA. Apache Camel maven archetype is used to generate the application. The main features are:

• Project setup automation and automatic dependency resolution
• Editor assistance and autocompletion
• Refactorings
• Dependency graph
• Lifecycle management using the tooling window

My Complains To JMS

JMS is a mess! I do not understand, why it should be so complicated!?

Take a look at this example:

public boolean sendMessage() throws Exception {
      Connection connection = null;
      InitialContext initialContext = null;
      try {
         //Step 1. Create an initial context to perform the JNDI lookup.
         initialContext = getContext(0);

         //Step 2. Perfom a lookup on the queue
         Queue queue = (Queue) initialContext.lookup("/queue/exampleQueue");

         //Step 3. Perform a lookup on the Connection Factory
         ConnectionFactory cf = (ConnectionFactory) 
                                initialContext.lookup("/ConnectionFactory");

         //Step 4.Create a JMS Connection
         connection = cf.createConnection();

         //Step 5. Create a JMS Session
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

         //Step 6. Create a JMS Message Producer
         MessageProducer producer = session.createProducer(queue);

         //Step 7. Create a Text Message
         TextMessage message = session.createTextMessage("This is a text message");
         
         System.out.println("Sent message: " + message.getText());

         //Step 8. Send the Message
         producer.send(message);

         //Step 9. Create a JMS Message Consumer
         MessageConsumer messageConsumer = session.createConsumer(queue);

         //Step 10. Start the Connection
         connection.start();

         //Step 11. Receive the message
         TextMessage messageReceived = (TextMessage) messageConsumer.receive(5000);

         System.out.println("Received message: " + messageReceived.getText());

         return true;
      } finally {
         //Step 12. Be sure to close our JMS resources!
         if (initialContext != null) {
            initialContext.close();
         }
         if(connection != null) {
            connection.close();
         }
      }
   }

You have to make 12 (!!!) steps to send and receive the message correctly. I think that most of the steps could be amended so that JMS spec would provide some default behavior.

Luckily, you can write same code with Camel:

from("bean:myBean?methodName=send").to("jms:queue:exampleQueue");
   from("jms:queue:exampleQueue").to("bean:myBean?methodName=receive");

Apache Camel: Route Synchronization

I had a case to implement with Apache Camel, where the application reads XML files produced by an external system, imports the data into Oracle database, and in a while, it should process the data which has been reviewed by some business person.



Here's the simplified main code implementing the process:

package my.app;

import org.apache.camel.Processor;
import org.apache.camel.Exchange;
import org.apache.camel.spring.SpringRouteBuilder;

public class MyRouteBuilder extends SpringRouteBuilder {

 protected void configureImportRoute() {
   String filesUri = "file:files/payments" + 
                     "?initialDelay=3000" + 
                     "&delay=1000" + 
                     "&useFixedDelay=true" +
                     "&include=.*[.]xml" +
                     "&move=backup/${file:name.noext}-${date:now:yyyyMMddHHmmssSSS}.xml" +
                     "&moveFailed=files/${file:name.noext}-${date:now:yyyyMMddHHmmssSSS}.xml.error";

   from(filesUri).convertBodyTo(MyBean.class).transacted().to("importProcessor");

   String executionTriggerUri = "timer:executionTimer"
                              + "?fixedRate=true"
                              + "&daemon=true"
                              + "&delay=3000"
                              + "&period=3000";

   from(executionTriggerUri)
    .pipeline("bean:myDao?method=listItemsForExecution")
    .to("executioncProcessor");
}

What I wanted to do is to synchronize the routes. There could be several ways for doing this: using Camel BAM module, using a polling consumer and checking a flag. I came up with a solution that logically looks almost the same as Claus proposed, but instead of some explicit flag I used thread synchronization trick. Here's how the proof of concept code looks like.

The route builder:

package my.dummy.app;

import org.apache.camel.Processor;
import org.apache.camel.Exchange;
import org.apache.camel.spring.SpringRouteBuilder;

public class Route extends SpringRouteBuilder {
  public void configure() {
    from("timer:TriggerA?delay=100&period=1").to("A");
    from("timer:TriggerB?delay=100&period=1").to("B");
    from("timer:TriggerC?delay=100&period=1").to("C");
  }
}

Here we can see 3 concurrent routes being executed via timer component periodically.

I created 3 dummy processors that are emulating the business logic processors in the real application. Its function is just to create a delay, using Thread.sleep() with random period, and logs a debug message.

package my.dummy.app;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.log4j.Logger;
import java.util.Random;

public class A /*B*/ /*C*/ implements Processor {
  Random r = new Random();
  private static final Logger log = Logger.getLogger(Processor.class);
  public void process(Exchange exchange) throws Exception {
    Thread.sleep(r.nextInt(1000));
    log.info("processing " + exchange + " in " + getClass().getName());
  }
}

The Spring configuration for this dummy application is as follows:

<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://www.springframework.org/schema/beans" xsi:schemalocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

<camelcontext id="importing" xmlns="http://camel.apache.org/schema/spring">
  <packagescan>
     <package>my.dummy.app</package>
  </packagescan>
</camelcontext>

<bean class="my.dummy.app.A" id="A">
<bean class="my.dummy.app.B" id="B">
<bean class="my.dummy.app.C" id="C">
</beans>

I noticed that there's a DelegateProcessor in Camel that could be used to wrap the real processors. So I can use it to synchronize the routes like this:

package my.dummy.app;

import org.apache.log4j.Logger;

import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.processor.DelegateProcessor;

public class RouteSynchronizer extends DelegateProcessor {
private Logger log = Logger.getLogger(RouteSynchronizer.class);
private final static Object sync = new Object();

public void process(Exchange exchange) throws Exception {
  synchronized (sync) {
    log.debug("begin exchange processing by " + Thread.currentThread().getName());
    super.process(exchange);
    try {
      if (exchange.isFailed()) {
        throw new RuntimeCamelException(exchange.getException());
      }
    } finally {
      log.debug("end exchange processing by " + Thread.currentThread().getName());
    }
  }
}

}

Now I can wrap my original processor beans my the brand new delegate processor as follows:


<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://www.springframework.org/schema/beans" xsi:schemalocation="</p><p>http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd</p><p>http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

<camelcontext id="importing" xmlns="http://camel.apache.org/schema/spring">
<packagescan>
<package>my.dummy.app</package>
</packagescan>


<bean class="my.dummy.app.RouteSynchronizer" id="A"/>
<property name="processor"/>
</bean>

<bean class="my.dummy.app.A"/>
<bean class="my.dummy.app.RouteSynchronizer" id="B"/>
<bean class="my.dummy.app.B"/>
<bean class="my.dummy.app.RouteSynchronizer" id="C"/>
<bean class="my.dymmy.app.C"/>

<camelcontext>
</beans>

With this solution there's a performance penalty due to synchronization but this is not critical for my application at least and confirms to the requirements.

Here's how the dummy application log looks like:

12-08-2009 15:06:01,970 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerB?delay=100&period=1
12-08-2009 15:06:02,298 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.B
12-08-2009 15:06:02,313 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerB?delay=100&period=1
12-08-2009 15:06:02,313 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerC?delay=100&period=1
12-08-2009 15:06:03,032 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.C
12-08-2009 15:06:03,032 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerC?delay=100&period=1
12-08-2009 15:06:03,032 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerA?delay=100&period=1
12-08-2009 15:06:03,173 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.A
12-08-2009 15:06:03,173 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerA?delay=100&period=1
12-08-2009 15:06:03,173 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerB?delay=100&period=1
12-08-2009 15:06:03,579 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.B
12-08-2009 15:06:03,579 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerB?delay=100&period=1
12-08-2009 15:06:03,579 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerC?delay=100&period=1
12-08-2009 15:06:04,251 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.C
12-08-2009 15:06:04,251 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerC?delay=100&period=1
12-08-2009 15:06:04,251 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerA?delay=100&period=1
12-08-2009 15:06:04,626 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.A
12-08-2009 15:06:04,626 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerA?delay=100&period=1
12-08-2009 15:06:04,626 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerB?delay=100&period=1
12-08-2009 15:06:05,251 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.B
12-08-2009 15:06:05,251 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerB?delay=100&period=1
12-08-2009 15:06:05,251 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerC?delay=100&period=1
12-08-2009 15:06:05,688 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.C
12-08-2009 15:06:05,688 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerC?delay=100&period=1
12-08-2009 15:06:05,688 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerA?delay=100&period=1
12-08-2009 15:06:05,782 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.A
12-08-2009 15:06:05,782 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerA?delay=100&period=1
12-08-2009 15:06:05,782 DEBUG RouteSynchronizer - begin exchange processing by timer://TriggerB?delay=100&period=1
12-08-2009 15:06:06,298 INFO  Processor - processing Exchange[Message: [Body is null]] in ee.hansa.markets.ktp.payments.tmp.B
12-08-2009 15:06:06,298 DEBUG RouteSynchronizer - end exchange processing by timer://TriggerB?delay=100&period=1

From the log we can see that this concept actually works - no crossing between the routes appeared.

What could be nice to have is the same support in the DSL, like:

from("...").synchronize().to("processorA");
from("...").synchronize().to("processorA");

Apache Camel: The File Componenet

A very common task for so-called batch processes, is to read some files in some directory, so some processing for these files (whereas it might be required to do some data instrumentation with the data from various sources), and store the data in the database.

Here's a file source endpoint definition:

String uri = "file:files" +
"?initialDelay=3000" +
"&delay=1000" +
"&useFixedDelay=true" +
"&include=.*[.]xml" +
"&move=backup/${file:name.noext}-${date:now:yyyyMMddHHmmssSSS}.xml" +
"&moveFailed=files/${file:name.noext}-${date:now:yyyyMMddHHmmssSSS}.xml.error";

So consider this route:

from(uri).process(someProcessor).to("someBean");

With file endpoint definition above, it means, that with consume the files from some directory called "files", the initial delay for the files polling is 3 seconds, and we will poll with fixed intervals - 1 second, filtering out non-xml files.

In addition to that, if a file is processed successfully, it is being moved to a "backup" directory appending a time stamp in the file's name.

A new feature is a moveFailure attribute, which in case of failure in someProcessor or in the target endpoint ("someBean") allows you to handle the failed files, e.g. renaming or moving to another directory. At the time of writing the feature is available in snapshot version of Camel.

FIX is digging the ESB world

After Apache Synapse got its FIX integration and Apache Camel followed with the FIX endpoint, we can see now same integration done for Mule ESB. It is interesting, that all the products use QuickFIX/J to implement the integration.

What is it? Is it a hype for the ESB world, to include FIX - there must be a trigger for this kind of features. Do the OMS really demand FIX more and more?

camel-quickfix update: Simple Acceptor Usage Scenario

The simplest usage scenario of a Quickfix/J Acceptor in Apache Camel would be as follows:

from("quickfix-server:server.cfg").
to("bean:someBean");

The bean will receive a class instance derived from quickfix.Message and the user can do what ever he wants with this model. However, very often you may want to map the FIX representation onto your own object model. For this purpose camel-bindy is a nice tool, where you can describe how the FIX string should be mapped onto the object model.

So now we have a bit improved version of the scenario:

DataFormat bindy = ...
from("quickfix-server:server.cfg")
.unmarshall(bindy)
.to("bean:someBean");

So, now the target bean will receive the message which is defined in your own object model.

This is the simplest scenario, but there's more to come.

camel-quickfix: Working with FIX messages using Apache Camel

Apache Camel

Apache Camel is a powerful open source integration framework based on known Enterprise Integration Patterns. The very nice feature in Camel is that it provides many out-of-the-box components to be reused for any kind of data sources or services.


FIX (Financial Information eXchange)

FIX is an open specification intended to streamline electronic communications in the financial industry. FIX is governed by fixprotocol.org. FIX messages are constructed with name value pairs that contains a standard header (which contains a message type property), a body and a trailer.

Here's a sample FIX message: Buy 1000 DELL @ MKT

8=FIX.4.0#9=105#35=D#34=2#49=BANZAI#52=20080711-06:42:26#56=SYNAPSE#11=1215758546278#21=1#38=1000#40=1#54=1#55=DELL#59=0#10=253

Thousands of small investment firms use FIX protocol and benefit from it as its specification is open and free. FIX supports derivatives like equity, options, FOREX (Foreign Exchange) and fixed incomes, like bonds.


Apache Camel and FIX

For the enterprise version of Apache Camel, the FUSE mediation router, there exists a component that should enable integration via FIX protocol using Camel. The current implementation depends on the Artix Data Services, which is not open-sourced, and also there are some requests for enhancements. There is also a JIRA issue: #CAMEL-1350.

Currently, we also have a project which requires integration with NASDAQ INET platform using FIX protocol, and Apache Camel would be a great tool for that, if it only had a FIX component without Artix DS dependency. For this purpose I have a small project at Google Project site: camel-quickfix.

There's a TODO list at the moment, which includes message normalization and component orchestration issues. But also there's a number of scenarios that are required to be supported in that component (or with its aid). For instance, if our application is an initiator, then in Camel DSL we write:

...to("camel-initiator:config.cfg[?params]

But, in FIX communication, the acceptor side is responding with ExecutionReport message. So, we have an issue here, how should it be supported? Either to request some sort a callback mechanism into Camel DSL, or force the QuickfixApplication to reuse the Camel context to route the response message back. This issue will go to another post.

Code Kata: The Business Rules

I'm thinking of an application for JBoss Drools. We are given a task to design a system for order management (OMS), which will sit behind a FIX gateway and serve several internal and external systems. One thing is for sure - we are to take a challenge to organize and maintain a large set of business rules and this brings up some memories.

In 2004 a colleague told me about code kata. These are the good exercises to do when learning a new language, or just empirically solve some common problems. One of the puzzlers sit in my mind permanently since then - The Business Rules Kata, which is about the hard to maintain complex business rules.

How can you tame these wild business rules? How can you build a system that will be flexible enough to handle both the complexity and the need for change? And how can you do it without condemming yourself to years and years of mindless support?

We face all these problems daily while coding the business logic for our banking system. One of the approaches some developers have taken is that

the if-the statements can be avoided while using appropriate OOP techniques

The problem is that sometimes while avoiding those if-statements the code becomes over-engineered due to a large number of classes. And it is still doesn't solve the problem of quick changes that are very hard to implement in a big organization with the "release process".

Recently I've found a discussion at joelonsoftware.com. It is dated 2005 but I think not much have changed since then. I think that most of the developers are still fighting the complexity of the business rules and in future it will still be a challenge.

My next step is to create a demo using JBoss Drools, QuickFIX/J. Apache Camel can be used to integrate both into one flow.

Also what is really cool about Drools is that it would allow our clients to maintain the order validation rules apart of the overall release process using Guvnor. So the changes in order validation logic could be introduced rapidly.