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:
Navigate<Processor>, Processor, Traceable, Service
Direct Known Subclasses:
Pipeline, Splitter

public class MulticastProcessor
extends ServiceSupport
implements Processor, 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: 17823 $
See Also:
Pipeline

Constructor Summary
MulticastProcessor(Collection<Processor> processors)
           
MulticastProcessor(Collection<Processor> processors, AggregationStrategy aggregationStrategy)
           
MulticastProcessor(Collection<Processor> processors, AggregationStrategy aggregationStrategy, boolean parallelProcessing, ExecutorService executorService, boolean streaming, boolean stopOnException)
           
 
Method Summary
protected static org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair createProcessorExchangePair(Processor processor, Exchange exchange)
          Creates the ProcessorExchangePair which holds the processor and exchange to be send out.
protected  Iterable<org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair> createProcessorExchangePairs(Exchange exchange)
           
protected  void doAggregate(AtomicExchange result, Exchange exchange)
          Aggregate the Exchange with the current result
protected  void doProcessParallel(AtomicExchange result, Iterable<org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair> pairs, boolean streaming)
           
protected  void doProcessSequential(AtomicExchange result, Iterable<org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair> pairs)
           
protected  void doStart()
           
protected  void doStop()
           
 AggregationStrategy getAggregationStrategy()
           
 ExecutorService getExecutorService()
           
 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
 void setExecutorService(ExecutorService executorService)
           
 String toString()
           
protected  void updateNewExchange(Exchange exchange, int index, Iterable<org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair> allPairs)
           
 
Methods inherited from class org.apache.camel.impl.ServiceSupport
addChildService, getStatus, getVersion, isRunAllowed, isStarted, isStarting, isStopped, isStopping, removeChildService, start, stop
 
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
 

Constructor Detail

MulticastProcessor

public MulticastProcessor(Collection<Processor> processors)

MulticastProcessor

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

MulticastProcessor

public MulticastProcessor(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

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.

doProcessParallel

protected void doProcessParallel(AtomicExchange result,
                                 Iterable<org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair> pairs,
                                 boolean streaming)
                          throws InterruptedException,
                                 ExecutionException
Throws:
InterruptedException
ExecutionException

doProcessSequential

protected void doProcessSequential(AtomicExchange result,
                                   Iterable<org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair> pairs)
                            throws Exception
Throws:
Exception

doAggregate

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

Parameters:
result - the current result
exchange - the exchange to be added to the result

updateNewExchange

protected void updateNewExchange(Exchange exchange,
                                 int index,
                                 Iterable<org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair> allPairs)

createProcessorExchangePairs

protected Iterable<org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair> createProcessorExchangePairs(Exchange exchange)
                                                                                                              throws Exception
Throws:
Exception

createProcessorExchangePair

protected static org.apache.camel.processor.MulticastProcessor.ProcessorExchangePair createProcessorExchangePair(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

doStop

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

doStart

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

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()

isParallelProcessing

public boolean isParallelProcessing()

getExecutorService

public ExecutorService getExecutorService()

setExecutorService

public void setExecutorService(ExecutorService executorService)

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.