A standard route takes its input from just a single endpoint, using the
from( syntax in the Java DSL. But
what if you need to define multiple inputs for your route? Apache Camel provides several
alternatives for specifying multiple inputs to a route. The approach to take depends on
whether you want the exchanges to be processed independently of each other or whether you
want the exchanges from different inputes to be combined in some way (in which case, you
should use the Content enricher pattern).EndpointURL)
The simplest way to specify multiple inputs is using the multi-argument form of the
from() DSL command, for example:
from("URI1", "URI2", "URI3").to("DestinationUri");Or you can use the following equivalent syntax:
from("URI1").from("URI2").from("URI3").to("DestinationUri");In both of these examples, exchanges from each of the input endpoints,
URI1, URI2, and
URI3, are processed independently of each other and in separate
threads. In fact, you can think of the preceding route as being equivalent to the following
three separate routes:
from("URI1").to("DestinationUri");
from("URI2").to("DestinationUri");
from("URI3").to("DestinationUri");For example, you might want to merge incoming messages from two different messaging systems and process them using the same route. In most cases, you can deal with multiple inputs by dividing your route into segments, as shown in Figure 2.5.
The initial segments of the route take their inputs from some external queues—for
example, activemq:Nyse and activemq:Nasdaq—and send the
incoming exchanges to an internal endpoint, InternalUrl. The
second route segment merges the incoming exchanges, taking them from the internal endpoint
and sending them to the destination queue, activemq:USTxn. The
InternalUrl is the URL for an endpoint that is intended only
for use within a router application. The following types of endpoints
are suitable for internal use:
The main purpose of these endpoints is to enable you to glue together different segments of a route. They all provide an effective way of merging multiple inputs into a single route.
![[Note]](imagesdb/note.gif) | Note |
|---|---|
The direct, SEDA, and VM components work only for the InOnly exchange pattern. If one of your inputs requires an InOut exchange pattern, refer to the Content enricher pattern instead. |
The direct component provides the simplest mechanism for linking together routes. The
event model for the direct component is synchronous, so that subsequent
segments of the route run in the same thread as the first segment. The general format of a
direct URL is direct:, where the endpoint
ID, EndpointIDEndpointID, is simply a unique alphanumeric string that
identifies the endpoint instance.
For example, if you want to take the input from two message queues,
activemq:Nyse and activemq:Nasdaq, and merge them into a single
message queue, activemq:USTxn, you can do this by defining the following set of
routes:
from("activemq:Nyse").to("direct:mergeTxns");
from("activemq:Nasdaq").to("direct:mergeTxns");
from("direct:mergeTxns").to("activemq:USTxn");Where the first two routes take the input from the message queues, Nyse and
Nasdaq, and send them to the endpoint, direct:mergeTxns. The
last queue combines the inputs from the previous two queues and sends the combined message
stream to the activemq:USTxn queue.
The implementation of the direct endpoint behaves as follows: whenever an exchange
arrives at a producer endpoint (for example, to("direct:mergeTxns")), the
direct endpoint passes the exchange directly to all of the consumers endpoints that have the
same endpoint ID (for example, from("direct:mergeTxns")). Direct endpoints can
only be used to communicate between routes that belong to the same CamelContext
in the same Java virtual machine (JVM) instance.
| Note |
|---|---|
If you connect multiple consumers to a direct endpoint, every exchange that passes through the endpoint will be processed by all of the attached consumers (multicast). Hence, each exchange would be processed more than once. If you don't want this to happen, consider using a SEDA endpoint, which behaves differently. |
The SEDA component provides an alternative mechanism for linking together routes. You can use it in a similar way to the direct component, but it has a different underlying event and threading model, as follows:
Processing of a SEDA endpoint is not synchronous. That is, when you send an exchange to a SEDA producer endpoint, control immediately returns to the preceding processor in the route.
SEDA endpoints contain a queue buffer (of
java.util.concurrent.BlockingQueuetype), which stores all of the incoming exchanges prior to processing by the next route segment.Each SEDA consumer endpoint creates a thread pool (the default size is 5) to process exchange objects from the blocking queue.
The SEDA component supports the competing consumers pattern, which guarantees that each incoming exchange is processed only once, even if there are multiple consumers attached to a specific endpoint.
One of the main advantages of using a SEDA endpoint is that the routes can be more responsive, owing to the built-in consumer thread pool. The stock transactions example can be re-written to use SEDA endpoints instead of direct endpoints, as follows:
from("activemq:Nyse").to("seda:mergeTxns");
from("activemq:Nasdaq").to("seda:mergeTxns");
from("seda:mergeTxns").to("activemq:USTxn");The main difference between this example and the direct example is that when using SEDA,
the second route segment (from seda:mergeTxns to activemq:USTxn)
is processed by a pool of five threads.
| Note |
|---|---|
There is more to SEDA than simply pasting together route segments. The staged event-driven architecture (SEDA) encompasses a design philosophy for building more manageable multi-threaded applications. The purpose of the SEDA component in Fuse Mediation Router is simply to enable you to apply this design philosophy to your applications. For more details about SEDA, see http://www.eecs.harvard.edu/~mdw/proj/seda/. |
The VM component is very similar to the SEDA endpoint. The only difference is that,
whereas the SEDA component is limited to linking together route segments from within the
same CamelContext, the VM component enables you to link together routes from
distinct Fuse Mediation Router applications, as long as they are running within the same Java virtual
machine.
The stock transactions example can be re-written to use VM endpoints instead of SEDA endpoints, as follows:
from("activemq:Nyse").to("vm:mergeTxns");
from("activemq:Nasdaq").to("vm:mergeTxns");And in a separate router application (running in the same Java VM), you can define the second segment of the route as follows:
from("vm:mergeTxns").to("activemq:USTxn");The content enricher pattern defines a fundamentally different way of dealing with multiple inputs to a route. When an exchange enters the enricher processor, the enricher contacts an external resource to retrieve information, which is then added to the original message. In this pattern, the external resource effectively represents a second input to the message.
For example, suppose you are writing an application that processes credit requests.
Before processing a credit request, you need to augment it with the data that assigns a
credit rating to the customer, where the ratings data is stored in a file in the directory,
src/data/ratings. You can combine the incoming credit request with data from
the ratings file using the pollEnrich() pattern and a
GroupedExchangeAggregationStrategy aggregation strategy, as follows:
from("jms:queue:creditRequests")
.pollEnrich("file:src/data/ratings?noop=true", new GroupedExchangeAggregationStrategy())
.bean(new MergeCreditRequestAndRatings(), "merge")
.to("jms:queue:reformattedRequests");Where the GroupedExchangeAggregationStrategy class is a standard
aggregation strategy from the org.apache.camel.processor.aggregate package that
adds each new exchange to a java.util.List instance and stores the resulting
list in the Exchange.GROUPED_EXCHANGE exchange property. In this case, the list
contains two elements: the original exchange (from the creditRequests JMS
queue); and the enricher exchange (from the file endpoint).
To access the grouped exchange, you can use code like the following:
public class MergeCreditRequestAndRatings {
public void merge(Exchange ex) {
// Obtain the grouped exchange
List<Exchange> list = ex.getProperty(Exchange.GROUPED_EXCHANGE, List.class);
// Get the exchanges from the grouped exchange
Exchange originalEx = list.get(0);
Exchange ratingsEx = list.get(1);
// Merge the exchanges
...
}
}An alternative approach to this application would be to put the merge code directly into the implementation of the custom aggregation strategy class.
For more details about the content enricher pattern, see Content Enricher.
 |  |  |
 |
