Resequencer

- -

The Resequencer from the EIP patterns allows you to reorganise messages based on some comparator. By default in Camel we use an Expression to create the comparator; so that you can compare by a message header or the body or a piece of a message etc.

- -

Change in Camel 2.7

-

The <batch-config> and <stream-config> tags in XML DSL in the Resequencer EIP must now be configured in the top, and not in the bottom. So if you use those, then move them up just below the <resequence> EIP starts in the XML. If you are using Camel older than 2.7, then those configs should be at the bottom.

- -

Camel supports two resequencing algorithms:

- -

Batch resequencing collects messages into a batch, sorts the messages and sends them to their output.
Stream resequencing re-orders (continuous) message streams based on the detection of gaps between messages.

- - -

By default the Resequencer does not support duplicate messages and will only keep the last message, in case a message arrives with the same message expression. However in the batch mode you can enable it to allow duplicates.

- -

Batch Resequencing

- -

The following example shows how to use the batch-processing resequencer so that messages are sorted in order of the body() expression. That is messages are collected into a batch (either by a maximum number of messages per batch or using a timeout) then they are sorted in order and then sent out to their output.

- -

Using the Fluent Builders

- -

+

Resequencer

The Resequencer from the EIP patterns allows you to reorganise messages based on some comparator. By default in Camel we use an Expression to create the comparator; so that you can compare by a message header or the body or a piece of a message etc.

Change in Camel 2.7

The <batch-config> and <stream-config> tags in XML DSL in the Resequencer EIP must now be configured in the top, and not in the bottom. So if you use those, then move them up just below the <resequence> EIP starts in the XML. If you are using Camel older than 2.7, then those configs should be at the bottom.

Camel supports two resequencing algorithms:

Batch resequencing collects messages into a batch, sorts the messages and sends them to their output.
Stream resequencing re-orders (continuous) message streams based on the detection of gaps between messages.

By default the Resequencer does not support duplicate messages and will only keep the last message, in case a message arrive s with the same message expression. However in the batch mode you can enable it to allow duplicates.

Batch Resequencing

The following example shows how to use the batch-processing resequencer so that messages are sorted in order of the body() expression. That is messages are collected into a batch (either by a maximum number of messages per batch or using a timeout) then they are sorted in order and then sent out to their output.

Using the Fluent Builders

-

- -

This is equvalent to

- -

-
-

- -

The batch-processing resequencer can be further configured via the size() and timeout() methods.

- -

-
-

- -

This sets the batch size to 300 and the batch timeout to 4000 ms (by default, the batch size is 100 and the timeout is 1000 ms). Alternatively, you can provide a configuration object.

- -

-
-

- - -

So the above example will reorder messages from endpoint direct:a in order of their bodies, to the endpoint mock:result.
-Typically you'd use a header rather than the body to order things; or maybe a part of the body. So you could replace this expression with

- -

-
-

- -

for example to reorder messages using a custom sequence number in the header mySeqNo.

- -

You can of course use many different Expression languages such as XPath, XQuery, SQL or various Scripting Languages.

- -

Using the Spring XML Extensions

- -

-
-

- -

Allow Duplicates

-

Available as of Camel 2.4

- -

In the batch mode, you can now allow duplicates. In Java DSL there is a allowDuplicates() method and in Spring XML there is an allowDuplicates=true attribute on the <batch-config/> you can use to enable it.

- -

Reverse

-

Available as of Camel 2.4

- -

In the batch mode, you can now reverse the expression ordering. By default the order is based on 0..9,A..Z, which would let messages with low numbers be ordered first, and thus also also outgoing first. In some cases you want to reverse order, which is now possible.

- -

In Java DSL there is a reverse() method and in Spring XML there is an reverse=true attribute on the <batch-config/> you can use to enable it.

- -

Resequence JMS messages based on JMSPriority

-

Available as of Camel 2.4

- -

It's now much easier to use the Resequencer to resequence messages from JMS queues based on JMSPriority. For that to work you need to use the two new options allowDuplicates and reverse.

- -

+

Allow Duplicates

Available as of Camel 2.4

In the batch mode, you can now allow duplicates. In Java DSL there is a allowDuplicates() method and in Spring XML there is an allowDuplicates=true attribute on the <batch-config/> you can use to enable it.

Reverse

Available as of Camel 2.4

In the batch mode, you can now reverse the expression ordering. By default the order is based on 0..9,A..Z, which would let messages with low numbers be ordered first, and thus also also outgoing first. In some cases you want to reverse order, which is now possible.

In Java DSL there is a reverse() method and in Spring XML there is an reverse=true attribute on the <batch-config/> you can use to enable it.

Resequence JMS messages based on JMSPriority

Available as of Camel 2.4

It's now much easier to use the Resequencer to resequence messages from JMS queues based on JMSPriority. For that to work you need to use the two new options allowDuplicates and reverse.

-

- -

Notice this is only possible in the batch mode of the Resequencer.

- -

Ignore invalid exchanges

-

Available as of Camel 2.9

- -

The resequencer EIP will from Camel 2.9 onwards throw a CamelExchangeException if the incoming Exchange is not valid for the resequencer - ie. the expression cannot be evaluated, such as a missing header. You can use the option ignoreInvalidExchanges to ignore these exceptions which means the Resequencer will then skip the invalid Exchange.

- -

+

Notice this is only possible in the batch mode of the Resequencer.

Ignore invalid exchanges

Available as of Camel 2.9

The Resequencer EIP will from Camel 2.9 onwards throw a CamelExchangeException if the incoming Exchange is not valid for the resequencer - ie. the expression cannot be evaluated, such as a missing header. You can use the option ignoreInvalidExchanges to ignore these exceptions which means the Resequencer will then skip the invalid Exchange.

-

- -

This option is available for both batch and stream resequencer.

- -

Reject Old Exchanges

-

Available as of Camel 2.11

- -

This option can be used to prevent out of order messages from being sent regardless of the event that delivered messages downstream (capacity, timeout, etc). If enabled using rejectOld(), the Resequencer will throw a MessageRejectedException when an incoming Exchange is "older" (based on the Comparator) than the last delivered message. This provides an extra level of control with regards to delayed message ordering.

- -

-
-

- -

This option is available for the stream resequencer only.

- -

Stream Resequencing

- -

The next example shows how to use the stream-processing resequencer. Messages are re-ordered based on their sequence numbers given by a seqnum header using gap detection and timeouts on the level of individual messages.

- -

Using the Fluent Builders

- -

+

This option is available for the stream resequencer only.

Stream Resequencing

The next example shows how to use the stream-processing resequencer. Messages are re-ordered based on their sequence numbers given by a seqnum header using gap detection and timeouts on the level of individual messages.

Using the Fluent Builders

-

- -

The stream-processing resequencer can be further configured via the capacity() and timeout() methods.

- -

-
-

- -

This sets the resequencer's capacity to 5000 and the timeout to 4000 ms (by default, the capacity is 1000 and the timeout is 1000 ms). Alternatively, you can provide a configuration object.

- -

-
-

- -

The stream-processing resequencer algorithm 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 sucessor of 3 is missing. The resequencer therefore has to retain message 5 until message 4 arrives (or a timeout occurs).

- -

If the maximum time difference between messages (with successor/predecessor relationship with respect to the sequence number) in a message stream is known, then the resequencer's timeout parameter should be set to this value. In this case it is guaranteed that all messages of a stream are delivered in correct order to the next processor. The lower the timeout value is compared to the out-of-sequence time difference the higher is the probability for out-of-sequence messages delivered by this resequencer. Large timeout values should be supported by sufficiently high capacity values. The capacity parameter is used to prevent the resequencer from running out of memory.

- -

By default, the stream resequencer expects long sequence numbers but other sequence numbers types can be supported as well by providing a custom expression.

- -

+

The stream-processing resequencer algorithm 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).

If the maximum time difference between messages (with successor/predecessor relationship with respect to the sequence number) in a message stream is known, then the resequencer's timeout parameter should be set to this value. In this case it is guaranteed that all messages of a stream are delivered in correct order to the next processor. The lower the timeout value is compared to the out-of-sequence time difference the higher is the probability for out-of-sequence messages delivered by this resequencer. Large timeout values should be supported by sufficiently high capacity values. The capacity parameter is used to prevent the resequencer from running out of memory.

By default, the stream resequencer expects long sequence numbers but other sequence numbers types can be supported as well by providing a custom expression.

-

- -

+

-

- -

or custom comparator via the comparator() method

- -

-
-

- -

or via a StreamResequencerConfig object.

- -

-
-

- - -

Using the Spring XML Extensions

- -

-
-

- -

Further Examples

- -

For further examples of this pattern in use you could look at the batch-processing resequencer junit test case and the stream-processing resequencer junit test case

- -

Using This Pattern

+

Further Examples

For further examples of this pattern in use you could look at the batch-processing resequencer junit test case and the stream-processing resequencer junit test case

Using This Pattern

If you would like to use this EIP Pattern then please read the Getting Started, you may also find the Architecture useful particularly the description of Endpoint and URIs. Then you could try out some of the Examples first before trying this pattern out.