public class StreamResequencer extends ServiceSupport implements SequenceSender<Exchange>, AsyncProcessor, Navigate<Processor>, Traceable, IdAware
Exchanges. The
algorithm implemented by ResequencerEngine is based on the detection
of gaps in a message stream rather than on a fixed batch size. Gap detection
in combination with timeouts removes the constraint of having to know the
number of messages of a sequence (i.e. the batch size) in advance.
Messages must contain a unique sequence number for which a predecessor and a successor is known. For example a message with the sequence number 3 has a predecessor message with the sequence number 2 and a successor message with the sequence number 4. The message sequence 2,3,5 has a gap because the successor of 3 is missing. The resequencer therefore has to retain message 5 until message 4 arrives (or a timeout occurs).
Instances of this class poll for Exchanges from a given
endpoint. Resequencing work and the delivery of messages to
the next processor is done within the single polling thread.
ResequencerEngineshutdown, shuttingdown, started, starting, stopped, stopping, suspended, suspending| Constructor and Description |
|---|
StreamResequencer(CamelContext camelContext,
Processor processor,
SequenceElementComparator<Exchange> comparator,
Expression expression)
Creates a new
StreamResequencer instance. |
| Modifier and Type | Method and Description |
|---|---|
protected void |
doStart()
Implementations override this method to support customized start/stop.
|
protected void |
doStop()
Implementations override this method to support customized start/stop.
|
int |
getCapacity()
Returns this resequencer's capacity.
|
ExceptionHandler |
getExceptionHandler()
Returns this resequencer's exception handler.
|
Expression |
getExpression() |
String |
getId()
Returns the id
|
Processor |
getProcessor()
Returns the next processor.
|
long |
getTimeout()
Returns this resequencer's timeout.
|
String |
getTraceLabel()
Gets the trace label used for logging when tracing is enabled.
|
boolean |
hasNext()
Are there more outputs?
Important only invoke this once, as this method do not carry state, and is not intended to be used in a while loop,
but used by a if statement instead.
|
boolean |
isIgnoreInvalidExchanges() |
boolean |
isRejectOld() |
List<Processor> |
next()
Next group of outputs
Important only invoke this once, as this method do not carry state, and is not intended to be used in a while loop,
but used by a if statement instead.
|
void |
process(Exchange exchange)
Processes the message exchange
|
boolean |
process(Exchange exchange,
AsyncCallback callback)
Processes the message exchange.
|
void |
sendElement(Exchange exchange)
Sends the
exchange to the next processor. |
void |
setCapacity(int capacity) |
void |
setDeliveryAttemptInterval(long deliveryAttemptInterval) |
void |
setId(String id)
Sets the id
|
void |
setIgnoreInvalidExchanges(boolean ignoreInvalidExchanges)
Sets whether to ignore invalid exchanges which cannot be used by this stream resequencer.
|
void |
setRejectOld(Boolean rejectOld) |
void |
setTimeout(long timeout) |
String |
toString() |
doResume, doShutdown, doSuspend, getStatus, getVersion, isRunAllowed, isStarted, isStarting, isStopped, isStopping, isStoppingOrStopped, isSuspended, isSuspending, isSuspendingOrSuspended, resume, shutdown, start, stop, suspendpublic StreamResequencer(CamelContext camelContext, Processor processor, SequenceElementComparator<Exchange> comparator, Expression expression)
StreamResequencer instance.processor - next processor that processes re-ordered exchanges.comparator - a sequence element comparator for exchanges.public Expression getExpression()
public ExceptionHandler getExceptionHandler()
public Processor getProcessor()
public int getCapacity()
timeout milliseconds giving exchanges the
possibility to time out and to be delivered after the waiting period.public long getTimeout()
ResequencerEngine.setTimeout(long). This value is
also used to define the polling timeout from the endpoint.ResequencerEngine.setTimeout(long)public void setCapacity(int capacity)
public void setTimeout(long timeout)
public void setDeliveryAttemptInterval(long deliveryAttemptInterval)
public boolean isIgnoreInvalidExchanges()
public void setRejectOld(Boolean rejectOld)
public boolean isRejectOld()
public void setIgnoreInvalidExchanges(boolean ignoreInvalidExchanges)
CamelExchangeException is thrown if the Exchange
is invalid.public String getTraceLabel()
TraceablegetTraceLabel in interface Traceableprotected void doStart() throws Exception
ServiceSupportServiceSupport.doStop() for more details.doStart in class ServiceSupportExceptionServiceSupport.doStop()protected void doStop() throws Exception
ServiceSupportServiceSupport.doStop() method when
the service is being stopped. This method will also be invoked
if the service is still in uninitialized state (eg has not
been started). The method is always called to allow the service
to do custom logic when the service is being stopped, such as when
CamelContext is shutting down.doStop in class ServiceSupportExceptionServiceSupport.doStart()public void sendElement(Exchange exchange) throws Exception
exchange to the next processor.sendElement in interface SequenceSender<Exchange>exchange - exchange to send.Exception - if delivery fails.public void process(Exchange exchange) throws Exception
Processorpublic boolean process(Exchange exchange, AsyncCallback callback)
AsyncProcessorProcessor.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.process in interface AsyncProcessorexchange - the message exchangecallback - 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.public boolean hasNext()
NavigateApache Camel