org.apache.camel.processor
Class MulticastProcessor

java.lang.Object
  extended by org.apache.camel.impl.ServiceSupport
      extended by org.apache.camel.processor.MulticastProcessor
All Implemented Interfaces:
AsyncProcessor, Navigate<Processor>, Processor, Traceable, Service, ShutdownableService
Direct Known Subclasses:
Pipeline, RecipientListProcessor, Splitter

public class MulticastProcessor
extends ServiceSupport
implements AsyncProcessor, Navigate<Processor>, Traceable

Implements the Multicast pattern to send a message exchange to a number of endpoints, each endpoint receiving a copy of the message exchange.

Version:
$Revision: 20191 $
See Also:
Pipeline

Constructor Summary
MulticastProcessor(CamelContext camelContext, Collection<Processor> processors)
           
MulticastProcessor(CamelContext camelContext, Collection<Processor> processors, AggregationStrategy aggregationStrategy)
           
MulticastProcessor(CamelContext camelContext, Collection<Processor> processors, AggregationStrategy aggregationStrategy, boolean parallelProcessing, ExecutorService executorService, boolean streaming, boolean stopOnException)
           
 
Method Summary
protected  ProcessorExchangePair createProcessorExchangePair(int index, Processor processor, Exchange exchange)
          Creates the ProcessorExchangePair which holds the processor and exchange to be send out.
protected  Iterable<ProcessorExchangePair> createProcessorExchangePairs(Exchange exchange)
           
protected  void doAggregate(AggregationStrategy strategy, AtomicExchange result, Exchange exchange)
          Aggregate the Exchange with the current result
protected  void doDone(Exchange original, Exchange subExchange, AsyncCallback callback, boolean doneSync)
          Common work which must be done when we are done multicasting.
protected  void doProcessParallel(Exchange original, AtomicExchange result, Iterable<ProcessorExchangePair> pairs, boolean streaming, AsyncCallback callback)
           
protected  boolean doProcessSequential(Exchange original, AtomicExchange result, Iterable<ProcessorExchangePair> pairs, AsyncCallback callback)
           
protected  void doStart()
           
protected  void doStop()
           
 AggregationStrategy getAggregationStrategy()
          Use getAggregationStrategy(org.apache.camel.Exchange) instead.
protected  AggregationStrategy getAggregationStrategy(Exchange exhange)
           
 CamelContext getCamelContext()
           
 Collection<Processor> getProcessors()
          Returns the producers to multicast to
 String getTraceLabel()
          Gets the trace label used for logging when tracing is enabled.
 boolean hasNext()
          Are there more outputs?
 boolean isParallelProcessing()
           
 boolean isStopOnException()
          Should the multicast processor stop processing further exchanges in case of an exception occurred?
 boolean isStreaming()
          Is the multicast processor working in streaming mode? In streaming mode: we use Iterable to ensure we can send messages as soon as the data becomes available for parallel processing, we start aggregating responses as they get send back to the processor; this means the AggregationStrategy has to take care of handling out-of-order arrival of exchanges
 List<Processor> next()
          Next group of outputs
 void process(Exchange exchange)
          Processes the message exchange
 boolean process(Exchange exchange, AsyncCallback callback)
          Processes the message exchange.
protected static void setToEndpoint(Exchange exchange, Processor processor)
           
 String toString()
           
protected  void updateNewExchange(Exchange exchange, int index, Iterable<ProcessorExchangePair> allPairs, Iterator<ProcessorExchangePair> it)
           
 
Methods inherited from class org.apache.camel.impl.ServiceSupport
addChildService, doShutdown, getStatus, getVersion, isRunAllowed, isStarted, isStarting, isStopped, isStopping, removeChildService, shutdown, start, stop
 
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
 

Constructor Detail

MulticastProcessor

public MulticastProcessor(CamelContext camelContext,
                          Collection<Processor> processors)

MulticastProcessor

public MulticastProcessor(CamelContext camelContext,
                          Collection<Processor> processors,
                          AggregationStrategy aggregationStrategy)

MulticastProcessor

public MulticastProcessor(CamelContext camelContext,
                          Collection<Processor> processors,
                          AggregationStrategy aggregationStrategy,
                          boolean parallelProcessing,
                          ExecutorService executorService,
                          boolean streaming,
                          boolean stopOnException)
Method Detail

toString

public String toString()
Overrides:
toString in class Object

getTraceLabel

public String getTraceLabel()
Description copied from interface: Traceable
Gets the trace label used for logging when tracing is enabled.

The label should be short and precise.

Specified by:
getTraceLabel in interface Traceable
Returns:
the label

getCamelContext

public CamelContext getCamelContext()

process

public void process(Exchange exchange)
             throws Exception
Description copied from interface: Processor
Processes the message exchange

Specified by:
process in interface Processor
Parameters:
exchange - the message exchange
Throws:
Exception - if an internal processing error has occurred.

process

public boolean process(Exchange exchange,
                       AsyncCallback callback)
Description copied from interface: AsyncProcessor
Processes the message exchange. Similar to Processor.process(org.apache.camel.Exchange), but the caller supports having the exchange asynchronously processed.

If there was a failure processing then the caused Exception would be set on the Exchange.

Specified by:
process in interface AsyncProcessor
Parameters:
exchange - the message exchange
callback - the AsyncCallback will be invoked when the processing of the exchange is completed. If the exchange is completed synchronously, then the callback is also invoked synchronously. The callback should therefore be careful of starting recursive loop.
Returns:
(doneSync) true to continue execute synchronously, false to continue being executed asynchronously
See Also:
org.apache.camel.util.AsyncProcessorHelper#process(AsyncProcessor, Exchange, AsyncCallback)}

doProcessParallel

protected void doProcessParallel(Exchange original,
                                 AtomicExchange result,
                                 Iterable<ProcessorExchangePair> pairs,
                                 boolean streaming,
                                 AsyncCallback callback)
                          throws InterruptedException,
                                 ExecutionException
Throws:
InterruptedException
ExecutionException

doProcessSequential

protected boolean doProcessSequential(Exchange original,
                                      AtomicExchange result,
                                      Iterable<ProcessorExchangePair> pairs,
                                      AsyncCallback callback)
                               throws Exception
Throws:
Exception

doDone

protected void doDone(Exchange original,
                      Exchange subExchange,
                      AsyncCallback callback,
                      boolean doneSync)
Common work which must be done when we are done multicasting.

This logic applies for both running synchronous and asynchronous as there are multiple exist points when using the asynchronous routing engine. And therefore we want the logic in one method instead of being scattered.

Parameters:
original - the original exchange
subExchange - the current sub exchange, can be null for the synchronous part
callback - the callback
doneSync - the doneSync parameter to call on callback

doAggregate

protected void doAggregate(AggregationStrategy strategy,
                           AtomicExchange result,
                           Exchange exchange)
Aggregate the Exchange with the current result

Parameters:
strategy - the aggregation strategy to use
result - the current result
exchange - the exchange to be added to the result

updateNewExchange

protected void updateNewExchange(Exchange exchange,
                                 int index,
                                 Iterable<ProcessorExchangePair> allPairs,
                                 Iterator<ProcessorExchangePair> it)

createProcessorExchangePairs

protected Iterable<ProcessorExchangePair> createProcessorExchangePairs(Exchange exchange)
                                                                throws Exception
Throws:
Exception

createProcessorExchangePair

protected ProcessorExchangePair createProcessorExchangePair(int index,
                                                            Processor processor,
                                                            Exchange exchange)
Creates the ProcessorExchangePair which holds the processor and exchange to be send out.

You must use this method to create the instances of ProcessorExchangePair as they need to be specially prepared before use.

Parameters:
processor - the processor
exchange - the exchange
Returns:
prepared for use

doStart

protected void doStart()
                throws Exception
Specified by:
doStart in class ServiceSupport
Throws:
Exception

doStop

protected void doStop()
               throws Exception
Specified by:
doStop in class ServiceSupport
Throws:
Exception

setToEndpoint

protected static void setToEndpoint(Exchange exchange,
                                    Processor processor)

getAggregationStrategy

protected AggregationStrategy getAggregationStrategy(Exchange exhange)

isStreaming

public boolean isStreaming()
Is the multicast processor working in streaming mode? In streaming mode:


isStopOnException

public boolean isStopOnException()
Should the multicast processor stop processing further exchanges in case of an exception occurred?


getProcessors

public Collection<Processor> getProcessors()
Returns the producers to multicast to


getAggregationStrategy

public AggregationStrategy getAggregationStrategy()
Use getAggregationStrategy(org.apache.camel.Exchange) instead.


isParallelProcessing

public boolean isParallelProcessing()

next

public List<Processor> next()
Description copied from interface: Navigate
Next group of outputs

Specified by:
next in interface Navigate<Processor>
Returns:
next group or null if no more outputs

hasNext

public boolean hasNext()
Description copied from interface: Navigate
Are there more outputs?

Specified by:
hasNext in interface Navigate<Processor>
Returns:
true if more outputs


Copyright © 2007-2010 IONA Open Source Community. All Rights Reserved.