org.apache.camel.processor.aggregate
Class AggregateProcessor

java.lang.Object
  extended by org.apache.camel.impl.ServiceSupport
      extended by org.apache.camel.processor.aggregate.AggregateProcessor
All Implemented Interfaces:
Navigate<Processor>, Processor, Traceable, Service, ShutdownableService

public class AggregateProcessor
extends ServiceSupport
implements Processor, Navigate<Processor>, Traceable

An implementation of the Aggregator pattern where a batch of messages are processed (up to a maximum amount or until some timeout is reached) and messages for the same correlation key are combined together using some kind of AggregationStrategy (by default the latest message is used) to compress many message exchanges into a smaller number of exchanges.

A good example of this is stock market data; you may be receiving 30,000 messages/second and you may want to throttle it right down so that multiple messages for the same stock are combined (or just the latest message is used and older prices are discarded). Another idea is to combine line item messages together into a single invoice message.

Version:
$Revision: 21477 $

Constructor Summary
AggregateProcessor(CamelContext camelContext, Processor processor, Expression correlationExpression, AggregationStrategy aggregationStrategy, ExecutorService executorService)
           
 
Method Summary
protected  void doShutdown()
          Implementations override this method to perform customized shutdown
protected  void doStart()
           
protected  void doStop()
           
 AggregationRepository getAggregationRepository()
           
 Integer getCloseCorrelationKeyOnCompletion()
           
 long getCompletionInterval()
           
 Predicate getCompletionPredicate()
           
 int getCompletionSize()
           
 Expression getCompletionSizeExpression()
           
 long getCompletionTimeout()
           
 Expression getCompletionTimeoutExpression()
           
 ExceptionHandler getExceptionHandler()
           
 String getTraceLabel()
          Gets the trace label used for logging when tracing is enabled.
 boolean hasNext()
          Are there more outputs?
protected  String isCompleted(String key, Exchange exchange)
          Tests whether the given exchange is complete or not
 boolean isCompletionFromBatchConsumer()
           
 boolean isDiscardOnCompletionTimeout()
           
 boolean isEagerCheckCompletion()
           
 boolean isIgnoreInvalidCorrelationKeys()
           
 boolean isParallelProcessing()
           
 List<Processor> next()
          Next group of outputs
protected  Exchange onAggregation(Exchange oldExchange, Exchange newExchange)
           
protected  void onCompletion(String key, Exchange exchange, boolean fromTimeout)
           
 void process(Exchange exchange)
          Processes the message exchange
 void setAggregationRepository(AggregationRepository aggregationRepository)
           
 void setCloseCorrelationKeyOnCompletion(Integer closeCorrelationKeyOnCompletion)
           
 void setCompletionFromBatchConsumer(boolean completionFromBatchConsumer)
           
 void setCompletionInterval(long completionInterval)
           
 void setCompletionPredicate(Predicate completionPredicate)
           
 void setCompletionSize(int completionSize)
           
 void setCompletionSizeExpression(Expression completionSizeExpression)
           
 void setCompletionTimeout(long completionTimeout)
           
 void setCompletionTimeoutExpression(Expression completionTimeoutExpression)
           
 void setDiscardOnCompletionTimeout(boolean discardOnCompletionTimeout)
           
 void setEagerCheckCompletion(boolean eagerCheckCompletion)
           
 void setExceptionHandler(ExceptionHandler exceptionHandler)
           
 void setIgnoreInvalidCorrelationKeys(boolean ignoreInvalidCorrelationKeys)
           
 void setParallelProcessing(boolean parallelProcessing)
           
 String toString()
           
 
Methods inherited from class org.apache.camel.impl.ServiceSupport
addChildService, doResume, doSuspend, getStatus, getVersion, isRunAllowed, isStarted, isStarting, isStopped, isStopping, isSuspended, isSuspending, removeChildService, resume, shutdown, start, start, stop, suspend
 
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
 

Constructor Detail

AggregateProcessor

public AggregateProcessor(CamelContext camelContext,
                          Processor processor,
                          Expression correlationExpression,
                          AggregationStrategy aggregationStrategy,
                          ExecutorService executorService)
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

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

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.

isCompleted

protected String isCompleted(String key,
                             Exchange exchange)
Tests whether the given exchange is complete or not

Parameters:
key - the correlation key
exchange - the incoming exchange
Returns:
null if not completed, otherwise a String with the type that triggered the completion

onAggregation

protected Exchange onAggregation(Exchange oldExchange,
                                 Exchange newExchange)

onCompletion

protected void onCompletion(String key,
                            Exchange exchange,
                            boolean fromTimeout)

getCompletionPredicate

public Predicate getCompletionPredicate()

setCompletionPredicate

public void setCompletionPredicate(Predicate completionPredicate)

isEagerCheckCompletion

public boolean isEagerCheckCompletion()

setEagerCheckCompletion

public void setEagerCheckCompletion(boolean eagerCheckCompletion)

getCompletionTimeout

public long getCompletionTimeout()

setCompletionTimeout

public void setCompletionTimeout(long completionTimeout)

getCompletionTimeoutExpression

public Expression getCompletionTimeoutExpression()

setCompletionTimeoutExpression

public void setCompletionTimeoutExpression(Expression completionTimeoutExpression)

getCompletionInterval

public long getCompletionInterval()

setCompletionInterval

public void setCompletionInterval(long completionInterval)

getCompletionSize

public int getCompletionSize()

setCompletionSize

public void setCompletionSize(int completionSize)

getCompletionSizeExpression

public Expression getCompletionSizeExpression()

setCompletionSizeExpression

public void setCompletionSizeExpression(Expression completionSizeExpression)

isIgnoreInvalidCorrelationKeys

public boolean isIgnoreInvalidCorrelationKeys()

setIgnoreInvalidCorrelationKeys

public void setIgnoreInvalidCorrelationKeys(boolean ignoreInvalidCorrelationKeys)

getCloseCorrelationKeyOnCompletion

public Integer getCloseCorrelationKeyOnCompletion()

setCloseCorrelationKeyOnCompletion

public void setCloseCorrelationKeyOnCompletion(Integer closeCorrelationKeyOnCompletion)

isCompletionFromBatchConsumer

public boolean isCompletionFromBatchConsumer()

setCompletionFromBatchConsumer

public void setCompletionFromBatchConsumer(boolean completionFromBatchConsumer)

getExceptionHandler

public ExceptionHandler getExceptionHandler()

setExceptionHandler

public void setExceptionHandler(ExceptionHandler exceptionHandler)

isParallelProcessing

public boolean isParallelProcessing()

setParallelProcessing

public void setParallelProcessing(boolean parallelProcessing)

getAggregationRepository

public AggregationRepository getAggregationRepository()

setAggregationRepository

public void setAggregationRepository(AggregationRepository aggregationRepository)

isDiscardOnCompletionTimeout

public boolean isDiscardOnCompletionTimeout()

setDiscardOnCompletionTimeout

public void setDiscardOnCompletionTimeout(boolean discardOnCompletionTimeout)

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

doShutdown

protected void doShutdown()
                   throws Exception
Description copied from class: ServiceSupport
Implementations override this method to perform customized shutdown

Overrides:
doShutdown in class ServiceSupport
Throws:
Exception


Apache CAMEL