Hi I have to implement the following usecase: - messages are processed and finally delivered to a queue - from this messages I need to generate multiple files (which message-content goes to which file is determined by header value) - I need to create a file based on multiple triggers: 1. when 1000 messages are available (for one file) 2. after waiting for an hour and there are less than 1000 messages available 3. if the context is shutted down - after a file is created, the collection of messages begins again until the next file is ready to create - the consumption of messages for the file should be transacted - this means that I consume 1000 messages and create the file. if the file is written, the 1000 messages are commited, otherwise the all remain in the queue.
my first thought was to use aggregator who can easily split the messages by header values. I have written a quick testcase that looks like this (numbers are reduced for testability): AggregationCollection aggregatePredicate = new PredicateAggregationCollection( header("brand"), new GroupedExchangeAggregationStrategy(), property(Exchange.AGGREGATED_SIZE).isEqualTo(3)); from("activemq:queue:aggregator.queue") .transacted().policy(txPolicy) .routeId("Aggregator-Route") .aggregate(aggregatePredicate) .batchTimeout(2000) .to("mock:result"); The following questions arised: - how can I trigger the time based consumption? That is, if the 1000 messages are not reached within an hour, it should nevertheless write a file with all available messages. I learned from other posts that this cannot be done with "batchTimeout" option. - how can I trigger the file creation when the context is shutted down? - the camel jms-consumer seems to consume message by message. how can I consume and commit/rollback batches of n messages? Because of these questions I found batchConsumer, but there I still have the problem that the camel jms-consumer does not support batchConsumer. How would a camel expert solve this usecase? Thanks Stefan