camel-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Martin Krasser (JIRA)" <j...@apache.org>
Subject [jira] Created: (CAMEL-125) Support for a stream-processing resequencer.
Date Sun, 26 Aug 2007 17:01:23 GMT
Support for a stream-processing resequencer.
--------------------------------------------

                 Key: CAMEL-125
                 URL: https://issues.apache.org/activemq/browse/CAMEL-125
             Project: Apache Camel
          Issue Type: Improvement
          Components: camel-core, camel-spring
    Affects Versions: 1.1.0
         Environment: Java 6, Windows XP
            Reporter: Martin Krasser
         Attachments: camel-core-patch.txt, camel-spring-patch.txt

Attached is a patch that adds a stream-processing resequencer to Camel. The resequencing 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 in advance (although a capacity parameter prevents the resequencer
from running out of memory) 

Route builder examples for the stream-processing resequencer:

{{from("direct:start").resequencer(header("seqnum")).stream().to("mock:result")}}

is equivalent to:

{{from("direct:start").resequencer(header("seqnum")).stream(StreamResequencerConfig.getDefault()).to("mock:result")}}

Custom values for the resequencer's capacity and timeout can be set like in this example:

{{from("direct:start").resequencer(header("seqnum")).stream(new StreamResequencerConfig(300,
4000L)).to("mock:result")}}

The XML configuration looks like:

{code:xml}
<camelContext id="camel" xmlns="http://activemq.apache.org/camel/schema/spring">
  <route>
    <from uri="direct:start"/>
    <resequencer>
      <simple>in.header.seqnum</simple>
      <to uri="mock:result" />
      <stream-config capacity="300", timeout="4000"/>
    </resequencer>
  </route>
</camelContext>
{code}

The existing batch-processing resequencer can be defined as usual:

{{from("direct:start").resequencer(header("seqnum")).to("mock:result")}}

which is now equivalent to 

{{from("direct:start").resequencer(header("seqnum")).batch().to("mock:result")}}

It is now also possible to define a custom configuration for the existing batch-processing
resequencer:  

{{from("direct:start").resequencer(header("seqnum")).batch(new BatchResequencerConfig(300,
4000L)).to("mock:result")}}

This set the batchSize to 300 and the batchTimeout to 4000 ms. 

For the stream-processing resequencer to work, messages must contain a 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). 

Gap detection is done with strategies that implement the SequenceNumberComparator<E>
interface. In addition to the java.util.Comparator<E>.compare(E, E) operation the SequenceNumberComparator<E>
interface defines the predecessor(E, E) and successor(E, E) operations. The stream resequencer
can be configured with cutstom SequenceNumberComparator<E> strategies.

The stream-processing resequencer uses the same algorithm as the one in ServiceMix-3.2-SNAPSHOT
(servicemix-eip). In order to avoid compile-time dependencies to ServiceMix I've copied the
ServiceMix-independent resequencing engine over to Camel. This redundancy should be removed
once Camel and servicemix-eip are going to be combined (are they?). I can contribute to this
task, if needed.


-- 
This message is automatically generated by JIRA.
-
You can reply to this email to add a comment to the issue online.


Mime
View raw message