Hi
I have a synchronization issue, which I wasn't able to sort out with core
camel components. I was wondering someone has a better solution.
I had a synchronous route:
<when>
<condition>daily close message</condition>
<to uri="direct:doAggregation/>
</when>
<otherwise>
<to uri="direct:storeMessages">
</otherwise>
Storing messages, happens all day. The order of the store messages doesn't
count, but the close message does. We shouldn't begin to process it, untill
all messages are stored. We shouldn't store new messages, untill aggregation
happen.
The service began to receive large inflow and we decided to make storing
parallel.
The route reads a durrable. So we wasn't able to consume asynchronusly. What
I came up with.
<when>
<condition>daily close message</condition>
<to uri="direct:doAggregation/>
</when>
<otherwise>
<to uri="seda:storeMessages">
</otherwise>
But now I have a race condition. If the daily report arrives, there could be
messages still under processing. Earlier it didn't existed a the same thread
was responsible for both scenario. So the report should wait untill all
message got consumed.
I've implemented a bean for synchronizing. It has a method, where workers
registers them self and another where they unregister. A third one, where
the aggregator, checks is there registered workers and if there is, it waits
untill all worker becomes unregistered. (Maybe a processor would be more
better, but I'm not really confident with the solution).
So now the solution looks like this:
<when>
<condition>daily close message</condition>
<bean id="workerSynch" methodName="waitForOutStandingWork"/>
<to uri="direct:doAggregation/>
</when>
<otherwise>
<doTry>
<bean id="workerSynch" methodName="registerWork"/>
</doTry>
<to uri="seda:myWork"/>
</otherwise>
<from uri="seda:myWork"/>
<doTry>
<to uri="direct:storeMessages/>
</doTry>
<doFinally>
<bean id="workerSynch" methodName="unRegisterWork"/>
</doFinally>
But I'm not satisfied with this. I'm worried about that the synchronization
spread through route definitions and later someone breaks it.
I'm not 100% sure that this is bullett proof. Like what happens that <to
uri="seda:myWork"/> throws excpetion. Than work stays unregistered.
I think the whole solution for this is to actually implement a new
requirement. Like previously we say process each messages one by one, but
now we have a new requirement, that process messages unless it is a close
down messages. Than the close down message should receive some sort of
business logic, where it can verify it self and act accordingly.
But I'd like to ask, that is there a way to sort this on function level,
like don't process untill other route work queue is empty instead
implementing a new functionality?
--
View this message in context:
http://camel.465427.n5.nabble.com/Synchronizing-event-with-asynchronous-route-tp5501946p5501946.html
Sent from the Camel - Users mailing list archive at Nabble.com.
