Spring integration service-interface gateway reply channel as shared Pub/Sub

Question

This is similar to Intermittent BridgeHandler & PublishSubscribeChannel call when gateways' reply channel is pub/sub but the scenario is different in that the reply-channel is not getting "lost". The question is what is the best resolution for my scenario.

I am using Spring integration to launch Spring batch jobs. I have a number of input routes e.g. file polling and http requests. These all route to a batch-int Job Launching Gateway. The referenced Job Launcher has a task executor so job launches are asynchronous. This gateway replies on a specified channel.

<int:gateway service-interface="c.c.c.etl.gateway.JobSubmissionService" id="jobSubmissionService" default-request-channel="jobLauchInputChannel" default-reply-channel="jobLaunchReplyChannel">
</int:gateway>

<int:bridge id="filePollerBridge" input-channel="filePollerOutputChannel" output-channel="jobLauchInputChannel" />

<batch-int:job-launching-gateway request-channel="jobLauchInputChannel" reply-channel="jobLaunchReplyChannel" job-launcher="jobLauncher">
</batch-int:job-launching-gateway>

<int:publish-subscribe-channel id="jobLaunchReplyChannel" />

<int:bridge id="jobLaunchReplyChannelBridge" input-channel="jobLaunchReplyChannel" output-channel="loggingChannel">
</int:bridge>

This specified channel 'jobLaunchReplyChannel' is pub/sub and has a logger listening to it. This channel is also used as the reply channel for a service-interface gateway.

The issue I am having is that when jobs are requested via sources that are not the gateway (e.g. the poller) the Bridge that is added by gateway throws an exception because no reply channel is set on replies.

I have resolved this by adding a header to messages sent via the gateway and filtering out messages only with this header to a new 'gatewayReplyChannel'.

<int:gateway service-interface="c.c.c.etl.gateway.JobSubmissionService" id="jobSubmissionService" default-request-channel="httpJobRequestInputChannel" default-reply-channel="jobSubmissionServiceReplyChannel">
  <int:default-header name="isJobSubmissionServiceMessage" value="true" />
</int:gateway>

<int:channel id="jobSubmissionServiceReplyChannel"></int:channel>

<int:filter id="jobSubmissionServiceReplyChannelFilter" input-channel="jobLaunchReplyChannel" expression="headers.get('isJobSubmissionServiceMessage') == null ? false : headers.get('isJobSubmissionServiceMessage')" output-channel="jobSubmissionServiceReplyChannel"
throw-exception-on-rejection="false" />

Is there a better way to do this?

Artem Bilan Artem Bilan · Accepted Answer · 2015-02-26T13:45:32

Eh.. That's an interesting issue. The main cause is because MessagingGatewaySupport creates replyMessageCorrelator endpoint for an internal BridgeHandler.

And we really have that strange behaviour when we send message to the reply-channel directly. That BridgeHandler tries to send message to the replyChannel from headers.

And we really can't do anything to prevent that logic. And can't protect that explicit reply-channel from direct messages.

I think your solution is correct. Another way to overcome that: add replyChannel header from the start of another flow (file polling in your case) just using something like this:

<header-enricher>
  <reply-channel ref="nullChannel"/>
</header-enricher>

Feel free to raise a JIRA issue on the matter and we'll take a look what we can do. At least we can document these specifics.

Spring integration service-interface gateway reply channel as shared Pub/Sub

1 Answers