Improving the Aggregator

[Christian Müller <ch...@gmail.com>]

It would be great to improve the Aggregator to support a completion size
which evaluates ALL exchanges (from all different aggregates).
We ofter have the requirement to split a file which contains items which we
process independently. At the end, we have to aggregate the items based on
the customer.

As an example, we have one input file with 1000 items which will result in
2 output files with 300 and 700 items (or 3 output files with 273, 493, 234
items - or 4 output files with ... I think you got it). At the end, the
aggregator will receive in total 1000 files which should be trigger a flush
on ALL existing aggregates.

What do you think?
Is there already a simple solution for this which I miss?

At present we use a combination of two aggregators to do this. Each will
receive a copy of an item. The second aggregator only counts the total
number of all received exchanges (we are using the same aggragation key
value for this aggregator). If this aggragor flushes the exchange, we send
a "command message" to the first aggregator to flush all the aggregates.

I think there should be a simpler solution out of the box which Camel
should offer. E.g. completionRepositorySize or repositoryCompletionSize

Best,
Christian

Re: Improving the Aggregator

[Claus Ibsen <cl...@gmail.com>]

On Fri, Dec 7, 2012 at 9:25 AM, Claus Ibsen <cl...@gmail.com> wrote:
> On Thu, Dec 6, 2012 at 11:56 PM, Christian Müller
> <ch...@gmail.com> wrote:
>> It would be great to improve the Aggregator to support a completion size
>> which evaluates ALL exchanges (from all different aggregates).
>> We ofter have the requirement to split a file which contains items which we
>> process independently. At the end, we have to aggregate the items based on
>> the customer.
>>
>> As an example, we have one input file with 1000 items which will result in
>> 2 output files with 300 and 700 items (or 3 output files with 273, 493, 234
>> items - or 4 output files with ... I think you got it). At the end, the
>> aggregator will receive in total 1000 files which should be trigger a flush
>> on ALL existing aggregates.
>>
>
>> What do you think?
>
>
>> Is there already a simple solution for this which I miss?
>>
>
> Yes.
>
> The EIPs are composable and you can build solutions like lego bricks.
>
> So what you should look at is the composed message processor
> http://camel.apache.org/composed-message-processor.html
> ... to do the rough split + aggregate, into eg 2-4 files.
>
> And then afterwards you can split.
>

Christian did you have a chance to look at this EIP.
I was referring to only using the splitter as it has built-in aggregation.
As you only have 1000 lines in the files, you can keep it in-memory,
and then the 1st splitter
will just do the coarse grained splitting into 2,3 .. or 4 "files".
You can do this in the AggregationStrategy
to store a Map with a key for the filename / correlation id


A pseudo route can be something like:

from file:inbox
  split myCoarseGrainedAggregator
    process myFileNameCorrelation
  end
  split body
     to file:outbox
  end

In the 1st splitter we use an aggregation strategy to split into the
2,3 or 4 files and set the data on the outgoing exchange.
As the splitter expects at least 1 child, then we use a processor to
compute a header with the filename for the given splitted line.

In the 2nd splitter we split those 2,3,4 files and store each file, or
whatever you want to do.





>
>
>> At present we use a combination of two aggregators to do this. Each will
>> receive a copy of an item. The second aggregator only counts the total
>> number of all received exchanges (we are using the same aggragation key
>> value for this aggregator). If this aggragor flushes the exchange, we send
>> a "command message" to the first aggregator to flush all the aggregates.
>>
>> I think there should be a simpler solution out of the box which Camel
>> should offer. E.g. completionRepositorySize or repositoryCompletionSize
>>
>> Best,
>> Christian
>
>
>
> --
> Claus Ibsen
> -----------------
> Red Hat, Inc.
> FuseSource is now part of Red Hat
> Email: cibsen@redhat.com
> Web: http://fusesource.com
> Twitter: davsclaus
> Blog: http://davsclaus.com
> Author of Camel in Action: http://www.manning.com/ibsen



-- 
Claus Ibsen
-----------------
Red Hat, Inc.
FuseSource is now part of Red Hat
Email: cibsen@redhat.com
Web: http://fusesource.com
Twitter: davsclaus
Blog: http://davsclaus.com
Author of Camel in Action: http://www.manning.com/ibsen

Re: Improving the Aggregator

[Claus Ibsen <cl...@gmail.com>]

On Thu, Dec 6, 2012 at 11:56 PM, Christian Müller
<ch...@gmail.com> wrote:
> It would be great to improve the Aggregator to support a completion size
> which evaluates ALL exchanges (from all different aggregates).
> We ofter have the requirement to split a file which contains items which we
> process independently. At the end, we have to aggregate the items based on
> the customer.
>
> As an example, we have one input file with 1000 items which will result in
> 2 output files with 300 and 700 items (or 3 output files with 273, 493, 234
> items - or 4 output files with ... I think you got it). At the end, the
> aggregator will receive in total 1000 files which should be trigger a flush
> on ALL existing aggregates.
>

> What do you think?


> Is there already a simple solution for this which I miss?
>

Yes.

The EIPs are composable and you can build solutions like lego bricks.

So what you should look at is the composed message processor
http://camel.apache.org/composed-message-processor.html
... to do the rough split + aggregate, into eg 2-4 files.

And then afterwards you can split.



> At present we use a combination of two aggregators to do this. Each will
> receive a copy of an item. The second aggregator only counts the total
> number of all received exchanges (we are using the same aggragation key
> value for this aggregator). If this aggragor flushes the exchange, we send
> a "command message" to the first aggregator to flush all the aggregates.
>
> I think there should be a simpler solution out of the box which Camel
> should offer. E.g. completionRepositorySize or repositoryCompletionSize
>
> Best,
> Christian



-- 
Claus Ibsen
-----------------
Red Hat, Inc.
FuseSource is now part of Red Hat
Email: cibsen@redhat.com
Web: http://fusesource.com
Twitter: davsclaus
Blog: http://davsclaus.com
Author of Camel in Action: http://www.manning.com/ibsen

Re: Improving the Aggregator

[Claus Ibsen <cl...@gmail.com>]

On Fri, Dec 7, 2012 at 9:07 AM, Charles Moulliard <ch...@gmail.com> wrote:
> Hi Christian,
>
> What you suggest is really interesting and could be added to the existing
> aggregator if we add some properties and global cache to inform the
> aggregator that it will receive exchanges from different input. But the
> question is "Is it still the role of the aggregator of doing that instead
> of placing the messages aggregated in a (persisted) cache and consume them
> from a separate component to verify if completion has been done globaly ?".
> Which info are we gonna to provide to the camel routes if we have received
> a part of the exchanges and not all ? This is an interesting business use
> case but we should perhaps address it in a different way to avoid too much
> complexity .... in our camel routes.
>

No its way to complicated. The ideas and success of the EIPs is that they are
very compose able, and you can build solutions like "lego bricks".

The idea has never been that there is one giant EIP for every business
problem out there.
Instead you use  the "lego bricks" to build your solution.


> Just 2€ cents for the debate ;-)
>
> Regards,
>
> Charles
>
>
> On Thu, Dec 6, 2012 at 11:56 PM, Christian Müller <
> christian.mueller@gmail.com> wrote:
>
>> It would be great to improve the Aggregator to support a completion size
>> which evaluates ALL exchanges (from all different aggregates).
>> We ofter have the requirement to split a file which contains items which we
>> process independently. At the end, we have to aggregate the items based on
>> the customer.
>>
>> As an example, we have one input file with 1000 items which will result in
>> 2 output files with 300 and 700 items (or 3 output files with 273, 493, 234
>> items - or 4 output files with ... I think you got it). At the end, the
>> aggregator will receive in total 1000 files which should be trigger a flush
>> on ALL existing aggregates.
>>
>> What do you think?
>> Is there already a simple solution for this which I miss?
>>
>> At present we use a combination of two aggregators to do this. Each will
>> receive a copy of an item. The second aggregator only counts the total
>> number of all received exchanges (we are using the same aggragation key
>> value for this aggregator). If this aggragor flushes the exchange, we send
>> a "command message" to the first aggregator to flush all the aggregates.
>>
>> I think there should be a simpler solution out of the box which Camel
>> should offer. E.g. completionRepositorySize or repositoryCompletionSize
>>
>> Best,
>> Christian
>>
>
>
>
> --
> Charles Moulliard
> Apache Committer / Sr. Enterprise Architect (RedHat)
> Twitter : @cmoulliard | Blog : http://cmoulliard.blogspot.com



-- 
Claus Ibsen
-----------------
Red Hat, Inc.
FuseSource is now part of Red Hat
Email: cibsen@redhat.com
Web: http://fusesource.com
Twitter: davsclaus
Blog: http://davsclaus.com
Author of Camel in Action: http://www.manning.com/ibsen

Re: Improving the Aggregator

[Charles Moulliard <ch...@gmail.com>]

Hi Christian,

What you suggest is really interesting and could be added to the existing
aggregator if we add some properties and global cache to inform the
aggregator that it will receive exchanges from different input. But the
question is "Is it still the role of the aggregator of doing that instead
of placing the messages aggregated in a (persisted) cache and consume them
from a separate component to verify if completion has been done globaly ?".
Which info are we gonna to provide to the camel routes if we have received
a part of the exchanges and not all ? This is an interesting business use
case but we should perhaps address it in a different way to avoid too much
complexity .... in our camel routes.

Just 2€ cents for the debate ;-)

Regards,

Charles


On Thu, Dec 6, 2012 at 11:56 PM, Christian Müller <
christian.mueller@gmail.com> wrote:

> It would be great to improve the Aggregator to support a completion size
> which evaluates ALL exchanges (from all different aggregates).
> We ofter have the requirement to split a file which contains items which we
> process independently. At the end, we have to aggregate the items based on
> the customer.
>
> As an example, we have one input file with 1000 items which will result in
> 2 output files with 300 and 700 items (or 3 output files with 273, 493, 234
> items - or 4 output files with ... I think you got it). At the end, the
> aggregator will receive in total 1000 files which should be trigger a flush
> on ALL existing aggregates.
>
> What do you think?
> Is there already a simple solution for this which I miss?
>
> At present we use a combination of two aggregators to do this. Each will
> receive a copy of an item. The second aggregator only counts the total
> number of all received exchanges (we are using the same aggragation key
> value for this aggregator). If this aggragor flushes the exchange, we send
> a "command message" to the first aggregator to flush all the aggregates.
>
> I think there should be a simpler solution out of the box which Camel
> should offer. E.g. completionRepositorySize or repositoryCompletionSize
>
> Best,
> Christian
>



-- 
Charles Moulliard
Apache Committer / Sr. Enterprise Architect (RedHat)
Twitter : @cmoulliard | Blog : http://cmoulliard.blogspot.com