jms:channel removes headers

[Ivan Velykorodnyy]

Hi Team!

Can you please advise what's the easiest way to pass headers via jms:channel?
To be precise I'd like to sort "no output-channel or replyChannel header available". I'm using SI 2.2 release.

Thanks

[Gary Russell]

You cannot persist "live" objects such as the replyChannel. The inbound thread is waiting to receive a message on that object; the object cannot be persisted so, when the message is retrieved from the channel, it has no replyChannel header, and hence you get that error.

If you must persist the message, you need to change the architecture to use async messaging instead of request/reply.

[Ivan Velykorodnyy]

Cool, thanks for explanation.

How does correlation work for async gateway?

I switched to asynch gateway but it works (as synch one) only in the case I have intublish-subscribe-channel instead of jms:channel.

Thanks

PS and yes, the error "no output-channel or replyChannel header available" disappeared

PS2 I've tried to pass headers around via jms:interceptor but postReceive is not called :\

[Gary Russell]

Hmmm. Async gateways work the same way - you still need a replyChannel to get the reply back to its future. So that, alone, shouldn't change anything. Can you show your config.

postReceive() is only called if the JMS channel is not message-driven (subscribable); it has to be a pollable channel.

[Ivan Velykorodnyy]

Hi Gary,

Shortened config looks as follows:

    <si:gateway id="theService"
    	service-interface="com.xxx.integration.TheGateway"
    	default-request-channel="userChangesChannel"
    	default-reply-channel="resultsChannel"
    	/>

    <si:channel id="userChangesChannel" datatype="com.xxx.model.Item"/>

 	<si:chain input-channel="userChangesChannel"
			output-channel="synchQueueChannel">
...
	</si:chain>

 	<jms:channel id="synchQueueChannel"
		queue="internalQueue"
		connection-factory="myCF"
		acknowledge="transacted"
		concurrency="1-5"
		error-handler="internalEH"
		container-class="com.xxx.integration.operations.ExceptionsAwareMessageListenerContainer"
		>
		<jms:interceptors>
			<bean class="com.xxx.integration.operations.CopyHeadersChannelInterceptor"/>
		</jms:interceptors>
	</jms:channel>

	<bean id="internalEH" class="org.springframework.integration.channel.MessagePublishingErrorHandler">
		<property name="defaultErrorChannel" ref="processingErrorChannel"/>
	</bean>

 	<si:chain input-channel="synchQueueChannel"
			output-channel="resultsChannel">
...
	</si:chain>

    <!--  error flow   -->

 	<si:channel id="processingErrorChannel"/>

 	<si:chain input-channel="synchQueueChannel"
			output-channel="resultsChannel">
...
	</si:chain>

    <si:publish-subscribe-channel id="resultsChannel" datatype="com.xxx.model.Result"/>

The main idea behind jms:channel is to synchronize concurrent modifications. The problematic flow is listed above and other concurrent flows are omitted as they are one way and I don't see issues there. But this flow is triggered via @Controller and I'd like to give a meaningful reply on request whether there was success or failure. And that worked before I introduced jms:channel. Also I faced some issues to make "no rollback on some exceptions" logic (so I hacked that with custom MessageListenerContainer) but this is likely for a separate thread, I'd like to fix successful flow first.

And I prefer to stay with jms queue rather than topic.

Thank you

[Ivan Velykorodnyy]

Can someone please advise on what is the most "correct" or easiest way to sort the issue with filtered headers after jms:channel : I'll appreciate any advise. Thanks!

[Gary Russell]

It is still not clear what you are trying to achieve.

You mentioned an interceptor above - if you change the jms channel to 'message-driven="false"', your postReceive() will be called, but you will need a poller on your <chain/> to pull messages from the channel.

If you don't want to go that route (message-driven is more efficient), you can "save" the header in some registry (keyed by something you can correlate on), invoked by a <service-activator/> and then restore it with a <header-enricher/>. Or, you can use the technique used in the tcp-client-server-multiplex sample, which sends a copy of the input message (and the response) to an aggregator, which restores the header for you, followed by a transformer to select the response from the aggregated message.

Of course, none of these techniques will work if you scale out horizontally (unless each instance has its own jms queue to back the channel).

Plus, of course, if the JVM goes down while there are messages in the queue, any such messages will fail when the JVM is restarted.

[Ivan Velykorodnyy]

Thanks for such a comprehensive response! I'll think about these options, there are so many

For now I just want messages to survive jvm shutdown and benefit from transactional capabilities of jms so failed messages are not lost. It's fine to execute write flow on the single instance as we need only read interface to be highly-available and possible to scale.