Complex flows with Apache Camel

At work, we’re mainly integrating services and systems, and since we’re on a constant lookout for new, better technologies, ways to do things easier, make them more sustainable, we’re trying to Usually we use Apache Camel for this task, which is a Swis…At work, we’re mainly integrating services and systems, and since we’re on a constant lookout for new, better technologies, ways to do things easier, make them more sustainable, we’re trying to Usually we use Apache Camel for this task, which is a Swis…

At work, we’re mainly integrating services and systems, and since we’re on a constant lookout for new, better technologies, ways to do things easier, make them more sustainable, we’re trying to Usually we use

Apache Camel for this task, which is a Swiss-knife for integration engineer. What’s more, this tools corresponds well with our approach to integration solutions:
* try to operate on XML messages, so you get the advantage of XPaths, XSL and other benefits,
* don’t convert XML into Java classes back and forth and be worried with problems like XML conversion,
* try to get a simple flow of the process. However, at first sight Apache Camel seems to have some drawbacks mainly in the area of practical solutions ;-). It’s very handy tool if you need to use it as a pipeline with some marginal processing of the data that passes through it. It gets a lot harder to wrap your head around if you consider some branching and intermediate calls to external services. This may be tricky to write properly in Camel’s DSL. Here is a simple pipeline example:

And here the exact scenario we’re discussing: What I’d like to show is the solution to this problem. Well, if you’re using a recent version of Camel this may be easier, a little different, but should still more-or-less work this way. This code is written for Apache Camel 1.4 – a rather antic version, but that’s what we’re forced to use. Oh, well. Ok, enough whining! So, I create a test class to illustrate the case. The route defined in TestRouter class is responsible for:
1. receiving input
2. setting exchange property to a given xpath, which effectively is the name of the first XML element in the input stream
3. than, the input data is sent to three different external services, each of them replies with some fictional data – notice routes a, b and c. The SimpleContentSetter processor is just for responding with a given text.
4. the response from all three services is somehow processed by RequestEnricher bean, which is described below
5. eventually the exchange is logged in specified category Here is some code for this:

public class SimpleTest {
    public void setUp() throws Exception {
        TestRouter tr = new TestRouter();
        ctx.addRoutes(tr);
    }

    @Test
    public void shouldCheck() throws Exception {
        ctx.createProducerTemplate().send("direct:in", getInOut(""));
    }

    class TestRouter extends RouteBuilder {

        public void configure() throws Exception {

            ((ProcessorType) from("direct:in")
                .setProperty("operation").xpath("local-name(/*)", String.class)
                .multicast(new MergeAggregationStrategy())
                .to("direct:a", "direct:b", "direct:c")
                .end()
                .setBody().simple("${in.body}"))
            .bean(RequestEnricher.class, "enrich")
                .to("log:pl.touk.debug");

            from("direct:a").process(new SimpleContentSetter(""));
            from("direct:b").process(new SimpleContentSetter(""));
            from("direct:c").process(new SimpleContentSetter(""));
        }
    }
}

What’s unusual in this code is the fact, that what normally Camel does when you write a piece of DSL like:

.to("direct:a", "direct:b", "direct:c")

is pass input to service

a, than a‘s output gets passed to b, becomes it’s input, than b‘s output becomes c‘s input. The problem being, you loose the output from a and b, not mentioning that you might want to send the same input to all three services. That’s where a little tool called multicast() comes in handy. It offers you the ability to aggregate the outputs of those services. You may even create an AggregationStrategy that will do it the way you like. Below class, MergeAggregationStrategy does exactly that kind of work – it joins outputs from all three services. A lot of info about proper use of AggregationStrategy-ies can be found in this post by Torsten Mielke.

public class MergeAggregationStrategy implements AggregationStrategy {

    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        if (oldExchange.isFailed()) {
            return oldExchange;
        }
        transformMessage(oldExchange.getIn(), newExchange.getIn());
        transformMessage(oldExchange.getOut(), newExchange.getOut());
        return newExchange;
    }

    private void transformMessage(Message oldM, Message newM) {
        String oldBody = oldM.getBody(String.class);
        String newBody = newM.getBody(String.class);
        newM.setBody(oldBody + newBody);
    }

}

However nice this may look (or not), what you’re left with is a mix of multiple XMLs. Normally this won’t do you much good. Better thing to do is to parse this output in some way. What we’re using for this is a Groovy :). Which is great for the task of parsing XML. A lot less verbose than ordinary Java. Let’s assume a scenario, that the aggregated output, currently looking like this:


is to be processed with the following steps in mind:

  • use ** as the result element
  • use attributes param1, param2, param3 from element ** and add it to result element **
public class RequestEnricher {

    public String enrich(@Property(name = "operation") String operation, Exchange ex) {

        use(DOMCategory) {
            def dhl = new groovy.xml.Namespace("http://example.com/common/dhl/schema", 'dhl')
            def pc = new groovy.xml.Namespace("http://example.com/pc/types", 'pc')
            def doc = new XmlParser().parseText(ex.in.body)

            def pcRequest = doc.
            "aaaa" [0]

            ["param1", "param2", "param3"].each() {
                def node = doc.
                '**' [("" + it)][0]
                if (node)
                    pcRequest['@' + it] = node.text()
            }

            gNodeListToString([pcRequest])
        }

    }

    String gNodeListToString(list) {
        StringBuilder sb = new StringBuilder();
        list.each {
            listItem ->
                StringWriter sw = new StringWriter();
            new XmlNodePrinter(new PrintWriter(sw)).print(listItem)
            sb.append(sw.toString());
        }
        return sb.toString();
    }

}

 

What we’re doing here, especially the last line of

enrich method is the conversion to String. There are some problems for Camel if we spit out Groovy objects. The rest is just some Groovy specific ways of manipulating XML. But looking into enrich method’s parameters, there is @Property annotation used, which binds the property assigned earlier in a router code to one of the arguments. That is really cool feature and there are more such annotations:
* @XPath
* @Header
* @Headers and @Properties – gives whole maps of properties or headers This pretty much concludes the subject :) Have fun, and if in doubt, leave a comment with your question!

You May Also Like

ODEO new release

Recently, I released a new version 1.1.34 of ODEO (it's ODE tuned for Oracle and ServiceMix).You can check details (and downloads) here: http://top.touk.pl/confluence/display/top/ODEO.This version contains yet another set of fixes, which drive your BPE...Recently, I released a new version 1.1.34 of ODEO (it's ODE tuned for Oracle and ServiceMix).You can check details (and downloads) here: http://top.touk.pl/confluence/display/top/ODEO.This version contains yet another set of fixes, which drive your BPE...

Log4j and MDC in Grails

Log4j provides very useful feature: MDC - mapped diagnostic context. It can be used to store data in context of current thread. It may sound scary a bit but idea is simple.

My post is based on post http://burtbeckwith.com/blog/?p=521 from Burt Beckwith's excellent blog, it's definitely worth checking if you are interested in Grails.

Short background story...


Suppose we want to do logging our brand new shopping system and we want to have in each log customer's shopping basket number. And our system can be used at once by many users who can perform many transactions, actions like adding items and so on. How can we achieve that? Of course we can add basket number in every place where we do some logging but this task would be boring and error-prone. 

Instead of this we can use MDC to store variable with basket number in map. 

In fact MDC can be treated as map of custom values for current thread that can be used by logger. 


How to do that with Grails?


Using MDC with Grails is quite simple. All we need to do is to create our own custom filter which works for given urls and puts our data in MDC.

Filters in Grails are classes in directory grails-app/conf/* which names end with *Filters.groovy postfix. We can create this class manually or use Grails command: 
grails create-filters info.rnowak.App.Basket

In result class named BasketFilters will be created in grails-app/conf/info/rnowak/UberApp.

Initially filter class looks a little bit empty:
class BasketFilters {
def filters = {
all(controller:'*', action:'*') {
before = {

}
after = { Map model ->

}
afterView = { Exception e ->

}
}
}
}
All we need to do is fill empty closures, modify filter properties and put some data into MDC.

all is the general name of our filter, as class BasketFilters (plural!) can contain many various filters. You can name it whatever you want, for this post let assume it will be named basketFilter

Another thing is change of filter parameters. According to official documentation (link) we can customize our filter in many ways. You can specify controller to be filtered, its actions, filtered urls and so on. In our example you can stay with default option where filter is applied to every action of every controller. If you are interested in filtering only some urls, use uri parameter with expression describing desired urls to be filtered.

Three closures that are already defined in template have their function and they are started in these conditions:

  • before - as name says, it is executed before filtered action takes place
  • after - similarly, it is called after the action
  • afterView - called after rendering of the actions view
Ok, so now we know what are these mysterious methods and when they are called. But what can be done within them? In official Grails docs (link again) under section 7.6.3 there is a list of properties that are available to use in filter.

With that knowledge, we can proceed to implementing filter.

Putting something into MDC in filter


What we want to do is quite easy: we want to retrieve basket number from parameters and put it into MDC in our filter:
class BasketFilters {
def filters = {
basketFilter(controller:'*', action:'*') {
before = {
MDC.put("basketNumber", params.basketNumber ?: "")
}
after = { Map model ->
MDC.remove("basketNumber")
}
}
}
}

We retrieve basket number from Grails params map and then we put in map under specified key ("basketNumber" in this case), which will be later used in logger conversion pattern. It is important to remove custom value after processing of action to avoid leaks.

So we are putting something into MDC. But how make use of it in logs?


We can refer to custom data in MDC in conversion patter using syntax: %X{key}, where key is our key we used in filter to put data, like:
def conversionPattern = "%d{yyyy-MM-dd HH:mm:ss} %-5p %t [%c{1}] %X{basketNumber} - %m%n"


And that's it :) We've put custom data in log4j MDC and successfully used it in logs to display interesting values.

Atom Feeds with Spring MVC

How to add feeds (Atom) to your web application with just two classes?
How about Spring MVC?

Here are my assumptions:
  • you are using Spring framework
  • you have some entity, say “News”, that you want to publish in your feeds
  • your "News" entity has creationDate, title, and shortDescription
  • you have some repository/dao, say "NewsRepository", that will return the news from your database
  • you want to write as little as possible
  • you don't want to format Atom (xml) by hand
You actually do NOT need to use Spring MVC in your application already. If you do, skip to step 3.


Step 1: add Spring MVC dependency to your application
With maven that will be:
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-webmvc</artifactId>
    <version>3.1.0.RELEASE</version>
</dependency>

Step 2: add Spring MVC DispatcherServlet
With web.xml that would be:
<servlet>
    <servlet-name>dispatcher</servlet-name>
    <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
    <init-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>classpath:spring-mvc.xml</param-value>
    </init-param>
    <load-on-startup>1</load-on-startup>
</servlet>
<servlet-mapping>
    <servlet-name>dispatcher</servlet-name>
    <url-pattern>/feed</url-pattern>
</servlet-mapping>
Notice, I set the url-pattern to “/feed” which means I don't want Spring MVC to handle any other urls in my app (I'm using a different web framework for the rest of the app). I also give it a brand new contextConfigLocation, where only the mvc configuration is kept.

Remember that, when you add a DispatcherServlet to an app that already has Spring (from ContextLoaderListener for example), your context is inherited from the global one, so you should not create beans that exist there again, or include xml that defines them. Watch out for Spring context getting up twice, and refer to spring or servlet documentation to understand what's happaning.

Step 3. add ROME – a library to handle Atom format
With maven that is:
<dependency>
    <groupId>net.java.dev.rome</groupId>
    <artifactId>rome</artifactId>
    <version>1.0.0</version>
</dependency>

Step 4. write your very simple controller
@Controller
public class FeedController {
    static final String LAST_UPDATE_VIEW_KEY = "lastUpdate";
    static final String NEWS_VIEW_KEY = "news";
    private NewsRepository newsRepository;
    private String viewName;

    protected FeedController() {} //required by cglib

    public FeedController(NewsRepository newsRepository, String viewName) {
        notNull(newsRepository); hasText(viewName);
        this.newsRepository = newsRepository;
        this.viewName = viewName;
    }

    @RequestMapping(value = "/feed", method = RequestMethod.GET)        
    @Transactional
    public ModelAndView feed() {
        ModelAndView modelAndView = new ModelAndView();
        modelAndView.setViewName(viewName);
        List<News> news = newsRepository.fetchPublished();
        modelAndView.addObject(NEWS_VIEW_KEY, news);
        modelAndView.addObject(LAST_UPDATE_VIEW_KEY, getCreationDateOfTheLast(news));
        return modelAndView;
    }

    private Date getCreationDateOfTheLast(List<News> news) {
        if(news.size() > 0) {
            return news.get(0).getCreationDate();
        }
        return new Date(0);
    }
}
And here's a test for it, in case you want to copy&paste (who doesn't?):
@RunWith(MockitoJUnitRunner.class)
public class FeedControllerShould {
    @Mock private NewsRepository newsRepository;
    private Date FORMER_ENTRY_CREATION_DATE = new Date(1);
    private Date LATTER_ENTRY_CREATION_DATE = new Date(2);
    private ArrayList<News> newsList;
    private FeedController feedController;

    @Before
    public void prepareNewsList() {
        News news1 = new News().title("title1").creationDate(FORMER_ENTRY_CREATION_DATE);
        News news2 = new News().title("title2").creationDate(LATTER_ENTRY_CREATION_DATE);
        newsList = newArrayList(news2, news1);
    }

    @Before
    public void prepareFeedController() {
        feedController = new FeedController(newsRepository, "viewName");
    }

    @Test
    public void returnViewWithNews() {
        //given
        given(newsRepository.fetchPublished()).willReturn(newsList);
        
        //when
        ModelAndView modelAndView = feedController.feed();
        
        //then
        assertThat(modelAndView.getModel())
                .includes(entry(FeedController.NEWS_VIEW_KEY, newsList));
    }

    @Test
    public void returnViewWithLastUpdateTime() {
        //given
        given(newsRepository.fetchPublished()).willReturn(newsList);

        //when
        ModelAndView modelAndView = feedController.feed();

        //then
        assertThat(modelAndView.getModel())
                .includes(entry(FeedController.LAST_UPDATE_VIEW_KEY, LATTER_ENTRY_CREATION_DATE));
    }

    @Test
    public void returnTheBeginningOfTimeAsLastUpdateInViewWhenListIsEmpty() {
        //given
        given(newsRepository.fetchPublished()).willReturn(new ArrayList<News>());

        //when
        ModelAndView modelAndView = feedController.feed();

        //then
        assertThat(modelAndView.getModel())
                .includes(entry(FeedController.LAST_UPDATE_VIEW_KEY, new Date(0)));
    }
}
Notice: here, I'm using fest-assert and mockito. The dependencies are:
<dependency>
 <groupId>org.easytesting</groupId>
 <artifactId>fest-assert</artifactId>
 <version>1.4</version>
 <scope>test</scope>
</dependency>
<dependency>
 <groupId>org.mockito</groupId>
 <artifactId>mockito-all</artifactId>
 <version>1.8.5</version>
 <scope>test</scope>
</dependency>

Step 5. write your very simple view
Here's where all the magic formatting happens. Be sure to take a look at all the methods of Entry class, as there is quite a lot you may want to use/fill.
import org.springframework.web.servlet.view.feed.AbstractAtomFeedView;
[...]

public class AtomFeedView extends AbstractAtomFeedView {
    private String feedId = "tag:yourFantastiSiteName";
    private String title = "yourFantastiSiteName: news";
    private String newsAbsoluteUrl = "http://yourfanstasticsiteUrl.com/news/"; 

    @Override
    protected void buildFeedMetadata(Map<String, Object> model, Feed feed, HttpServletRequest request) {
        feed.setId(feedId);
        feed.setTitle(title);
        setUpdatedIfNeeded(model, feed);
    }

    private void setUpdatedIfNeeded(Map<String, Object> model, Feed feed) {
        @SuppressWarnings("unchecked")
        Date lastUpdate = (Date)model.get(FeedController.LAST_UPDATE_VIEW_KEY);
        if (feed.getUpdated() == null || lastUpdate != null || lastUpdate.compareTo(feed.getUpdated()) > 0) {
            feed.setUpdated(lastUpdate);
        }
    }

    @Override
    protected List<Entry> buildFeedEntries(Map<String, Object> model, HttpServletRequest request, HttpServletResponse response) throws Exception {
        @SuppressWarnings("unchecked")
        List<News> newsList = (List<News>)model.get(FeedController.NEWS_VIEW_KEY);
        List<Entry> entries = new ArrayList<Entry>();
        for (News news : newsList) {
            addEntry(entries, news);
        }
        return entries;
    }

    private void addEntry(List<Entry> entries, News news) {
        Entry entry = new Entry();
        entry.setId(feedId + ", " + news.getId());
        entry.setTitle(news.getTitle());
        entry.setUpdated(news.getCreationDate());
        entry = setSummary(news, entry);
        entry = setLink(news, entry);
        entries.add(entry);
    }

    private Entry setSummary(News news, Entry entry) {
        Content summary = new Content();
        summary.setValue(news.getShortDescription());
        entry.setSummary(summary);
        return entry;
    }

    private Entry setLink(News news, Entry entry) {
        Link link = new Link();
        link.setType("text/html");
        link.setHref(newsAbsoluteUrl + news.getId()); //because I have a different controller to show news at http://yourfanstasticsiteUrl.com/news/ID
        entry.setAlternateLinks(newArrayList(link));
        return entry;
    }

}

Step 6. add your classes to your Spring context
I'm using xml approach. because I'm old and I love xml. No, seriously, I use xml because I may want to declare FeedController a few times with different views (RSS 1.0, RSS 2.0, etc.).

So this is the forementioned spring-mvc.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"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">

    <bean class="org.springframework.web.servlet.view.ContentNegotiatingViewResolver">
        <property name="mediaTypes">
            <map>
                <entry key="atom" value="application/atom+xml"/>
                <entry key="html" value="text/html"/>
            </map>
        </property>
        <property name="viewResolvers">
            <list>
                <bean class="org.springframework.web.servlet.view.BeanNameViewResolver"/>
            </list>
        </property>
    </bean>

    <bean class="eu.margiel.pages.confitura.feed.FeedController">
        <constructor-arg index="0" ref="newsRepository"/>
        <constructor-arg index="1" value="atomFeedView"/>
    </bean>

    <bean id="atomFeedView" class="eu.margiel.pages.confitura.feed.AtomFeedView"/>
</beans>

And you are done.

I've been asked a few times before to put all the working code in some public repo, so this time it's the other way around. I've describe things that I had already published, and you can grab the commit from the bitbucket.

Hope that helps.