Cameleer

Splitting and aggregating messages

The splitter breaks one message into many, the aggregator combines many messages into one, and the resequencer and sort EIPs put messages and lists back in order.

These are the patterns for batch files, JSON arrays, order lines, sensor readings and document chunks. Every example runs in the test suite as YAML and as Rust DSL. See the EIP guide for the other patterns.

Splitter

split evaluates an expression that returns a list (a JSON array, lines of text, XML nodes) and runs its steps once per item. Each part has the exchange properties CamelSplitIndex, CamelSplitSize and CamelSplitComplete.

yaml example: eip-split-json
- route:
    from:
      uri: direct:order
      steps:
        - split:
            jsonpath: "$.items"
            steps:
              - setHeader:
                  name: sku
                  simple: "${body[sku]}"
              - to: mock:order-line
rust
b.from("direct:order")
    .split(jsonpath("$.items"))
        .set_header("sku", simple("${body[sku]}"))
        .to("mock:order-line")
    .end();

The order {"items":[{"sku":"A1"},{"sku":"B2"}]} gives two messages on mock:order-line, with sku set to A1 and B2.

Options: parallelProcessing, streaming (split a large text body lazily, one part at a time), stopOnException, timeout, shareUnitOfWork, delimiter and aggregationStrategy. Without an aggregation strategy the original message continues after the split, as in Camel.

Split and aggregate the results

With an aggregation strategy, the split combines the outputs of its parts, and the route continues with the combined message. This is the "composed message processor": split a request, handle each part, and put the answers back together.

yaml example: eip-split-aggregate
- route:
    from:
      uri: direct:lines
      steps:
        - split:
            tokenize: "\n"
            aggregationStrategy: "#class:org.apache.camel.processor.aggregate.GroupedBodyAggregationStrategy"
            steps:
              - setBody:
                  simple: "${body.toUpperCase()}"
        - to: mock:upper-lines
rust
b.from("direct:lines")
    .split(tokenize("\n"))
        .aggregation_strategy(strategies::grouped_body())
        .set_body(simple("${body.toUpperCase()}"))
    .end()
    .to("mock:upper-lines");

The body a\nb\nc reaches mock:upper-lines as the list ["A", "B", "C"]. Built-in strategies by name: GroupedBodyAggregationStrategy, GroupedExchangeAggregationStrategy, UseLatestAggregationStrategy, UseOriginalAggregationStrategy and StringAggregationStrategy, written as #class: references like in Camel or bound in the registry under a name.

Aggregator

aggregate collects messages with the same correlation key and releases them as one message when a completion condition is met: completionSize, completionTimeout (an inactivity timeout), completionInterval, completionPredicate, or completionFromBatchConsumer. The exchange property CamelAggregatedCompletedBy tells which condition fired (size, timeout, predicate, interval, force...).

yaml example: eip-aggregate
- route:
    from:
      uri: direct:readings
      steps:
        - aggregate:
            correlationExpression:
              simple: "${header.device}"
            aggregationStrategy: "#class:org.apache.camel.processor.aggregate.GroupedBodyAggregationStrategy"
            completionSize: 3
            completionTimeout: 5000
            steps:
              - to: mock:device-batch
rust
b.from("direct:readings")
    .aggregate(header("device"), strategies::grouped_body())
        .completion_size(3)
        .completion_timeout(Duration::from_millis(5000))
        .to("mock:device-batch")
    .end();

Three readings from device t1 are released together as ["20", "21", "22"]; readings from other devices are kept in their own groups. A group that stops receiving messages is released after five seconds, so a quiet device still sends what it has.

Notes for Camel users: aggregation repositories are in memory by default; JDBC and Redis repositories are available in cameleer-sql and cameleer-datastores, there is no LevelDB repository. forceCompletionOnStop, discardOnCompletionTimeout, closeCorrelationKeyOnCompletion, eagerCheckCompletion and completionOnNewCorrelationGroup work as in Camel.

Resequencer

The resequencer puts messages back in order by an expression, usually a sequence number. The batch mode collects messages for batchTimeout milliseconds or batchSize messages, sorts them and sends them on; the stream mode releases each message as soon as the gap before it is filled.

yaml example: eip-resequence
- route:
    from:
      uri: direct:out-of-order
      steps:
        - resequence:
            simple: "${header.seq}"
            batchConfig:
              batchSize: 3
              batchTimeout: 2000
            steps:
              - to: mock:in-order
rust
b.from("direct:out-of-order")
    .resequence(header("seq"))
        .batch()
        .batch_size(3)
        .batch_timeout(Duration::from_millis(2000))
        .to("mock:in-order")
    .end();

Messages with seq 3, 1 and 2 leave in the order 1, 2, 3. The stream mode is set with streamConfig (capacity, timeout, rejectOld).

Sort

sort sorts the list an expression returns and replaces the body with the sorted list. It works on the body of one message, unlike the resequencer, which orders several messages.

yaml example: eip-sort
- route:
    from:
      uri: direct:names
      steps:
        - sort:
            tokenize: ","
        - to: mock:sorted
rust
b.from("direct:names")
    .sort(tokenize(","))
    .to("mock:sorted");

The body pear,apple,fig becomes ["apple", "fig", "pear"]. A custom comparator can be bound in the registry and named with comparator.

Common questions

How do I split a JSON array into separate messages?

Use split with jsonpath: "$.items" (or simple: "${body}" when the body already is a list). Each item becomes the body of its own message.

How do I split a large file without loading it all?

Set streaming: true on the split with a tokenize expression. The text is split one part at a time and each part's exchange is created when it is needed.

How do I batch messages, for example 100 at a time or every 10 seconds?

Use aggregate with a constant correlation expression (constant: "all"), the GroupedBodyAggregationStrategy, completionSize: 100 and completionInterval: 10000.

Why does my split not return the processed parts?

Without an aggregationStrategy the original message continues after the split, as in Apache Camel. Add a strategy such as GroupedBodyAggregationStrategy to get the parts back.

Found a problem?

If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: open an issue that describes what you saw, then fork the repository, fix it, add or update a test, run cargo test --workspace, and open a pull request. CONTRIBUTING.md explains the ground rules.

Edit this page on GitHub