1

I'm looking for a way to conditionally handle messages based on the aggregation of messages. I've looked into a lot of ways to do this, but it seems that Apache Camel doesn't support it. I'll explain the scenario and then the solutions I tried.

Scenario: I'm trying to conditionally clean a directory. I poll from the directory every x days and fetch all the files (file://...). I route this into an aggregation, that aggregates the files into a single size (directorySize). I then check if this size passes a certain threshold.

Here is where the problem lies. I now want to remove certain files if this condition passes, but I don't have access to the original messages anymore because they were aggregated in a new exchange.

Solutions:

  • I tried to fetch the files again to process them. Problem is that you can't make a consumer fetch on demand as far as I know. I tried using pollEnrich, but that will only fetch a single file and not all files in the directory.
  • I tried to filter/stop the parent route. The problem here is that filter()/choice...stop()/end() will only stop the aggregated route with the directory size and not the parent route with the file messages. I can't conditionally process these.
  • I tried to move the aggregated condition to another route that I would call, but this causes the same problem as the first solution.

Things I consider doing:

  • Rewrite the aggregation strategy to not only aggregate the size, but also the files itself into a groupedExchange. This way I can split the aggregation again after the check. I don't really like this solution because it causes a lot boilerplate, both in code as during runtime.
  • Move the file size calculator to a processor instead of the aggregator. This would defeat the purpose of using camel in the first place.. I would manually be fetching the files and adding the sizes.. And that for every single file..
  • Use a ControlBus to dynamically start the delete route on that directory. Once again a lot of workaround to achieve something that I feel should be able to be done in a simple route.
  • I would like to set the calculated size on every parent message, but I have no clue how this could be achieved?
  • Another way to stop the parent route that I haven't thought of?

I'm a bit stunned that you can't elegantly filter messages based on the aggregation of these messages. Is there something that I missed in Camel that would provide an elegant solution? Or is this a case of the least bad solution?

Simple Schema

Message(File)

Message(File) --> AggregatedMessage(directorySize) --> delete certain Files?

Message(File)

user2890248
  • 127
  • 2
  • 8

1 Answers1

1

Camel is really awesome, but sometimes it's sure difficult to see exactly which design pattern to use ;)

Firstly, you need to keep a copy of the file objects, because you don't know whether to delete them or not until you reach your threshold - there are basically (at least) two ways to do this.

Alternative 1

The first way is to use a List in an exchange property. This property will hang around no matter what you do with the exchange body. If you have a look at the source code for GroupedExchangeAggregationStrategy, it does precisely this:

        list = new ArrayList<Exchange>();
        answer.setProperty(Exchange.GROUPED_EXCHANGE, list);
        // ...
        list.add(newExchange);

Or you could do the same thing manually on your own exchange property. In any case, it's completely fine to use the Grouped aggregation strategy as you have done.

Alternative 2

The second way to "keep" old messages is to send a copy to a stopped SEDA queue. So you would do to("seda:xyz"). You define this queue as .noAutoStartup(). Then you can send messages to it and they will queue up on an internal queue, managed by camel. When you want to process the messages, you simply start it up via controlbus and stop it again afterwards.

Generally, messing around with starting and stopping queues should be avoided unless absolutely necessary, but that's certainly another way to do it

Suggested solution

I suggest you do as you have done (i.e. alternative 1):

  • aggregate via GroupedExchangeAggregationStrategy to keep the individual files in a list
  • Compute the total file size (use a processor, or do it along the way with a custom aggregation strategy)
  • Use a filter(simple("${body} < 123"))
  • "Unwind" your aggregation via a splitter(simple("${property.CamelGroupedExchange}"))
  • Delete your files one by one

Please let me know if this doesn'y makes sense, or if I have misunderstood your problem in any way.

vikingsteve
  • 38,481
  • 23
  • 112
  • 156
  • How do you feel about this: from("file://location").aggregate(...).to("direct:test"); from("file//location?cron=...").pollEnrich("direct:test").filter(..); It was suggested to me by someone else. I feel like this would cause even more overhead, but gives more flexibility to remove eg only files as long as the size is over the limit because you do the calculation every time again. – user2890248 Dec 16 '13 at 12:58
  • It also requires some tweaking such as removing the read lock because I'm reading the directory from two consumer points at the same time. I'm not that familiar with the impact of removing files without the camel read lock. – user2890248 Dec 16 '13 at 13:06
  • I haven't used poll enricher - it looks very similar to polling consumer, does it work for you? File locks - you'll need to find a way to release any read locks before you delete files, I have unfortunately no experience doing this in camel. – vikingsteve Dec 16 '13 at 13:54