Quantcast

Curious routing case

Previous Topic Next Topic
 
classic Classic list List threaded Threaded
11 messages Options
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Curious routing case

Artur
Hello.

I wonder if someone could push me in the right direction trying to express
quite curious case in Camel route.

Imagine there's a stream of messages some of which can be processed in
parallel and some have to be processed serially. You can group the messages
that require serial processing together by looking at the message body. You
don't know upfront how many groups can occur in the stream.

The way I thought about doing this is having a route for each message
group. Since I don't know upfront how many and what groups there will be
then I would need to create routes dynamically. If a message comes
belonging to a group that doesn't have it's handling route, then i could
create it (is that even possible??) Then if there's no messages coming for
a given group in some time I could remove the route for the group to
cleanup (is that possible?)

New to Camel

Thx!
Artur
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Quinn Stevenson
I’d probably use ActiveMQ Message Groups for this http://activemq.apache.org/message-groups.html <http://activemq.apache.org/message-groups.html>

> On Mar 31, 2017, at 9:08 AM, Artur Jablonski <[hidden email]> wrote:
>
> Hello.
>
> I wonder if someone could push me in the right direction trying to express
> quite curious case in Camel route.
>
> Imagine there's a stream of messages some of which can be processed in
> parallel and some have to be processed serially. You can group the messages
> that require serial processing together by looking at the message body. You
> don't know upfront how many groups can occur in the stream.
>
> The way I thought about doing this is having a route for each message
> group. Since I don't know upfront how many and what groups there will be
> then I would need to create routes dynamically. If a message comes
> belonging to a group that doesn't have it's handling route, then i could
> create it (is that even possible??) Then if there's no messages coming for
> a given group in some time I could remove the route for the group to
> cleanup (is that possible?)
>
> New to Camel
>
> Thx!
> Artur

Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Artur
Hmmm. I am getting messages from Amazon sqs and can't change it. Let's say
I want to see if I can do it in Camel without putting another messaging
system in between.

Interesting feature of amq though.
Thx!

On 31 Mar 2017 5:15 p.m., "Quinn Stevenson" <[hidden email]>
wrote:

I’d probably use ActiveMQ Message Groups for this
http://activemq.apache.org/message-groups.html <http://activemq.apache.org/
message-groups.html>

> On Mar 31, 2017, at 9:08 AM, Artur Jablonski <[hidden email]>
wrote:
>
> Hello.
>
> I wonder if someone could push me in the right direction trying to express
> quite curious case in Camel route.
>
> Imagine there's a stream of messages some of which can be processed in
> parallel and some have to be processed serially. You can group the
messages
> that require serial processing together by looking at the message body.
You

> don't know upfront how many groups can occur in the stream.
>
> The way I thought about doing this is having a route for each message
> group. Since I don't know upfront how many and what groups there will be
> then I would need to create routes dynamically. If a message comes
> belonging to a group that doesn't have it's handling route, then i could
> create it (is that even possible??) Then if there's no messages coming for
> a given group in some time I could remove the route for the group to
> cleanup (is that possible?)
>
> New to Camel
>
> Thx!
> Artur
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Zoran Regvart-2
In reply to this post by Artur
Hi Artur,
have a look at Camel EIP page[1], what you describe sounds to me like
Resequencer and Content based router patterns,

zoran

[1] https://camel.apache.org/eip.html

On Fri, Mar 31, 2017 at 5:08 PM, Artur Jablonski
<[hidden email]> wrote:

> Hello.
>
> I wonder if someone could push me in the right direction trying to express
> quite curious case in Camel route.
>
> Imagine there's a stream of messages some of which can be processed in
> parallel and some have to be processed serially. You can group the messages
> that require serial processing together by looking at the message body. You
> don't know upfront how many groups can occur in the stream.
>
> The way I thought about doing this is having a route for each message
> group. Since I don't know upfront how many and what groups there will be
> then I would need to create routes dynamically. If a message comes
> belonging to a group that doesn't have it's handling route, then i could
> create it (is that even possible??) Then if there's no messages coming for
> a given group in some time I could remove the route for the group to
> cleanup (is that possible?)
>
> New to Camel
>
> Thx!
> Artur



--
Zoran Regvart
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Artur
Hey Zoran.

I read again the patterns you mentioned. In my use case the order of
processing within a group doesn't matter as long as two messages from the
same group are never processed in parallel. So i guess resenquencer is out
of the picture unless I didn't get the intention.

So what we are left with is the content based router. Sure. The message
comes, i can see what group it belongs two... And what next? Perhaps it's
the very first message from that group so I would need to trigger creating
route/processor for that group somehow, perhaps messages from this group
were processed before in which case the processor for the group should
already exist...




On 31 Mar 2017 7:58 p.m., "Zoran Regvart" <[hidden email]> wrote:

> Hi Artur,
> have a look at Camel EIP page[1], what you describe sounds to me like
> Resequencer and Content based router patterns,
>
> zoran
>
> [1] https://camel.apache.org/eip.html
>
> On Fri, Mar 31, 2017 at 5:08 PM, Artur Jablonski
> <[hidden email]> wrote:
> > Hello.
> >
> > I wonder if someone could push me in the right direction trying to
> express
> > quite curious case in Camel route.
> >
> > Imagine there's a stream of messages some of which can be processed in
> > parallel and some have to be processed serially. You can group the
> messages
> > that require serial processing together by looking at the message body.
> You
> > don't know upfront how many groups can occur in the stream.
> >
> > The way I thought about doing this is having a route for each message
> > group. Since I don't know upfront how many and what groups there will be
> > then I would need to create routes dynamically. If a message comes
> > belonging to a group that doesn't have it's handling route, then i could
> > create it (is that even possible??) Then if there's no messages coming
> for
> > a given group in some time I could remove the route for the group to
> > cleanup (is that possible?)
> >
> > New to Camel
> >
> > Thx!
> > Artur
>
>
>
> --
> Zoran Regvart
>
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Zoran Regvart-2
Hi Artur,
I was thinking that the order of the messages would be important as
you need to process them sequentially.

So I think you could use the dynamic message routing[1] with
aggregator[2], something like:

    from("aws-sqs:...")
        .process("#preProcess")
        .toD("direct:${header.nextRoute}");

    from("direct:parallel")...;
    from("direct:sequential").aggregate(simple("${header.group}")).completion..;

So from yout SQS queue you would use a processor to pre-process
message whose responsibility would be to set the (custom) `nextRoute`
and (custom) `group` headers. `nextRoute` would be `parallel` or
`sequential`, and if `sequential` the messages would be aggregated
using the `group` header.

You would want to define your own custom aggregation strategy or use
the completion* options that are available to you. There also might be
need to use seda[3] to fine tune any parallel processing. You might
throw in there a data format unmarshaller[4] instead of the
`preProcess` processor and use something like `${body.xyz} == foo` in
the `toD` expression.

And I would guess that you need to examine transactions or persistence
at some point also in case your aggregation step runs for a long time
or if your use case is sensitive to message loss if interrupted --
which would undoubtedly lead you back to using queues to separate
those two ways of processing,

HTH,

zoran

[1] https://camel.apache.org/message-endpoint.html
[2] https://camel.apache.org/aggregator2.html
[3] https://camel.apache.org/seda.html
[4] https://camel.apache.org/data-format.html

On Sat, Apr 1, 2017 at 1:09 PM, Artur Jablonski
<[hidden email]> wrote:

> Hey Zoran.
>
> I read again the patterns you mentioned. In my use case the order of
> processing within a group doesn't matter as long as two messages from the
> same group are never processed in parallel. So i guess resenquencer is out
> of the picture unless I didn't get the intention.
>
> So what we are left with is the content based router. Sure. The message
> comes, i can see what group it belongs two... And what next? Perhaps it's
> the very first message from that group so I would need to trigger creating
> route/processor for that group somehow, perhaps messages from this group
> were processed before in which case the processor for the group should
> already exist...
>
>
>
>
> On 31 Mar 2017 7:58 p.m., "Zoran Regvart" <[hidden email]> wrote:
>
>> Hi Artur,
>> have a look at Camel EIP page[1], what you describe sounds to me like
>> Resequencer and Content based router patterns,
>>
>> zoran
>>
>> [1] https://camel.apache.org/eip.html
>>
>> On Fri, Mar 31, 2017 at 5:08 PM, Artur Jablonski
>> <[hidden email]> wrote:
>> > Hello.
>> >
>> > I wonder if someone could push me in the right direction trying to
>> express
>> > quite curious case in Camel route.
>> >
>> > Imagine there's a stream of messages some of which can be processed in
>> > parallel and some have to be processed serially. You can group the
>> messages
>> > that require serial processing together by looking at the message body.
>> You
>> > don't know upfront how many groups can occur in the stream.
>> >
>> > The way I thought about doing this is having a route for each message
>> > group. Since I don't know upfront how many and what groups there will be
>> > then I would need to create routes dynamically. If a message comes
>> > belonging to a group that doesn't have it's handling route, then i could
>> > create it (is that even possible??) Then if there's no messages coming
>> for
>> > a given group in some time I could remove the route for the group to
>> > cleanup (is that possible?)
>> >
>> > New to Camel
>> >
>> > Thx!
>> > Artur
>>
>>
>>
>> --
>> Zoran Regvart
>>



--
Zoran Regvart
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Artur
Hi Zoran,

Thank you for such detailed response. This looks very promising. i will
need to get my head around the aggregator pattern.
For this week I will be busy with other tasks, but I will get back to it as
soon as I can to see if I can get Camel work for the use case.

Cheerio
Artur

On Mon, Apr 3, 2017 at 11:09 AM, Zoran Regvart <[hidden email]> wrote:

> Hi Artur,
> I was thinking that the order of the messages would be important as
> you need to process them sequentially.
>
> So I think you could use the dynamic message routing[1] with
> aggregator[2], something like:
>
>     from("aws-sqs:...")
>         .process("#preProcess")
>         .toD("direct:${header.nextRoute}");
>
>     from("direct:parallel")...;
>     from("direct:sequential").aggregate(simple("${header.
> group}")).completion..;
>
> So from yout SQS queue you would use a processor to pre-process
> message whose responsibility would be to set the (custom) `nextRoute`
> and (custom) `group` headers. `nextRoute` would be `parallel` or
> `sequential`, and if `sequential` the messages would be aggregated
> using the `group` header.
>
> You would want to define your own custom aggregation strategy or use
> the completion* options that are available to you. There also might be
> need to use seda[3] to fine tune any parallel processing. You might
> throw in there a data format unmarshaller[4] instead of the
> `preProcess` processor and use something like `${body.xyz} == foo` in
> the `toD` expression.
>
> And I would guess that you need to examine transactions or persistence
> at some point also in case your aggregation step runs for a long time
> or if your use case is sensitive to message loss if interrupted --
> which would undoubtedly lead you back to using queues to separate
> those two ways of processing,
>
> HTH,
>
> zoran
>
> [1] https://camel.apache.org/message-endpoint.html
> [2] https://camel.apache.org/aggregator2.html
> [3] https://camel.apache.org/seda.html
> [4] https://camel.apache.org/data-format.html
>
> On Sat, Apr 1, 2017 at 1:09 PM, Artur Jablonski
> <[hidden email]> wrote:
> > Hey Zoran.
> >
> > I read again the patterns you mentioned. In my use case the order of
> > processing within a group doesn't matter as long as two messages from the
> > same group are never processed in parallel. So i guess resenquencer is
> out
> > of the picture unless I didn't get the intention.
> >
> > So what we are left with is the content based router. Sure. The message
> > comes, i can see what group it belongs two... And what next? Perhaps it's
> > the very first message from that group so I would need to trigger
> creating
> > route/processor for that group somehow, perhaps messages from this group
> > were processed before in which case the processor for the group should
> > already exist...
> >
> >
> >
> >
> > On 31 Mar 2017 7:58 p.m., "Zoran Regvart" <[hidden email]> wrote:
> >
> >> Hi Artur,
> >> have a look at Camel EIP page[1], what you describe sounds to me like
> >> Resequencer and Content based router patterns,
> >>
> >> zoran
> >>
> >> [1] https://camel.apache.org/eip.html
> >>
> >> On Fri, Mar 31, 2017 at 5:08 PM, Artur Jablonski
> >> <[hidden email]> wrote:
> >> > Hello.
> >> >
> >> > I wonder if someone could push me in the right direction trying to
> >> express
> >> > quite curious case in Camel route.
> >> >
> >> > Imagine there's a stream of messages some of which can be processed in
> >> > parallel and some have to be processed serially. You can group the
> >> messages
> >> > that require serial processing together by looking at the message
> body.
> >> You
> >> > don't know upfront how many groups can occur in the stream.
> >> >
> >> > The way I thought about doing this is having a route for each message
> >> > group. Since I don't know upfront how many and what groups there will
> be
> >> > then I would need to create routes dynamically. If a message comes
> >> > belonging to a group that doesn't have it's handling route, then i
> could
> >> > create it (is that even possible??) Then if there's no messages coming
> >> for
> >> > a given group in some time I could remove the route for the group to
> >> > cleanup (is that possible?)
> >> >
> >> > New to Camel
> >> >
> >> > Thx!
> >> > Artur
> >>
> >>
> >>
> >> --
> >> Zoran Regvart
> >>
>
>
>
> --
> Zoran Regvart
>
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Artur
Hello,

I don't think this route definition is fitting my use case, though I learnt
a thing or two about the interesting patterns linked. Thanks!

Ok, so let me try to clarify the use case.


1. The stream is infinite, it's not a batch job. The messages keep on
coming from SQS 'all the time'

2. More important thing is about parallel processing.

Let A1 denote a message 1 from group A, B2 message 2 from group B, etc.

Let's say this is the order in which the messages happen to appear
 in the route from SQS

A1, A2, B1, C1, B2, A3, C2, B3

Now what I am trying to achieve is grouping the messages that have to be
processed sequentially (order doesn't matter as long as no two messages
from the same group are processed at the same time).
So I am trying to somehow get these streams

A1, A2, A3

B1, B2, B3

C1, C2


So, A1 B1 and C1 can be processed in parallel because they are from
different groups, but the messages within groups need to be processed one
by one.

In my example, there are 3 groups, but there can be many and I don't know
what they are in advance. The processing logic between the groups is
similar and is a function of the group so I can get a processor for group A
from a method call getProcessor(A), B getProcessor(B), etc.

I am stuck at how to do that in Camel, because since I don't know the
groups in advance, I would need to create processing routes dynamically.

Say the system starts, and A1 arrives, there can't be any processor for
group A yet, since it's the first message from the group and I need to
somehow dynamically add processing capability of the group A to the route
and then perhaps if the messages from group A stop arriving for some time,
that processor could be removed.

How to add the parallel part between the group messages is also blurry to
me. One way of doing this I was thinking was to do a multicast to all the
dynamically created processing routes for groups and stick a filter before
so that only messages from particular group can go through. From multicast
page:

from("direct:a").multicast().parallelProcessing().to("direct:x",
"direct:y", "direct:z");

But here the x,y,z endpoints are hardcoded. I could write up some custom
multicast I suppose to search the routes in CamelContext...... not sure.

Thanks
Artur





On Mon, Apr 3, 2017 at 1:36 PM, Artur Jablonski <[hidden email]>
wrote:

> Hi Zoran,
>
> Thank you for such detailed response. This looks very promising. i will
> need to get my head around the aggregator pattern.
> For this week I will be busy with other tasks, but I will get back to it
> as soon as I can to see if I can get Camel work for the use case.
>
> Cheerio
> Artur
>
> On Mon, Apr 3, 2017 at 11:09 AM, Zoran Regvart <[hidden email]> wrote:
>
>> Hi Artur,
>> I was thinking that the order of the messages would be important as
>> you need to process them sequentially.
>>
>> So I think you could use the dynamic message routing[1] with
>> aggregator[2], something like:
>>
>>     from("aws-sqs:...")
>>         .process("#preProcess")
>>         .toD("direct:${header.nextRoute}");
>>
>>     from("direct:parallel")...;
>>     from("direct:sequential").aggregate(simple("${header.group}"
>> )).completion..;
>>
>> So from yout SQS queue you would use a processor to pre-process
>> message whose responsibility would be to set the (custom) `nextRoute`
>> and (custom) `group` headers. `nextRoute` would be `parallel` or
>> `sequential`, and if `sequential` the messages would be aggregated
>> using the `group` header.
>>
>> You would want to define your own custom aggregation strategy or use
>> the completion* options that are available to you. There also might be
>> need to use seda[3] to fine tune any parallel processing. You might
>> throw in there a data format unmarshaller[4] instead of the
>> `preProcess` processor and use something like `${body.xyz} == foo` in
>> the `toD` expression.
>>
>> And I would guess that you need to examine transactions or persistence
>> at some point also in case your aggregation step runs for a long time
>> or if your use case is sensitive to message loss if interrupted --
>> which would undoubtedly lead you back to using queues to separate
>> those two ways of processing,
>>
>> HTH,
>>
>> zoran
>>
>> [1] https://camel.apache.org/message-endpoint.html
>> [2] https://camel.apache.org/aggregator2.html
>> [3] https://camel.apache.org/seda.html
>> [4] https://camel.apache.org/data-format.html
>>
>> On Sat, Apr 1, 2017 at 1:09 PM, Artur Jablonski
>> <[hidden email]> wrote:
>> > Hey Zoran.
>> >
>> > I read again the patterns you mentioned. In my use case the order of
>> > processing within a group doesn't matter as long as two messages from
>> the
>> > same group are never processed in parallel. So i guess resenquencer is
>> out
>> > of the picture unless I didn't get the intention.
>> >
>> > So what we are left with is the content based router. Sure. The message
>> > comes, i can see what group it belongs two... And what next? Perhaps
>> it's
>> > the very first message from that group so I would need to trigger
>> creating
>> > route/processor for that group somehow, perhaps messages from this group
>> > were processed before in which case the processor for the group should
>> > already exist...
>> >
>> >
>> >
>> >
>> > On 31 Mar 2017 7:58 p.m., "Zoran Regvart" <[hidden email]> wrote:
>> >
>> >> Hi Artur,
>> >> have a look at Camel EIP page[1], what you describe sounds to me like
>> >> Resequencer and Content based router patterns,
>> >>
>> >> zoran
>> >>
>> >> [1] https://camel.apache.org/eip.html
>> >>
>> >> On Fri, Mar 31, 2017 at 5:08 PM, Artur Jablonski
>> >> <[hidden email]> wrote:
>> >> > Hello.
>> >> >
>> >> > I wonder if someone could push me in the right direction trying to
>> >> express
>> >> > quite curious case in Camel route.
>> >> >
>> >> > Imagine there's a stream of messages some of which can be processed
>> in
>> >> > parallel and some have to be processed serially. You can group the
>> >> messages
>> >> > that require serial processing together by looking at the message
>> body.
>> >> You
>> >> > don't know upfront how many groups can occur in the stream.
>> >> >
>> >> > The way I thought about doing this is having a route for each message
>> >> > group. Since I don't know upfront how many and what groups there
>> will be
>> >> > then I would need to create routes dynamically. If a message comes
>> >> > belonging to a group that doesn't have it's handling route, then i
>> could
>> >> > create it (is that even possible??) Then if there's no messages
>> coming
>> >> for
>> >> > a given group in some time I could remove the route for the group to
>> >> > cleanup (is that possible?)
>> >> >
>> >> > New to Camel
>> >> >
>> >> > Thx!
>> >> > Artur
>> >>
>> >>
>> >>
>> >> --
>> >> Zoran Regvart
>> >>
>>
>>
>>
>> --
>> Zoran Regvart
>>
>
>
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Artur
I guess one thing that come to my mind is to hide all this parallel stuff
inside a processor, that would just spit out on the other end the result of
processing all those messages. It would handle grouping and serializing and
stuff I guess that would reduce the complexity of the route with a cost of
complexity of the processor. I have no better ideas anyway, so I will give
it a go

On Tue, Apr 11, 2017 at 10:45 AM, Artur Jablonski <[hidden email]>
wrote:

> Hello,
>
> I don't think this route definition is fitting my use case, though I
> learnt a thing or two about the interesting patterns linked. Thanks!
>
> Ok, so let me try to clarify the use case.
>
>
> 1. The stream is infinite, it's not a batch job. The messages keep on
> coming from SQS 'all the time'
>
> 2. More important thing is about parallel processing.
>
> Let A1 denote a message 1 from group A, B2 message 2 from group B, etc.
>
> Let's say this is the order in which the messages happen to appear
>  in the route from SQS
>
> A1, A2, B1, C1, B2, A3, C2, B3
>
> Now what I am trying to achieve is grouping the messages that have to be
> processed sequentially (order doesn't matter as long as no two messages
> from the same group are processed at the same time).
> So I am trying to somehow get these streams
>
> A1, A2, A3
>
> B1, B2, B3
>
> C1, C2
>
>
> So, A1 B1 and C1 can be processed in parallel because they are from
> different groups, but the messages within groups need to be processed one
> by one.
>
> In my example, there are 3 groups, but there can be many and I don't know
> what they are in advance. The processing logic between the groups is
> similar and is a function of the group so I can get a processor for group A
> from a method call getProcessor(A), B getProcessor(B), etc.
>
> I am stuck at how to do that in Camel, because since I don't know the
> groups in advance, I would need to create processing routes dynamically.
>
> Say the system starts, and A1 arrives, there can't be any processor for
> group A yet, since it's the first message from the group and I need to
> somehow dynamically add processing capability of the group A to the route
> and then perhaps if the messages from group A stop arriving for some time,
> that processor could be removed.
>
> How to add the parallel part between the group messages is also blurry to
> me. One way of doing this I was thinking was to do a multicast to all the
> dynamically created processing routes for groups and stick a filter before
> so that only messages from particular group can go through. From multicast
> page:
>
> from("direct:a").multicast().parallelProcessing().to("direct:x",
> "direct:y", "direct:z");
>
> But here the x,y,z endpoints are hardcoded. I could write up some custom
> multicast I suppose to search the routes in CamelContext...... not sure.
>
> Thanks
> Artur
>
>
>
>
>
> On Mon, Apr 3, 2017 at 1:36 PM, Artur Jablonski <[hidden email]>
> wrote:
>
>> Hi Zoran,
>>
>> Thank you for such detailed response. This looks very promising. i will
>> need to get my head around the aggregator pattern.
>> For this week I will be busy with other tasks, but I will get back to it
>> as soon as I can to see if I can get Camel work for the use case.
>>
>> Cheerio
>> Artur
>>
>> On Mon, Apr 3, 2017 at 11:09 AM, Zoran Regvart <[hidden email]> wrote:
>>
>>> Hi Artur,
>>> I was thinking that the order of the messages would be important as
>>> you need to process them sequentially.
>>>
>>> So I think you could use the dynamic message routing[1] with
>>> aggregator[2], something like:
>>>
>>>     from("aws-sqs:...")
>>>         .process("#preProcess")
>>>         .toD("direct:${header.nextRoute}");
>>>
>>>     from("direct:parallel")...;
>>>     from("direct:sequential").aggregate(simple("${header.group}"
>>> )).completion..;
>>>
>>> So from yout SQS queue you would use a processor to pre-process
>>> message whose responsibility would be to set the (custom) `nextRoute`
>>> and (custom) `group` headers. `nextRoute` would be `parallel` or
>>> `sequential`, and if `sequential` the messages would be aggregated
>>> using the `group` header.
>>>
>>> You would want to define your own custom aggregation strategy or use
>>> the completion* options that are available to you. There also might be
>>> need to use seda[3] to fine tune any parallel processing. You might
>>> throw in there a data format unmarshaller[4] instead of the
>>> `preProcess` processor and use something like `${body.xyz} == foo` in
>>> the `toD` expression.
>>>
>>> And I would guess that you need to examine transactions or persistence
>>> at some point also in case your aggregation step runs for a long time
>>> or if your use case is sensitive to message loss if interrupted --
>>> which would undoubtedly lead you back to using queues to separate
>>> those two ways of processing,
>>>
>>> HTH,
>>>
>>> zoran
>>>
>>> [1] https://camel.apache.org/message-endpoint.html
>>> [2] https://camel.apache.org/aggregator2.html
>>> [3] https://camel.apache.org/seda.html
>>> [4] https://camel.apache.org/data-format.html
>>>
>>> On Sat, Apr 1, 2017 at 1:09 PM, Artur Jablonski
>>> <[hidden email]> wrote:
>>> > Hey Zoran.
>>> >
>>> > I read again the patterns you mentioned. In my use case the order of
>>> > processing within a group doesn't matter as long as two messages from
>>> the
>>> > same group are never processed in parallel. So i guess resenquencer is
>>> out
>>> > of the picture unless I didn't get the intention.
>>> >
>>> > So what we are left with is the content based router. Sure. The message
>>> > comes, i can see what group it belongs two... And what next? Perhaps
>>> it's
>>> > the very first message from that group so I would need to trigger
>>> creating
>>> > route/processor for that group somehow, perhaps messages from this
>>> group
>>> > were processed before in which case the processor for the group should
>>> > already exist...
>>> >
>>> >
>>> >
>>> >
>>> > On 31 Mar 2017 7:58 p.m., "Zoran Regvart" <[hidden email]> wrote:
>>> >
>>> >> Hi Artur,
>>> >> have a look at Camel EIP page[1], what you describe sounds to me like
>>> >> Resequencer and Content based router patterns,
>>> >>
>>> >> zoran
>>> >>
>>> >> [1] https://camel.apache.org/eip.html
>>> >>
>>> >> On Fri, Mar 31, 2017 at 5:08 PM, Artur Jablonski
>>> >> <[hidden email]> wrote:
>>> >> > Hello.
>>> >> >
>>> >> > I wonder if someone could push me in the right direction trying to
>>> >> express
>>> >> > quite curious case in Camel route.
>>> >> >
>>> >> > Imagine there's a stream of messages some of which can be processed
>>> in
>>> >> > parallel and some have to be processed serially. You can group the
>>> >> messages
>>> >> > that require serial processing together by looking at the message
>>> body.
>>> >> You
>>> >> > don't know upfront how many groups can occur in the stream.
>>> >> >
>>> >> > The way I thought about doing this is having a route for each
>>> message
>>> >> > group. Since I don't know upfront how many and what groups there
>>> will be
>>> >> > then I would need to create routes dynamically. If a message comes
>>> >> > belonging to a group that doesn't have it's handling route, then i
>>> could
>>> >> > create it (is that even possible??) Then if there's no messages
>>> coming
>>> >> for
>>> >> > a given group in some time I could remove the route for the group to
>>> >> > cleanup (is that possible?)
>>> >> >
>>> >> > New to Camel
>>> >> >
>>> >> > Thx!
>>> >> > Artur
>>> >>
>>> >>
>>> >>
>>> >> --
>>> >> Zoran Regvart
>>> >>
>>>
>>>
>>>
>>> --
>>> Zoran Regvart
>>>
>>
>>
>
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Claus Ibsen-2
If you use ActiveMQ you can use its message grouping support
http://activemq.apache.org/message-groups.html

And Apache Kafka has topics where you can use partition ids to group
by A, B, C and then have parallel processing of each group.

Not sure SQS has such advanced features.

Its a bit brittle to have to write java code or split / aggregate eip
patterns in camel to do this, and if you want to have some level of
transaction guarantees. All EIP patterns in camel are transient and
dont persist their state etc.

in other words try to use a better messaging system.


On Tue, Apr 11, 2017 at 3:31 PM, Artur Jablonski
<[hidden email]> wrote:

> I guess one thing that come to my mind is to hide all this parallel stuff
> inside a processor, that would just spit out on the other end the result of
> processing all those messages. It would handle grouping and serializing and
> stuff I guess that would reduce the complexity of the route with a cost of
> complexity of the processor. I have no better ideas anyway, so I will give
> it a go
>
> On Tue, Apr 11, 2017 at 10:45 AM, Artur Jablonski <[hidden email]>
> wrote:
>
>> Hello,
>>
>> I don't think this route definition is fitting my use case, though I
>> learnt a thing or two about the interesting patterns linked. Thanks!
>>
>> Ok, so let me try to clarify the use case.
>>
>>
>> 1. The stream is infinite, it's not a batch job. The messages keep on
>> coming from SQS 'all the time'
>>
>> 2. More important thing is about parallel processing.
>>
>> Let A1 denote a message 1 from group A, B2 message 2 from group B, etc.
>>
>> Let's say this is the order in which the messages happen to appear
>>  in the route from SQS
>>
>> A1, A2, B1, C1, B2, A3, C2, B3
>>
>> Now what I am trying to achieve is grouping the messages that have to be
>> processed sequentially (order doesn't matter as long as no two messages
>> from the same group are processed at the same time).
>> So I am trying to somehow get these streams
>>
>> A1, A2, A3
>>
>> B1, B2, B3
>>
>> C1, C2
>>
>>
>> So, A1 B1 and C1 can be processed in parallel because they are from
>> different groups, but the messages within groups need to be processed one
>> by one.
>>
>> In my example, there are 3 groups, but there can be many and I don't know
>> what they are in advance. The processing logic between the groups is
>> similar and is a function of the group so I can get a processor for group A
>> from a method call getProcessor(A), B getProcessor(B), etc.
>>
>> I am stuck at how to do that in Camel, because since I don't know the
>> groups in advance, I would need to create processing routes dynamically.
>>
>> Say the system starts, and A1 arrives, there can't be any processor for
>> group A yet, since it's the first message from the group and I need to
>> somehow dynamically add processing capability of the group A to the route
>> and then perhaps if the messages from group A stop arriving for some time,
>> that processor could be removed.
>>
>> How to add the parallel part between the group messages is also blurry to
>> me. One way of doing this I was thinking was to do a multicast to all the
>> dynamically created processing routes for groups and stick a filter before
>> so that only messages from particular group can go through. From multicast
>> page:
>>
>> from("direct:a").multicast().parallelProcessing().to("direct:x",
>> "direct:y", "direct:z");
>>
>> But here the x,y,z endpoints are hardcoded. I could write up some custom
>> multicast I suppose to search the routes in CamelContext...... not sure.
>>
>> Thanks
>> Artur
>>
>>
>>
>>
>>
>> On Mon, Apr 3, 2017 at 1:36 PM, Artur Jablonski <[hidden email]>
>> wrote:
>>
>>> Hi Zoran,
>>>
>>> Thank you for such detailed response. This looks very promising. i will
>>> need to get my head around the aggregator pattern.
>>> For this week I will be busy with other tasks, but I will get back to it
>>> as soon as I can to see if I can get Camel work for the use case.
>>>
>>> Cheerio
>>> Artur
>>>
>>> On Mon, Apr 3, 2017 at 11:09 AM, Zoran Regvart <[hidden email]> wrote:
>>>
>>>> Hi Artur,
>>>> I was thinking that the order of the messages would be important as
>>>> you need to process them sequentially.
>>>>
>>>> So I think you could use the dynamic message routing[1] with
>>>> aggregator[2], something like:
>>>>
>>>>     from("aws-sqs:...")
>>>>         .process("#preProcess")
>>>>         .toD("direct:${header.nextRoute}");
>>>>
>>>>     from("direct:parallel")...;
>>>>     from("direct:sequential").aggregate(simple("${header.group}"
>>>> )).completion..;
>>>>
>>>> So from yout SQS queue you would use a processor to pre-process
>>>> message whose responsibility would be to set the (custom) `nextRoute`
>>>> and (custom) `group` headers. `nextRoute` would be `parallel` or
>>>> `sequential`, and if `sequential` the messages would be aggregated
>>>> using the `group` header.
>>>>
>>>> You would want to define your own custom aggregation strategy or use
>>>> the completion* options that are available to you. There also might be
>>>> need to use seda[3] to fine tune any parallel processing. You might
>>>> throw in there a data format unmarshaller[4] instead of the
>>>> `preProcess` processor and use something like `${body.xyz} == foo` in
>>>> the `toD` expression.
>>>>
>>>> And I would guess that you need to examine transactions or persistence
>>>> at some point also in case your aggregation step runs for a long time
>>>> or if your use case is sensitive to message loss if interrupted --
>>>> which would undoubtedly lead you back to using queues to separate
>>>> those two ways of processing,
>>>>
>>>> HTH,
>>>>
>>>> zoran
>>>>
>>>> [1] https://camel.apache.org/message-endpoint.html
>>>> [2] https://camel.apache.org/aggregator2.html
>>>> [3] https://camel.apache.org/seda.html
>>>> [4] https://camel.apache.org/data-format.html
>>>>
>>>> On Sat, Apr 1, 2017 at 1:09 PM, Artur Jablonski
>>>> <[hidden email]> wrote:
>>>> > Hey Zoran.
>>>> >
>>>> > I read again the patterns you mentioned. In my use case the order of
>>>> > processing within a group doesn't matter as long as two messages from
>>>> the
>>>> > same group are never processed in parallel. So i guess resenquencer is
>>>> out
>>>> > of the picture unless I didn't get the intention.
>>>> >
>>>> > So what we are left with is the content based router. Sure. The message
>>>> > comes, i can see what group it belongs two... And what next? Perhaps
>>>> it's
>>>> > the very first message from that group so I would need to trigger
>>>> creating
>>>> > route/processor for that group somehow, perhaps messages from this
>>>> group
>>>> > were processed before in which case the processor for the group should
>>>> > already exist...
>>>> >
>>>> >
>>>> >
>>>> >
>>>> > On 31 Mar 2017 7:58 p.m., "Zoran Regvart" <[hidden email]> wrote:
>>>> >
>>>> >> Hi Artur,
>>>> >> have a look at Camel EIP page[1], what you describe sounds to me like
>>>> >> Resequencer and Content based router patterns,
>>>> >>
>>>> >> zoran
>>>> >>
>>>> >> [1] https://camel.apache.org/eip.html
>>>> >>
>>>> >> On Fri, Mar 31, 2017 at 5:08 PM, Artur Jablonski
>>>> >> <[hidden email]> wrote:
>>>> >> > Hello.
>>>> >> >
>>>> >> > I wonder if someone could push me in the right direction trying to
>>>> >> express
>>>> >> > quite curious case in Camel route.
>>>> >> >
>>>> >> > Imagine there's a stream of messages some of which can be processed
>>>> in
>>>> >> > parallel and some have to be processed serially. You can group the
>>>> >> messages
>>>> >> > that require serial processing together by looking at the message
>>>> body.
>>>> >> You
>>>> >> > don't know upfront how many groups can occur in the stream.
>>>> >> >
>>>> >> > The way I thought about doing this is having a route for each
>>>> message
>>>> >> > group. Since I don't know upfront how many and what groups there
>>>> will be
>>>> >> > then I would need to create routes dynamically. If a message comes
>>>> >> > belonging to a group that doesn't have it's handling route, then i
>>>> could
>>>> >> > create it (is that even possible??) Then if there's no messages
>>>> coming
>>>> >> for
>>>> >> > a given group in some time I could remove the route for the group to
>>>> >> > cleanup (is that possible?)
>>>> >> >
>>>> >> > New to Camel
>>>> >> >
>>>> >> > Thx!
>>>> >> > Artur
>>>> >>
>>>> >>
>>>> >>
>>>> >> --
>>>> >> Zoran Regvart
>>>> >>
>>>>
>>>>
>>>>
>>>> --
>>>> Zoran Regvart
>>>>
>>>
>>>
>>



--
Claus Ibsen
-----------------
http://davsclaus.com @davsclaus
Camel in Action 2: https://www.manning.com/ibsen2
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

Re: Curious routing case

Henrique Viecili-2
Hi Artur,

As your group set is unbounded, you may need to look at creating
(in-memory) queues and routes at runtime and attaching them to a camel
context (or spawning their own context). You can use some camel trickery
with DefaultCamelContext + RouteDefinition, it's ugly but it should work.

Regards,
Henrique



Henrique Viecili

On 12 April 2017 at 01:06, Claus Ibsen <[hidden email]> wrote:

> If you use ActiveMQ you can use its message grouping support
> http://activemq.apache.org/message-groups.html
>
> And Apache Kafka has topics where you can use partition ids to group
> by A, B, C and then have parallel processing of each group.
>
> Not sure SQS has such advanced features.
>
> Its a bit brittle to have to write java code or split / aggregate eip
> patterns in camel to do this, and if you want to have some level of
> transaction guarantees. All EIP patterns in camel are transient and
> dont persist their state etc.
>
> in other words try to use a better messaging system.
>
>
> On Tue, Apr 11, 2017 at 3:31 PM, Artur Jablonski
> <[hidden email]> wrote:
> > I guess one thing that come to my mind is to hide all this parallel stuff
> > inside a processor, that would just spit out on the other end the result
> of
> > processing all those messages. It would handle grouping and serializing
> and
> > stuff I guess that would reduce the complexity of the route with a cost
> of
> > complexity of the processor. I have no better ideas anyway, so I will
> give
> > it a go
> >
> > On Tue, Apr 11, 2017 at 10:45 AM, Artur Jablonski <
> [hidden email]>
> > wrote:
> >
> >> Hello,
> >>
> >> I don't think this route definition is fitting my use case, though I
> >> learnt a thing or two about the interesting patterns linked. Thanks!
> >>
> >> Ok, so let me try to clarify the use case.
> >>
> >>
> >> 1. The stream is infinite, it's not a batch job. The messages keep on
> >> coming from SQS 'all the time'
> >>
> >> 2. More important thing is about parallel processing.
> >>
> >> Let A1 denote a message 1 from group A, B2 message 2 from group B, etc.
> >>
> >> Let's say this is the order in which the messages happen to appear
> >>  in the route from SQS
> >>
> >> A1, A2, B1, C1, B2, A3, C2, B3
> >>
> >> Now what I am trying to achieve is grouping the messages that have to be
> >> processed sequentially (order doesn't matter as long as no two messages
> >> from the same group are processed at the same time).
> >> So I am trying to somehow get these streams
> >>
> >> A1, A2, A3
> >>
> >> B1, B2, B3
> >>
> >> C1, C2
> >>
> >>
> >> So, A1 B1 and C1 can be processed in parallel because they are from
> >> different groups, but the messages within groups need to be processed
> one
> >> by one.
> >>
> >> In my example, there are 3 groups, but there can be many and I don't
> know
> >> what they are in advance. The processing logic between the groups is
> >> similar and is a function of the group so I can get a processor for
> group A
> >> from a method call getProcessor(A), B getProcessor(B), etc.
> >>
> >> I am stuck at how to do that in Camel, because since I don't know the
> >> groups in advance, I would need to create processing routes dynamically.
> >>
> >> Say the system starts, and A1 arrives, there can't be any processor for
> >> group A yet, since it's the first message from the group and I need to
> >> somehow dynamically add processing capability of the group A to the
> route
> >> and then perhaps if the messages from group A stop arriving for some
> time,
> >> that processor could be removed.
> >>
> >> How to add the parallel part between the group messages is also blurry
> to
> >> me. One way of doing this I was thinking was to do a multicast to all
> the
> >> dynamically created processing routes for groups and stick a filter
> before
> >> so that only messages from particular group can go through. From
> multicast
> >> page:
> >>
> >> from("direct:a").multicast().parallelProcessing().to("direct:x",
> >> "direct:y", "direct:z");
> >>
> >> But here the x,y,z endpoints are hardcoded. I could write up some custom
> >> multicast I suppose to search the routes in CamelContext...... not sure.
> >>
> >> Thanks
> >> Artur
> >>
> >>
> >>
> >>
> >>
> >> On Mon, Apr 3, 2017 at 1:36 PM, Artur Jablonski <
> [hidden email]>
> >> wrote:
> >>
> >>> Hi Zoran,
> >>>
> >>> Thank you for such detailed response. This looks very promising. i will
> >>> need to get my head around the aggregator pattern.
> >>> For this week I will be busy with other tasks, but I will get back to
> it
> >>> as soon as I can to see if I can get Camel work for the use case.
> >>>
> >>> Cheerio
> >>> Artur
> >>>
> >>> On Mon, Apr 3, 2017 at 11:09 AM, Zoran Regvart <[hidden email]>
> wrote:
> >>>
> >>>> Hi Artur,
> >>>> I was thinking that the order of the messages would be important as
> >>>> you need to process them sequentially.
> >>>>
> >>>> So I think you could use the dynamic message routing[1] with
> >>>> aggregator[2], something like:
> >>>>
> >>>>     from("aws-sqs:...")
> >>>>         .process("#preProcess")
> >>>>         .toD("direct:${header.nextRoute}");
> >>>>
> >>>>     from("direct:parallel")...;
> >>>>     from("direct:sequential").aggregate(simple("${header.group}"
> >>>> )).completion..;
> >>>>
> >>>> So from yout SQS queue you would use a processor to pre-process
> >>>> message whose responsibility would be to set the (custom) `nextRoute`
> >>>> and (custom) `group` headers. `nextRoute` would be `parallel` or
> >>>> `sequential`, and if `sequential` the messages would be aggregated
> >>>> using the `group` header.
> >>>>
> >>>> You would want to define your own custom aggregation strategy or use
> >>>> the completion* options that are available to you. There also might be
> >>>> need to use seda[3] to fine tune any parallel processing. You might
> >>>> throw in there a data format unmarshaller[4] instead of the
> >>>> `preProcess` processor and use something like `${body.xyz} == foo` in
> >>>> the `toD` expression.
> >>>>
> >>>> And I would guess that you need to examine transactions or persistence
> >>>> at some point also in case your aggregation step runs for a long time
> >>>> or if your use case is sensitive to message loss if interrupted --
> >>>> which would undoubtedly lead you back to using queues to separate
> >>>> those two ways of processing,
> >>>>
> >>>> HTH,
> >>>>
> >>>> zoran
> >>>>
> >>>> [1] https://camel.apache.org/message-endpoint.html
> >>>> [2] https://camel.apache.org/aggregator2.html
> >>>> [3] https://camel.apache.org/seda.html
> >>>> [4] https://camel.apache.org/data-format.html
> >>>>
> >>>> On Sat, Apr 1, 2017 at 1:09 PM, Artur Jablonski
> >>>> <[hidden email]> wrote:
> >>>> > Hey Zoran.
> >>>> >
> >>>> > I read again the patterns you mentioned. In my use case the order of
> >>>> > processing within a group doesn't matter as long as two messages
> from
> >>>> the
> >>>> > same group are never processed in parallel. So i guess resenquencer
> is
> >>>> out
> >>>> > of the picture unless I didn't get the intention.
> >>>> >
> >>>> > So what we are left with is the content based router. Sure. The
> message
> >>>> > comes, i can see what group it belongs two... And what next? Perhaps
> >>>> it's
> >>>> > the very first message from that group so I would need to trigger
> >>>> creating
> >>>> > route/processor for that group somehow, perhaps messages from this
> >>>> group
> >>>> > were processed before in which case the processor for the group
> should
> >>>> > already exist...
> >>>> >
> >>>> >
> >>>> >
> >>>> >
> >>>> > On 31 Mar 2017 7:58 p.m., "Zoran Regvart" <[hidden email]>
> wrote:
> >>>> >
> >>>> >> Hi Artur,
> >>>> >> have a look at Camel EIP page[1], what you describe sounds to me
> like
> >>>> >> Resequencer and Content based router patterns,
> >>>> >>
> >>>> >> zoran
> >>>> >>
> >>>> >> [1] https://camel.apache.org/eip.html
> >>>> >>
> >>>> >> On Fri, Mar 31, 2017 at 5:08 PM, Artur Jablonski
> >>>> >> <[hidden email]> wrote:
> >>>> >> > Hello.
> >>>> >> >
> >>>> >> > I wonder if someone could push me in the right direction trying
> to
> >>>> >> express
> >>>> >> > quite curious case in Camel route.
> >>>> >> >
> >>>> >> > Imagine there's a stream of messages some of which can be
> processed
> >>>> in
> >>>> >> > parallel and some have to be processed serially. You can group
> the
> >>>> >> messages
> >>>> >> > that require serial processing together by looking at the message
> >>>> body.
> >>>> >> You
> >>>> >> > don't know upfront how many groups can occur in the stream.
> >>>> >> >
> >>>> >> > The way I thought about doing this is having a route for each
> >>>> message
> >>>> >> > group. Since I don't know upfront how many and what groups there
> >>>> will be
> >>>> >> > then I would need to create routes dynamically. If a message
> comes
> >>>> >> > belonging to a group that doesn't have it's handling route, then
> i
> >>>> could
> >>>> >> > create it (is that even possible??) Then if there's no messages
> >>>> coming
> >>>> >> for
> >>>> >> > a given group in some time I could remove the route for the
> group to
> >>>> >> > cleanup (is that possible?)
> >>>> >> >
> >>>> >> > New to Camel
> >>>> >> >
> >>>> >> > Thx!
> >>>> >> > Artur
> >>>> >>
> >>>> >>
> >>>> >>
> >>>> >> --
> >>>> >> Zoran Regvart
> >>>> >>
> >>>>
> >>>>
> >>>>
> >>>> --
> >>>> Zoran Regvart
> >>>>
> >>>
> >>>
> >>
>
>
>
> --
> Claus Ibsen
> -----------------
> http://davsclaus.com @davsclaus
> Camel in Action 2: https://www.manning.com/ibsen2
>
Loading...