Direct Component/SEDA Component/VM Component/Timer Component/Log Component/Cache Component

http://camel.apache.org/direct.html

Direct Component

The direct: component provides direct, synchronous invocation of any consumers when a producer sends a message exchange.
This endpoint can be used to connect existing routes in the same camel context.

URI format

direct:someName[?options]

Where someName can be any string to uniquely identify the endpoint

Options

You can append query options to the URI in the following format, ?option=value&option=value&...

Samples

In the route below we use the direct component to link the two routes together:

from("activemq:queue:order.in")
    .to("bean:orderServer?method=validate")
    .to("direct:processOrder");

from("direct:processOrder")
    .to("bean:orderService?method=process")
    .to("activemq:queue:order.out");

And the sample using spring DSL:

<route>
     <from uri="activemq:queue:order.in"/>
     <to uri="bean:orderService?method=validate"/>
     <to uri="direct:processOrder"/>
  </route>

  <route>
     <from uri="direct:processOrder"/>
     <to uri="bean:orderService?method=process"/>
     <to uri="activemq:queue:order.out"/>
  </route>

See also samples from the SEDA component, how they can be used together.

Note:direct procuder(<to:/>) will use direct consumer<from:/>) 's processor to process its exchange.

===========

http://camel.apache.org/seda.html

SEDA Component

The seda: component provides asynchronous SEDA behavior, so that messages are exchanged on a BlockingQueue and consumers are invoked in a separate thread from the producer.

Note that queues are only visible within a single CamelContext. If you want to communicate across CamelContext instances (for example, communicating between Web applications), see theVM component.

This component does not implement any kind of persistence or recovery, if the VM terminates while messages are yet to be processed. If you need persistence, reliability or distributed SEDA, try using eitherJMS or ActiveMQ.

URI format

seda:someName[?options]

Where someName can be any string that uniquely identifies the endpoint within the currentCamelContext.

You can append query options to the URI in the following format: ?option=value&option=value&…

Options

Use of Request Reply

The SEDA component supports usingRequest Reply, where the caller will wait for the Async route to complete. For instance:

from("mina:tcp://0.0.0.0:9876?textline=true&sync=true").to("seda:input");

from("seda:input").to("bean:processInput").to("bean:createResponse");

In the route above, we have a TCP listener on port 9876 that accepts incoming requests. The request is routed to theseda:input queue. As it is a Request Reply message, we wait for the response. When the consumer on the seda:input

Concurrent consumers

By default, the SEDA endpoint uses a single consumer thread, but you can configure it to use concurrent consumer threads. So instead of thread pools you can use:

from("seda:stageName?concurrentConsumers=5").process(...)

As for the difference between the two, note a thread pool can increase/shrink dynamically at runtime depending on load, whereas the number of concurrent consumers is always fixed.

Thread pools

Be aware that adding a thread pool to a SEDA endpoint by doing something like:

from("seda:stageName").thread(5).process(...)

Can wind up with two BlockQueues: one from the SEDA endpoint, and one from the workqueue of the thread pool, which may not be what you want. Instead, you might wish to configure aDirect endpoint with a thread pool, which can process messages both synchronously and asynchronously. For example:

from("direct:stageName").thread(5).process(...)

You can also directly configure number of threads that process messages on a SEDA endpoint using theconcurrentConsumers

Sample

In the route below we use the SEDA queue to send the request to this async queue to be able to send a fire-and-forget message for further processing in another thread, and return a constant reply in this thread to the original caller.

public void configure() throws Exception {
    from("direct:start")
        // send it to the seda queue that is async
        .to("seda:next")
        // return a constant response
        .transform(constant("OK"));

    from("seda:next").to("mock:result");
}

Here we send a Hello World message and expects the reply to be OK.

Object out = template.requestBody("direct:start", "Hello World");
assertEquals("OK", out);

The "Hello World" message will be consumed from the SEDA queue from another thread for further processing. Since this is from a unit test, it will be sent to amock

Using multipleConsumers

Available as of Camel 2.2

In this example we have defined two consumers and registered them as spring beans.

<!-- define the consumers as spring beans -->
<bean id="consumer1" class="org.apache.camel.spring.example.FooEventConsumer"/>

<bean id="consumer2" class="org.apache.camel.spring.example.AnotherFooEventConsumer"/>

<camelContext xmlns="http://camel.apache.org/schema/spring">
    <!-- define a shared endpoint which the consumers can refer to instead of using url -->
    <endpoint id="foo" uri="seda:foo?multipleConsumers=true"/>
</camelContext>

Since we have specified multipleConsumers=true on the seda foo endpoint we can have those two consumers receive their own copy of the message as a kind of pub-sub style messaging.

As the beans are part of an unit test they simply send the message to a mock endpoint, but notice how we can use @Consume to consume from the seda queue.

public class FooEventConsumer {

    @EndpointInject(uri = "mock:result")
    private ProducerTemplate destination;

    @Consume(ref = "foo")
    public void doSomething(String body) {
        destination.sendBody("foo" + body);
    }

}

Extracting queue information.

If needed, information such as queue size, etc. can be obtained without using JMX in this fashion:

SedaEndpoint seda = context.getEndpoint("seda:xxxx");int

Note:seda producer will add exchanges to queue, then seda consumer will retrieve exchange then process.

===========

http://camel.apache.org/vm.html

VM Component

The vm: component provides asynchronous SEDA behavior, exchanging messages on a BlockingQueue and invoking consumers in a separate thread pool.

This component differs from the SEDA component in that VM supports communication across CamelContext instances - so you can use this mechanism to communicate across web applications (provided thatcamel-core.jar is on the system/boot

VM is an extension to the SEDA component.

URI format

vm:queueName[?options]

Where queueName can be any string to uniquely identify the endpoint within the JVM (or at least within the classloader that loaded camel-core.jar)

You can append query options to the URI in the following format: ?option=value&option=value&...

from("direct:foo").to("vm:bar?concurrentConsumers=5");

from("vm:bar?concurrentConsumers=5").to("file://output");

from("direct:foo").to("vm:bar");

from("vm:bar?concurrentConsumers=5").to("file://output");

Options

See the SEDA component for options and other important usage details as the same rules apply to theVM component.

Samples

In the route below we send exchanges across CamelContext instances to a VM queue namedorder.email:

from("direct:in").bean(MyOrderBean.class).to("vm:order.email");

And then we receive exchanges in some other Camel context (such as deployed in another.war

from("vm:order.email").bean(MyOrderEmailSender.class);

Note:seda component's queues is not static, it's instance queues, so each camel context will differ, vm component's queues is static, so every seda component's instance will have same queues.

 ===============

http://camel.apache.org/timer.html

Timer Component

The timer: component is used to generate message exchanges when a timer fires You can only consume events from this endpoint.

URI format

timer:name[?options]

Where name is the name of the Timer object, which is created and shared across endpoints. So if you use the same name for all your timer endpoints, only one Timer

You can append query options to the URI in the following format, ?option=value&option=value&...

Note: The IN body of the generated exchange is null. So exchange.getIn().getBody() returns null.

Options

Exchange Properties

When the timer is fired, it adds the following information as properties to the Exchange:

Message Headers

When the timer is fired, it adds the following information as headers to the IN message

Sample

To set up a route that generates an event every 60 seconds:

from("timer://foo?fixedRate=true&period=60000").to("bean:myBean?method=someMethodName");

The above route will generate an event and then invoke the someMethodName method on the bean called myBean in the Registry such as JNDI or Spring.

And the route in Spring DSL:

<route>
    <from uri="timer://foo?fixedRate=true&period=60000"/>
    <to uri="bean:myBean?method=someMethodName"/>
  </route>

Firing only once

Available as of Camel 2.8

You may want to fire a message in a Camel route only once, such as when starting the route. To do that you use the repeatCount option as shown:

<route> "timer://foo?repeatCount=1"/> "bean:myBean?method=someMethodName"/> </route>

==========

Log Component

The log: component logs message exchanges to the underlying logging mechanism.

Camel 2.7 or better uses sfl4j which allows you to configure logging via, among others:

• Log4j
• Logback
• JDK Util Logging logging

Camel 2.6 or lower uses commons-logging which allows you to configure logging via, among others:

• Log4j
• JDK Util Logging logging
• SimpleLog - a simple provider in commons-logging

Refer to the commons-logging user guide for a more complete overview of how to use and configure commons-logging.

URI format

log:loggingCategory[?options]

Where loggingCategory is the name of the logging category to use. You can append query options to the URI in the following format, ?option=value&option=value&...

For example, a log endpoint typically specifies the logging level using the level

log:org.apache.camel.example?level=DEBUG

The default logger logs every exchange (regular logging). But Camel also ships with the Throughput logger, which is used whenever the groupSize

Options

note: groupDelay and groupActiveOnly are only applicable when using groupInterval

Formatting

The log formats the execution of exchanges to log lines.
By default, the log uses LogFormatter to format the log output, where LogFormatter

camelContext.getProperties().put(Exchange.LOG_DEBUG_BODY_STREAMS, true);

Regular logger sample

In the route below we log the incoming orders at DEBUG

from("activemq:orders").to("log:com.mycompany.order?level=DEBUG").to("bean:processOrder");

Or using Spring XML to define the route:

<route>
    <from uri="activemq:orders"/>
    <to uri="log:com.mycompany.order?level=DEBUG"/>
    <to uri="bean:processOrder"/>
  </route>

Regular logger with formatter sample

In the route below we log the incoming orders at INFO

from("activemq:orders").
    to("log:com.mycompany.order?showAll=true&multiline=true").to("bean:processOrder");

Throughput logger with groupSize sample

In the route below we log the throughput of the incoming orders at DEBUG

from("activemq:orders").
    to("log:com.mycompany.order?level=DEBUG&groupSize=10").to("bean:processOrder");

Throughput logger with groupInterval sample

This route will result in message stats logged every 10s, with an initial 60s delay and stats should be displayed even if there isn't any message traffic.

from("activemq:orders").
to("log:com.mycompany.order?level=DEBUG&groupInterval=10000&groupDelay=60000&groupActiveOnly=false").to("bean:processOrder");

The following will be logged:

"Received: 1000 new messages, with total 2000 so far. Last group took: 10000 millis which is: 100 messages per second. average: 100"

===================

Cache Component

Available as of Camel 2.1

The cache component enables you to perform caching operations using EHCache as the Cache Implementation. The cache itself is created on demand or if a cache of that name already exists then it is simply utilized with its original settings.

This component supports producer and event based consumer endpoints.

The Cache consumer is an event based consumer and can be used to listen and respond to specific cache activities. If you need to perform selections from a pre-existing cache, use the processors defined for the cache component.

Maven users will need to add the following dependency to their pom.xml

<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-cache</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>

URI format

cache://cacheName[?options]

You can append query options to the URI in the following format, ?option=value&option=#beanRef&...

Options

Sending/Receiving Messages to/from the cache

Message Headers up to Camel 2.7

Message Headers Camel 2.8+

The CamelCacheAdd and CamelCacheUpdate

Cache Producer

Sending data to the cache involves the ability to direct payloads in exchanges to be stored in a pre-existing or created-on-demand cache. The mechanics of doing this involve

• setting the Message Exchange Headers shown above.
• ensuring that the Message Exchange Body contains the message directed to the cache

Cache Consumer

Receiving data from the cache involves the ability of the CacheConsumer to listen on a pre-existing or created-on-demand Cache using an event Listener and receive automatic notifications when any cache activity take place (i.e CamelCacheGet/CamelCacheUpdate/CamelCacheDelete/CamelCacheDeleteAll). Upon such an activity taking place

• an exchange containing Message Exchange Headers and a Message Exchange Body containing the just added/updated payload is placed and sent.
• in case of a CamelCacheDeleteAll operation, the Message Exchange Header CamelCacheKey and the Message Exchange Body are not populated.

Cache Processors

There are a set of nice processors with the ability to perform cache lookups and selectively replace payload content at the

• body
• token
• xpath level

Cache Usage Samples

Example 1: Configuring the cache

from("cache://MyApplicationCache" +
          "?maxElementsInMemory=1000" +
          "&memoryStoreEvictionPolicy=" +
              "MemoryStoreEvictionPolicy.LFU" +
          "&overflowToDisk=true" +
          "&eternal=true" +
          "&timeToLiveSeconds=300" +
          "&timeToIdleSeconds=true" +
          "&diskPersistent=true" +
          "&diskExpiryThreadIntervalSeconds=300")

Example 2: Adding keys to the cache

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
     from("direct:start")
     .setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_ADD))
     .setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson"))
     .to("cache://TestCache1")
   }
};

Example 2: Updating existing keys in a cache

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
     from("direct:start")
     .setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_UPDATE))
     .setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson"))
     .to("cache://TestCache1")
   }
};

Example 3: Deleting existing keys in a cache

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
     from("direct:start")
     .setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_DELETE))
     .setHeader(CacheConstants.CACHE_KEY", constant("Ralph_Waldo_Emerson"))
     .to("cache://TestCache1")
   }
};

Example 4: Deleting all existing keys in a cache

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
     from("direct:start")
     .setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_DELETEALL))
     .to("cache://TestCache1");
    }
};

Example 5: Notifying any changes registering in a Cache to Processors and other Producers

RouteBuilder builder = new RouteBuilder() {
    public void configure() {
     from("cache://TestCache1")
     .process(new Processor() {
        public void process(Exchange exchange)
               throws Exception {
           String operation = (String) exchange.getIn().getHeader(CacheConstants.CACHE_OPERATION);
           String key = (String) exchange.getIn().getHeader(CacheConstants.CACHE_KEY);
           Object body = exchange.getIn().getBody();
           // Do something
        }
     })
   }
};

Example 6: Using Processors to selectively replace payload with cache values

RouteBuilder builder = new RouteBuilder() {
   public void configure() {
     //Message Body Replacer
     from("cache://TestCache1")
     .filter(header(CacheConstants.CACHE_KEY).isEqualTo("greeting"))
     .process(new CacheBasedMessageBodyReplacer("cache://TestCache1","farewell"))
     .to("direct:next");

    //Message Token replacer
    from("cache://TestCache1")
    .filter(header(CacheConstants.CACHE_KEY).isEqualTo("quote"))
    .process(new CacheBasedTokenReplacer("cache://TestCache1","novel","#novel#"))
    .process(new CacheBasedTokenReplacer("cache://TestCache1","author","#author#"))
    .process(new CacheBasedTokenReplacer("cache://TestCache1","number","#number#"))
    .to("direct:next");

    //Message XPath replacer
    from("cache://TestCache1").
    .filter(header(CacheConstants.CACHE_KEY).isEqualTo("XML_FRAGMENT"))
    .process(new CacheBasedXPathReplacer("cache://TestCache1","book1","/books/book1"))
    .process (new CacheBasedXPathReplacer("cache://TestCache1","book2","/books/book2"))
    .to("direct:next");
   }
};

Example 7: Getting an entry from the Cache

from("direct:start")
    // Prepare headers
    .setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_GET))
    .setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")).
    .to("cache://TestCache1").
    // Check if entry was not found
    .choice().when(header(CacheConstants.CACHE_ELEMENT_WAS_FOUND).isNull()).
        // If not found, get the payload and put it to cache
        .to("cxf:bean:someHeavyweightOperation").
        .setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_ADD))
        .setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson"))
        .to("cache://TestCache1")
    .end()
    .to("direct:nextPhase");

Example 8: Checking for an entry in the Cache

Note: The CHECK command tests existence of an entry in the cache but doesn't place a message in the body.

from("direct:start")
    // Prepare headers
    .setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_CHECK))
    .setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")).
    .to("cache://TestCache1").
    // Check if entry was not found
    .choice().when(header(CacheConstants.CACHE_ELEMENT_WAS_FOUND).isNull()).
        // If not found, get the payload and put it to cache
        .to("cxf:bean:someHeavyweightOperation").
        .setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_ADD))
        .setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson"))
        .to("cache://TestCache1")
    .end();

Management of EHCache

EHCache has its own statistics and management from JMX.

Here's a snippet on how to expose them via JMX in a Spring application context:

<bean id="ehCacheManagementService" class="net.sf.ehcache.management.ManagementService" init-method="init" lazy-init="false">
  <constructor-arg>
    <bean class="net.sf.ehcache.CacheManager" factory-method="getInstance"/>
  </constructor-arg>
  <constructor-arg>
    <bean class="org.springframework.jmx.support.JmxUtils" factory-method="locateMBeanServer"/>
  </constructor-arg>
  <constructor-arg value="true"/>
  <constructor-arg value="true"/>
  <constructor-arg value="true"/>
  <constructor-arg value="true"/>
</bean>

Of course you can do the same thing in straight Java:

ManagementService.registerMBeans(CacheManager.getInstance(), mbeanServer, true, true, true, true);

You can get cache hits, misses, in-memory hits, disk hits, size stats this way. You can also change CacheConfiguration parameters on the fly.

Cache replication Camel 2.8+

The Camel Cache component is able to distribute a cache across server nodes using several different replication mechanisms including: RMI, JGroups, JMS and Cache Server.

There are two different ways to make it work:

1. You can configure ehcache.xml

OR

2. You can configure these three options:

• cacheManagerFactory
• eventListenerRegistry
• cacheLoaderRegistry

Configuring Camel Cache replication using the first option is a bit of hard work as you have to configure all caches separately. So in a situation when the all names of caches are not known, using ehcache.xml

The second option is much better when you want to use many different caches as you do not need to define options per cache. This is because replication options are set per CacheManager and per CacheEndpoint. Also it is the only way when cache names are not know at the development phase.

Example: JMS cache replication

JMS replication is the most powerful and secured replication method. Used together with Camel Cache replication makes it also rather simple.
An example is available on a separate page.