Topic Message Replies Aggregation

[fsalceda]

Hi,  the following camel context doesn't work as I expected, because the aggregation isn't working or never called, only the first suscriber reply is returned and no more replies are processed:

Is there anybody who could help me? I would appreciate it a lot, thx.

Fernando.

PD.: Using Fuse ESB (4.4.1-fuse-01-06)

Edited by: fsalceda on Jan 11, 2012 12:03 PM

[davsclaus]

You are using JMS correlation ID to correlate. So you must make sure the messages uses the same correlation id.

If you do not explicit provide a JMSCorrelationID, then an uid will be auto generated.

If you for example enable logging of headers in the log endpoint, showHeaders=true, then you can see if the correlation id is the same or different.

http://fusesource.com/docs/router/2.8/component_ref/_IDU_Log.html

[fsalceda]

Thanks for your reply, I logged the header and the second suscriber is sending exactly the same message including header, so I tried to change the correlationExpresion to

and the behavior didn't change. I always get an instantly response from the service with the first message, the rest of them cause the following exception call:

4:16:47,234 | WARN  | tenerContainer-1 | TemporaryQueueReplyManager       | 87 - org.apache.camel.camel-jms - 2.8.0.fuse-01-06 | Reply received for unknown correlationID ID:csp-fuse-devel-36603-1324570921095-14:3:1:1:1 -> ActiveMQTextMessage {commandId = 7, responseRequired = false, messageId = ID:fsalceda-desktop-58595-1324646209494-2:1:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:fsalceda-desktop-58595-1324646209494-2:1:1:1, destination = temp-queue://ID:csp-fuse-devel-36603-1324570921095-14:1:1, transactionId = TX:ID:fsalceda-desktop-58595-1324646209494-2:1:1, expiration = 0, timestamp = 1324646229224, arrival = 0, brokerInTime = 1324646207218, brokerOutTime = 1324646207224, correlationId = ID:csp-fuse-devel-36603-1324570921095-14:3:1:1:1, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, text = A

}

14:16:47,234 | WARN  | tenerContainer-1 | DefaultMessageListenerContainer  | 88 - org.springframework.jms - 3.0.5.RELEASE | Execution of JMS message listener failed, and no ErrorHandler has been set.

org.apache.camel.component.jms.reply.UnknownReplyMessageException: Reply received for unknown correlationID ID:csp-fuse-devel-36603-1324570921095-14:3:1:1:1 -> ActiveMQTextMessage {commandId = 7, responseRequired = false, messageId = ID:fsalceda-desktop-58595-1324646209494-2:1:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:fsalceda-desktop-58595-1324646209494-2:1:1:1, destination = temp-queue://ID:csp-fuse-devel-36603-1324570921095-14:1:1, transactionId = TX:ID:fsalceda-desktop-58595-1324646209494-2:1:1, expiration = 0, timestamp = 1324646229224, arrival = 0, brokerInTime = 1324646207218, brokerOutTime = 1324646207224, correlationId = ID:csp-fuse-devel-36603-1324570921095-14:3:1:1:1, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, text = A

}

        at org.apache.camel.component.jms.reply.TemporaryQueueReplyManager.handleReplyMessage(TemporaryQueueReplyManager.java:71)[87:org.apache.camel.camel-jms:2.8.0.fuse-01-06]

        at org.apache.camel.component.jms.reply.ReplyManagerSupport.onMessage(ReplyManagerSupport.java:106)[87:org.apache.camel.camel-jms:2.8.0.fuse-01-06]

        at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:560)[88:org.springframework.jms:3.0.5.RELEASE]

        at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:498)[88:org.springframework.jms:3.0.5.RELEASE]

        at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:467)[88:org.springframework.jms:3.0.5.RELEASE]

        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:325)[88:org.springframework.jms:3.0.5.RELEASE]

        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:263)[88:org.springframework.jms:3.0.5.RELEASE

Thanks for help!

Fernando.