Throttling Service Requests Using Camel

This tutorial shows how to use Camel to throttle the sending of messages to a service. This is necessary when you don't want to overload the service. For example, the service might have a published service level agreement (SLA) requiring you to only send a certain number of requests per second or minute. If you're not familiar with Camel, read my blog post on Implementing Pipes and Filters Using Camel.

The source code for this example is available on GitHub: https://github.com/travisdazell/camel-throttler

Let's consider an integration scenario where we receive orders via an incoming file drop. We then pick up the file, process it with some custom Java code, and then place a message on a JMS queue. Here's the Camel route, with a throttler that only allows one message to be sent to the JMS queue every five seconds.

<camelcontext xmlns="http://camel.apache.org/schema/spring">
    <route>
        <from uri="file://incoming" />
  
        <to uri="bean:orderProcessor?method=processIncomingOrder" />
  
        <throttle timePeriodMillis="5000">
            <constant>1</constant>
            <to uri="activemq:queue:ORDERS.INCOMING" />
        </throttle>
    </route>
</camelcontext>

Implementing a Content-Based Router Using Camel

Often times, we're faced with the problem of routing a message at run-time based on the content of the message. Consider the following example. Let's say we receive all incoming orders in a single queue. However, we have two order processing services. The first service processes our normal customer orders, but we have a separate service that processes orders placed internally by employees. How do we solve this problem? A content-based router pattern solves this problem of determining where to route a message at run-time. Here's an example of a Content-Based Router implementing the example I described (the image is credited to http://www.enterpriseintegrationpatterns.com/ContentBasedRouter.html).

Thankfully, it's extremely easy to implement a content-based router pattern in Camel. If you're not familiar with using Camel, refer to my tutorial on Implementing Pipes and Filters Using Camel. It's a different integration pattern, but it walks through the basics of using Camel.

Now that you're familiar with Camel, let's look at how you would define a content-based router in your Camel route.

    <!-- content based router -->
    <route>
        <!-- get message from JMS ORDERS.INCOMING queue -->
        <from uri="activemq:queue:ORDERS.INCOMING">

            <!-- route the order to the correct queue -->
            <choice>
                <when>
                    <simple>${header.type} == 'customer'</simple>
                    <to uri="activemq:queue:ORDERS.CUSTOMER"></to>
                </when>
                <when>
                    <simple>${header.type} == 'employee'</simple>
                    <to uri="activemq:queue:ORDERS.EMPLOYEE"></to>
                </when>
            </choice>
        </from>
    </route>

Implementing Pipes and Filters Using Camel

Pipes and Filters is an integration pattern that solves the problem of having to process a job or task in a series of sub-tasks. If you're not familiar with pipes and filters, read my post on describing the design: Pipes and Filters - Integration Pattern.

This tutorial describes how to implement the pipes and filters integration pattern using Apache Camel. The source code is on GitHub: https://github.com/travisdazell/camel-pipes-and-filters

The key component of this application is the Camel route that's defined in spring-context.xml:

<camelContext xmlns="http://camel.apache.org/schema/spring">

  <!-- pipes and filters example -->
  <route>
 <!-- get message from JMS FOO.BAR queue -->
 <from uri="activemq:queue:FOO.BAR" />

 <!-- use the simple language to transform the input to upper case -->
 <transform>
  <simple>${body.toUpperCase()}</simple>
 </transform>
 
 <!-- route the input to the messageReceiver bean, which will modify the message slightly -->
 <to uri="bean:messageReceiver?method=processMessage"/>
 
 <!-- route the output from the messageReceiver bean to the output console -->
 <to uri="stream:out"/>
  </route>

</camelContext>

This route has four steps/filters. The pipeline begins when a JMS message is posted on the FOO.BAR queue. Next, I'm using Camel to convert the message body to upper-case. The third step in the pipeline sends the result of the upper-case transformation to a Spring bean with an ID of "messageReceiver". Notice the use of "?method=processMessage". This allows us to route the message to the processMessage() method on the messageReceiver bean. The last step of the pipeline is to output the results of the processMessage() method to the console.

In summary, the following items are the main components when implementing the pipes and filters integration pattern in Camel:

1. Defining the Camel route. In this example, I'm using Spring to define the route.
2. The result of the first step is the input to the second step. The result of the second step is the input to the third step. And so on...
3. Use the following Camel URI to route a specific message in a Java class, where {beanId} is substituted with the ID of the Spring bean and {methodName} is the name of the method in the bean.

uri="bean:{beanId}?method={methodName}"

Pipes and Filters - Integration Pattern

Arguably one the simplest integration problems we encounter is having to process a complex job in a sequence of steps. Most of these problems are part of a workflow requirement. Examples include processing an order, approving a purchase order, and applying a credit to an account. Each of these business problems usually requires a series of steps to be performed in a certain sequence. This integration problem can be easily solved using Pipes and Filters. Here's an example of Pipes and Filters being applied to the processing of an incoming order (image is credited to http://www.eaipatterns.com/PipesAndFilters.html).





Essentially, Pipes and Filters is just a breakdown of a requirement into a series of connected steps. Each step is called a "filter" and the transitions between two filters is called a "pipe". When faced with having to process a large job in a series of processing steps, you should consider implementing the Pipes and Filters pattern.

It's important to implement a Pipes and Filters pattern in such a way that makes it easy to add and remove filters at any location in the process. To achieve this flexibility, I recommend using a framework such as Camel. Camel makes it very easy to add, remove, replace, or rearrange the processing steps with very little work and risk to your project. Read my tutorial on Implementing Pipes and Filters Using Camel to learn how Apache Camel makes this easy to implement and maintain.

Camel File Poller and JMS Consumer in Scala

I put the code on  GitHub: https://github.com/travisdazell/Camel-Scala-FilePoller

In this tutorial, I'm going to demonstrate using Apache Camel to write a Scala file poller. When a file arrives in the "incoming" directory, I'm placing a JMS message onto a queue. Lastly, I'll have a JMS consumer written in Scala that will receive the message from the queue. I'll be using ActiveMQ as my JMS message broker. The main focal point in the tutorial is using Apache Camel for creating a file poller. I'm not doing anything special with ActiveMQ and the JMS consumer could just as easily have been written in Java. I just chose to use Scala, because I like it for its expressiveness.

I won't go through the steps of setting up the architecture. I'll cover that in a future tutorial. For now, I assume you have ActiveMQ running, the Camel JARs downloaded, and your Scala development environment ready to go. Now on to the code.

There are only two Scala classes needed. The first component is the Scala class that defines our Camel routes. We have a single route that routes incoming files to a JMS consumer. Here's our Camel route definition in Scala.

import org.apache.camel.scala.dsl.builder.RouteBuilder
import org.apache.camel.model.dataformat.BindyType

class JmsRoutes extends RouteBuilder {
    "file://incoming" --> "activemq:queue:test.queue"
}

The second component in the application is the JMS consumer. This consumer simply blocks, waiting for a file to arrive in a directory named "incoming". Here's the code for the JMS message consumer.

import javax.jms.ConnectionFactory
import org.apache.camel.CamelContext
import org.apache.camel.ConsumerTemplate
import org.apache.camel.impl.DefaultCamelContext
import org.apache.camel.component.jms.JmsComponent
import org.apache.activemq.ActiveMQConnectionFactory

object JmsConsumer {
    def main(args : Array[String]) {
        val connectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false")
        val context = new DefaultCamelContext()

        context.addComponent("activemq", JmsComponent.jmsComponentAutoAcknowledge(connectionFactory))

        context.addRoutes(new JmsRoutes())

        val consumerTemplate = context.createConsumerTemplate()

        context.start()
       
        consumerTemplate.receive("activemq:queue:test.queue")

        println("received message")
       
        context.stop()
    }
}

This shows how simple it is to use Camel for routing messages in a Scala application. I like the expressiveness of the routing syntax. Camel is a very powerful integration framework, and yet it's extremely easy to use.