It works, but for each group of messages, you need to generate a unique id for the SPLITTER_CORRID property. The easiest way would be to use the id of the incoming exchange (which will always be unique).
On Wed, Feb 27, 2008 at 11:15 AM, sachin2008 <[EMAIL PROTECTED]> wrote: > > Thanks for your reply.. > > For aggregator how should i set correlationid,index and count. > > In my usecase i already said that: > I have hardcoded like this...... > Component A: > in.setProperty(AbstractSplitter.SPLITTER_COUNT, new Integer(3)); > in.setProperty(AbstractSplitter.SPLITTER_CORRID, "id"); > in.setProperty(AbstractSplitter.SPLITTER_INDEX, new Integer(0)); > Component B: > in.setProperty(AbstractSplitter.SPLITTER_COUNT, new Integer(3)); > in.setProperty(AbstractSplitter.SPLITTER_CORRID, "id"); > in.setProperty(AbstractSplitter.SPLITTER_INDEX, new Integer(1)); > Component C: > in.setProperty(AbstractSplitter.SPLITTER_COUNT, new Integer(3)); > in.setProperty(AbstractSplitter.SPLITTER_CORRID, "id"); > in.setProperty(AbstractSplitter.SPLITTER_INDEX, new Integer(2)); > > Can you please help me in configuring the aggregator properities... > > > > > gnodet wrote: > > > > The aggregator uses three properties: the correlationId (which > > identifies messages related together), the index and the count. > > Each group of messages must have a different correlationId, while all > > the messages in a given group must have the same correlationId. Then, > > inside a group, no two messages can have the same index. > > You need to make sure these rules are followed, or you'll have to hack > > your own aggregation strategy. > > > > By checking the log at DEBUG level (or remote debugging the > > aggregator), you should be able to see what message are received, > > hence the cause of the error. > > > > On Wed, Feb 27, 2008 at 10:23 AM, sachin2008 <[EMAIL PROTECTED]> wrote: > >> > >> Presently i am able to aggregate the messages through aggregator. > >> > >> But there is a problem..... > >> > >> First of all i will explain my usecase: > >> > >> JMSConsumer------>static > >> receipientlist----->lw-container------>aggregator------>JMSProvider > >> > >> I am sending a request from JMSconsumer to static receipient list. from > >> static receipientlist i am sending three inonly messages to > >> lw-container.In > >> lw-container i have used three components namely A,B and C. From > >> component > >> A, i am sending an inonly message to aggregator by setting index as > >> 0,count > >> as 3 and corelationid as id and From Component B, i am sending an > >> inonly > >> message to aggregator by setting index as 1,count as 3 and corelationid > >> as > >> id and From Component C, i am sending an inonly message to aggregator by > >> setting index as 2,count as 3 and corelationid as id.I am able to get > >> the > >> aggregated message from the aggregator to JMSProvider. > >> > >> Problems: > >> I am getting aggregated message in JMSProvider successfully only for the > >> first request . > >> > >> But if i have given another request it is giving some exception like: > >> > >> > >> ERROR - EIPComponent - Error processing exchange > >> InOnly[ > >> id: ID:pc007869-2244-1204099875217-12:313 > >> status: Active > >> role: provider > >> endpoint: aggregate > >> in: <?xml version="1.0" encoding="UTF-8"?><EOS><PAYL_HDR> <DUNS_NBR> > >> 313955098 > >> </DUNS_NBR></PAYL_HDR></EOS> > >> ] > >> java.lang.IllegalStateException: Message with index 0 has already been > >> received > >> at > >> org.apache.servicemix.eip.patterns.SplitAggregator.addMessage(SplitAg > >> gregator.java:213) > >> > >> at > >> org.apache.servicemix.eip.support.AbstractAggregator.process(Abstract > >> Aggregator.java:159) > >> > >> > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.doProcess(AsyncBaseLi > >> feCycle.java:489) > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.processExchange(Async > >> BaseLifeCycle.java:441) > >> at > >> org.apache.servicemix.common.BaseLifeCycle.onMessageExchange(BaseLife > >> Cycle.java:46) > >> at > >> org.apache.servicemix.jbi.messaging.DeliveryChannelImpl.processInBoun > >> d(DeliveryChannelImpl.java:593) > >> at > >> org.apache.servicemix.jbi.nmr.flow.AbstractFlow.doRouting(AbstractFlo > >> w.java:174) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaFlow.doRouting(SedaFlow.j > >> ava:176) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaQueue$1.run(SedaQueue.jav > >> a:134) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.runTask(ThreadPoolExecutor.java:665) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.run(ThreadPoolExecutor.java:690) > >> at java.lang.Thread.run(Thread.java:595) > >> ERROR - EIPComponent - Error processing exchange > >> InOnly[ > >> id: ID:pc007869-2244-1204099875217-13:286 > >> status: Active > >> role: provider > >> endpoint: aggregate > >> in: <?xml version="1.0" encoding="UTF-8"?><EWOW><FAM_TREE_LINK> <ASSN> > >> <ASSN_T > >> YPE_CD>1019</ASSN_TYPE_CD></ASSN></FAM_TREE_LINK></EWOW> > >> ] > >> java.lang.IllegalStateException: Message with index 1 has already been > >> received > >> at > >> org.apache.servicemix.eip.patterns.SplitAggregator.addMessage(SplitAg > >> gregator.java:213) > >> > >> at > >> org.apache.servicemix.eip.support.AbstractAggregator.process(Abstract > >> Aggregator.java:159) > >> > >> > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.doProcess(AsyncBaseLi > >> feCycle.java:489) > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.processExchange(Async > >> BaseLifeCycle.java:441) > >> at > >> org.apache.servicemix.common.BaseLifeCycle.onMessageExchange(BaseLife > >> Cycle.java:46) > >> at > >> org.apache.servicemix.jbi.messaging.DeliveryChannelImpl.processInBoun > >> d(DeliveryChannelImpl.java:593) > >> at > >> org.apache.servicemix.jbi.nmr.flow.AbstractFlow.doRouting(AbstractFlo > >> w.java:174) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaFlow.doRouting(SedaFlow.j > >> ava:176) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaQueue$1.run(SedaQueue.jav > >> a:134) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.runTask(ThreadPoolExecutor.java:665) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.run(ThreadPoolExecutor.java:690) > >> at java.lang.Thread.run(Thread.java:595) > >> ERROR - EIPComponent - Error processing exchange > >> InOnly[ > >> id: ID:pc007869-2244-1204099875217-14:293 > >> status: Active > >> role: provider > >> endpoint: aggregate > >> in: <?xml version="1.0" encoding="UTF-8"?><MI><PAYL_HDR> <DUNS_NBR> > >> 313955098< > >> /DUNS_NBR></PAYL_HDR></MI> > >> ] > >> java.lang.IllegalStateException: Message with index 2 has already been > >> received > >> at > >> org.apache.servicemix.eip.patterns.SplitAggregator.addMessage(SplitAg > >> gregator.java:213) > >> > >> at > >> org.apache.servicemix.eip.support.AbstractAggregator.process(Abstract > >> Aggregator.java:159) > >> > >> > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.doProcess(AsyncBaseLi > >> feCycle.java:489) > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.processExchange(Async > >> BaseLifeCycle.java:441) > >> at > >> org.apache.servicemix.common.BaseLifeCycle.onMessageExchange(BaseLife > >> Cycle.java:46) > >> at > >> org.apache.servicemix.jbi.messaging.DeliveryChannelImpl.processInBoun > >> d(DeliveryChannelImpl.java:593) > >> at > >> org.apache.servicemix.jbi.nmr.flow.AbstractFlow.doRouting(AbstractFlo > >> w.java:174) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaFlow.doRouting(SedaFlow.j > >> ava:176) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaQueue$1.run(SedaQueue.jav > >> a:134) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.runTask(ThreadPoolExecutor.java:665) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.run(ThreadPoolExecutor.java:690) > >> at java.lang.Thread.run(Thread.java:595) > >> ERROR - EIPComponent - Error processing exchange > >> InOnly[ > >> id: ID:pc007869-2244-1204099875217-12:315 > >> status: Active > >> role: provider > >> endpoint: aggregate > >> in: <?xml version="1.0" encoding="UTF-8"?><EOS><PAYL_HDR> <DUNS_NBR> > >> 313955098 > >> </DUNS_NBR></PAYL_HDR></EOS> > >> ] > >> java.lang.IllegalStateException: Message with index 0 has already been > >> received > >> at > >> org.apache.servicemix.eip.patterns.SplitAggregator.addMessage(SplitAg > >> gregator.java:213) > >> > >> at > >> org.apache.servicemix.eip.support.AbstractAggregator.process(Abstract > >> Aggregator.java:159) > >> > >> > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.doProcess(AsyncBaseLi > >> feCycle.java:489) > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.processExchange(Async > >> BaseLifeCycle.java:441) > >> at > >> org.apache.servicemix.common.BaseLifeCycle.onMessageExchange(BaseLife > >> Cycle.java:46) > >> at > >> org.apache.servicemix.jbi.messaging.DeliveryChannelImpl.processInBoun > >> d(DeliveryChannelImpl.java:593) > >> at > >> org.apache.servicemix.jbi.nmr.flow.AbstractFlow.doRouting(AbstractFlo > >> w.java:174) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaFlow.doRouting(SedaFlow.j > >> ava:176) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaQueue$1.run(SedaQueue.jav > >> a:134) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.runTask(ThreadPoolExecutor.java:665) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.run(ThreadPoolExecutor.java:690) > >> at java.lang.Thread.run(Thread.java:595) > >> ERROR - EIPComponent - Error processing exchange > >> InOnly[ > >> id: ID:pc007869-2244-1204099875217-12:317 > >> status: Active > >> role: provider > >> endpoint: aggregate > >> in: <?xml version="1.0" encoding="UTF-8"?><EOS><PAYL_HDR> <DUNS_NBR> > >> 313955098 > >> </DUNS_NBR></PAYL_HDR></EOS> > >> ] > >> java.lang.IllegalStateException: Message with index 0 has already been > >> received > >> at > >> org.apache.servicemix.eip.patterns.SplitAggregator.addMessage(SplitAg > >> gregator.java:213) > >> > >> at > >> org.apache.servicemix.eip.support.AbstractAggregator.process(Abstract > >> Aggregator.java:159) > >> > >> > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.doProcess(AsyncBaseLi > >> feCycle.java:489) > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.processExchange(Async > >> BaseLifeCycle.java:441) > >> at > >> org.apache.servicemix.common.BaseLifeCycle.onMessageExchange(BaseLife > >> Cycle.java:46) > >> at > >> org.apache.servicemix.jbi.messaging.DeliveryChannelImpl.processInBoun > >> d(DeliveryChannelImpl.java:593) > >> at > >> org.apache.servicemix.jbi.nmr.flow.AbstractFlow.doRouting(AbstractFlo > >> w.java:174) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaFlow.doRouting(SedaFlow.j > >> ava:176) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaQueue$1.run(SedaQueue.jav > >> a:134) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.runTask(ThreadPoolExecutor.java:665) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.run(ThreadPoolExecutor.java:690) > >> at java.lang.Thread.run(Thread.java:595) > >> ERROR - EIPComponent - Error processing exchange > >> InOnly[ > >> id: ID:pc007869-2244-1204099875217-12:319 > >> status: Active > >> role: provider > >> endpoint: aggregate > >> in: <?xml version="1.0" encoding="UTF-8"?><EOS><PAYL_HDR> <DUNS_NBR> > >> 313955098 > >> </DUNS_NBR></PAYL_HDR></EOS> > >> ] > >> java.lang.IllegalStateException: Message with index 0 has already been > >> received > >> at > >> org.apache.servicemix.eip.patterns.SplitAggregator.addMessage(SplitAg > >> gregator.java:213) > >> > >> at > >> org.apache.servicemix.eip.support.AbstractAggregator.process(Abstract > >> Aggregator.java:159) > >> > >> > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.doProcess(AsyncBaseLi > >> feCycle.java:489) > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.processExchange(Async > >> BaseLifeCycle.java:441) > >> at > >> org.apache.servicemix.common.BaseLifeCycle.onMessageExchange(BaseLife > >> Cycle.java:46) > >> at > >> org.apache.servicemix.jbi.messaging.DeliveryChannelImpl.processInBoun > >> d(DeliveryChannelImpl.java:593) > >> at > >> org.apache.servicemix.jbi.nmr.flow.AbstractFlow.doRouting(AbstractFlo > >> w.java:174) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaFlow.doRouting(SedaFlow.j > >> ava:176) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaQueue$1.run(SedaQueue.jav > >> a:134) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.runTask(ThreadPoolExecutor.java:665) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.run(ThreadPoolExecutor.java:690) > >> at java.lang.Thread.run(Thread.java:595) > >> ERROR - EIPComponent - Error processing exchange > >> InOnly[ > >> id: ID:pc007869-2244-1204099875217-13:288 > >> status: Active > >> role: provider > >> endpoint: aggregate > >> in: <?xml version="1.0" encoding="UTF-8"?><EWOW><FAM_TREE_LINK> <ASSN> > >> <ASSN_T > >> YPE_CD>1019</ASSN_TYPE_CD></ASSN></FAM_TREE_LINK></EWOW> > >> ] > >> java.lang.IllegalStateException: Message with index 1 has already been > >> received > >> at > >> org.apache.servicemix.eip.patterns.SplitAggregator.addMessage(SplitAg > >> gregator.java:213) > >> > >> at > >> org.apache.servicemix.eip.support.AbstractAggregator.process(Abstract > >> Aggregator.java:159) > >> > >> > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.doProcess(AsyncBaseLi > >> feCycle.java:489) > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.processExchange(Async > >> BaseLifeCycle.java:441) > >> at > >> org.apache.servicemix.common.BaseLifeCycle.onMessageExchange(BaseLife > >> Cycle.java:46) > >> at > >> org.apache.servicemix.jbi.messaging.DeliveryChannelImpl.processInBoun > >> d(DeliveryChannelImpl.java:593) > >> at > >> org.apache.servicemix.jbi.nmr.flow.AbstractFlow.doRouting(AbstractFlo > >> w.java:174) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaFlow.doRouting(SedaFlow.j > >> ava:176) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaQueue$1.run(SedaQueue.jav > >> a:134) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.runTask(ThreadPoolExecutor.java:665) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.run(ThreadPoolExecutor.java:690) > >> at java.lang.Thread.run(Thread.java:595) > >> ERROR - EIPComponent - Error processing exchange > >> InOnly[ > >> id: ID:pc007869-2244-1204099875217-14:295 > >> status: Active > >> role: provider > >> endpoint: aggregate > >> in: <?xml version="1.0" encoding="UTF-8"?><MI><PAYL_HDR> <DUNS_NBR> > >> 313955098< > >> /DUNS_NBR></PAYL_HDR></MI> > >> ] > >> java.lang.IllegalStateException: Message with index 2 has already been > >> received > >> at > >> org.apache.servicemix.eip.patterns.SplitAggregator.addMessage(SplitAg > >> gregator.java:213) > >> > >> at > >> org.apache.servicemix.eip.support.AbstractAggregator.process(Abstract > >> Aggregator.java:159) > >> > >> > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.doProcess(AsyncBaseLi > >> feCycle.java:489) > >> at > >> org.apache.servicemix.common.AsyncBaseLifeCycle.processExchange(Async > >> BaseLifeCycle.java:441) > >> at > >> org.apache.servicemix.common.BaseLifeCycle.onMessageExchange(BaseLife > >> Cycle.java:46) > >> at > >> org.apache.servicemix.jbi.messaging.DeliveryChannelImpl.processInBoun > >> d(DeliveryChannelImpl.java:593) > >> at > >> org.apache.servicemix.jbi.nmr.flow.AbstractFlow.doRouting(AbstractFlo > >> w.java:174) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaFlow.doRouting(SedaFlow.j > >> ava:176) > >> at > >> org.apache.servicemix.jbi.nmr.flow.seda.SedaQueue$1.run(SedaQueue.jav > >> a:134) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.runTask(ThreadPoolExecutor.java:665) > >> at > >> edu.emory.mathcs.backport.java.util.concurrent.ThreadPoolExecutor$Wor > >> ker.run(ThreadPoolExecutor.java:690) > >> at java.lang.Thread.run(Thread.java:595) > >> > >> > >> Can anyone please help me to resolve this issue... > >> > >> Thanks in advanced. > >> > >> > >> > >> > >> gnodet wrote: > >> > > >> > The jsr181 component can be configured with a given instance of the > >> > class that will be used to process all the incoming exchanges. You > >> > just need to make sure your code is designed to be used that way: a > >> > single instance will receive all calls concurrently, as in the > >> > servlets world. The servicemix-bean behaves the same way. > >> > > >> > Another way would be to implement your own aggregation strategy using > >> > servicemix-eip or camel. > >> > Servicemix-eip already provides some aggregation and can be extended > >> > (but does not use any jaxb marshaling, so it depends if you need it or > >> > not). Camel is really flexible and powerful and it may be worth to > >> > take a look at it. > >> > > >> > On Wed, Feb 20, 2008 at 6:37 AM, sachin2008 <[EMAIL PROTECTED]> > >> wrote: > >> >> > >> >> Hi, > >> >> > >> >> I am having a scenario: > >> >> > >> >> HTTP(BC)-------> JSR----------->Aggregator---------------->HTTP(BC) > >> >> > >> >> My requirement is: > >> >> > >> >> In JSR, we need incorporate the multi threading logic for getting > >> the > >> >> data > >> >> from different data sources and by using the aggregator pattern we > >> need > >> >> to > >> >> aggregate the xml data coming from different datasources and send > >> the > >> >> response back to the HTTP(BC). > >> >> > >> >> Can anyone please tell me how to use the aggregator pattern in this > >> >> regard. > >> >> > >> >> -- > >> >> View this message in context: > >> >> > >> > http://www.nabble.com/Reg%3AAggregator-Pattern-tp15582806s12049p15582806.html > >> >> Sent from the ServiceMix - User mailing list archive at Nabble.com. > >> >> > >> >> > >> > > >> > > >> > > >> > -- > >> > Cheers, > >> > Guillaume Nodet > >> > ------------------------ > >> > Blog: http://gnodet.blogspot.com/ > >> > > >> > > >> > >> -- > >> View this message in context: > >> > http://www.nabble.com/Reg%3AAggregator-Pattern-tp15582806s12049p15709304.html > >> > >> > >> Sent from the ServiceMix - User mailing list archive at Nabble.com. > >> > >> > > > > > > > > -- > > Cheers, > > Guillaume Nodet > > ------------------------ > > Blog: http://gnodet.blogspot.com/ > > > > > > -- > View this message in context: > http://www.nabble.com/Reg%3AAggregator-Pattern-tp15582806s12049p15710141.html > > > Sent from the ServiceMix - User mailing list archive at Nabble.com. > > -- Cheers, Guillaume Nodet ------------------------ Blog: http://gnodet.blogspot.com/
