Using spring integration aggregator for web services

Question

I have some services some of them are initiated from JMS messages, others are initiated from WS calls. For all of them I've implemented logging, validation, adaptation (XSD-to-Domain) and persistence solutions using spring integration. It worked until now. After some time new request types are added to the requirements. These requests have child items, which will be processed in parallel. These child items will be combined and then form a single response. I've tried aggregator, it works for JMS. I collect responses for each child item and create the main response when required.

The problem is for WS calls it doesn't work. I'm using soapUI for testing, I've sent request, logging, adaptation, aggregation etc. works but I don't see the response.

Summary: Does using aggregator in ws call cause exit from the thread of web service call and breaks request-response matching mechanism?

EDIT: As Gary Russell suggested, I'm adding details to the question:

<int-ws:inbound-gateway id="ws-inbound-gateway"
                        request-channel="requestMessagesChannelForWS"
                        reply-channel="ws-response-channel"
                        error-channel="ws-error-channel"
                        header-mapper="xmlMessageExtractorSoapHeaderMapper"/>

<int:header-enricher input-channel="requestMessagesChannelForWS" output-channel="headerEnricherChannel">
    <int:header name="from" ref="headersMapManagerService" method="getFromWSValue"/>
</int:header-enricher>

<int:header-enricher input-channel="headerEnricherChannel" output-channel="requestMessagesRoutingChannel">
    <int:header name="messageHeader" expression="payload.getMessageHeader()"/>
</int:header-enricher>

<int:recipient-list-router id="xmlMessageRouter"
                           input-channel="requestMessagesRoutingChannel">
    <int:recipient channel="messageToLogAdapterChannel"/>
    <int:recipient channel="requestMessageValidatorChannel"/>
</int:recipient-list-router>

I'm omitting messageToLogAdapterChannel and requestMessageValidatorChannel steps. After validation the flow continues here:

<int:chain input-channel="serviceAvailabilityCheckMessageChannelForAdaption">
    <int:transformer expression="@serviceAvailabilityCheckMessageAdapterService.adaptMessage(headers, payload)"/>
    <int:service-activator
            expression="@serviceAvailabilityCheckHandler.handleServiceAvailabilityCheckRequest(payload)"/>
</int:chain>

As you see there is no output-channel defined on chain. I do it manually. serviceAvailabilityCheckHandler.handleServiceAvailabilityCheckRequest method creates a process instance on a BPMN engine(Activiti). I wait a signal from this engine about process completion and create a new object(not a message) and send it to a channel.

public void processEnded(OrderData orderData) {
    sendMessageToChannel(orderData, processResultChannel);
}

<int:router input-channel="processResultChannel" expression="payload.orderClass.name">
    <int:mapping value="domain.ServiceAvailabilityOrderRequestItem"
                 channel="serviceAvailabilityCheckCompletionChannel"/>
</int:router>

<int:service-activator input-channel="serviceAvailabilityCheckCompletionChannel"
                       output-channel="serviceAvailabilityCheckResponseItemAggregationChannel"
                       ref="serviceAvailabilityCheckHandler"
                       method="onItemComplete"/>

<int:aggregator input-channel="serviceAvailabilityCheckResponseItemAggregationChannel"
                output-channel="ws-response-channel"
                method="combineResponseItems"
                ref="serviceAvailabilityCheckHandler"
                correlation-strategy="serviceAvailabilityCheckHandler"
                correlation-strategy-method="correlateResponseItems"
                release-strategy-method="isResponseComplete"
                release-strategy="serviceAvailabilityCheckHandler"/>

With this configuration my aggregator works, the items come to the aggregator and then my custom implementation(combineResponseItems, correlateResponseItems and isResponseComplete) combines the response items and send it to the ws-response-channel.

I'm getting this exception:

org.springframework.integration.support.channel.ChannelResolutionException: no output-channel or replyChannel header available

IMHO, the problem here is this: I'm getting request at handleServiceAvailabilityCheckRequest and start a process, after completion I create an object as response but spring integration can't associate the object with the request. I don't have any header or something related to the request in my response object. I can't pass request related things to the process, I need to handle it in integration layer. This is where I stuck badly.

Yes it works fine; you need to show your configuration and describe exactly what problems you are seeing. — Gary Russell
Thank you Gary. I've added details and my thoughts about the problem to the question. — GokcenG

Gary Russell Gary Russell · Accepted Answer · 2013-04-12T15:34:45

If your flows are complex, you may need to increase the reply-timeout on the inbound gateway (it defaults to 1 second).

EDIT:

"org.springframework.integration.support.channel.ChannelResolutionException: no output-channel or replyChannel header available"

This implies some component is dropping the replyChannel header.

Again, you need to show the entire configuration. You can't arbitrarily create a new message and expect the framework to know what to do with it.

It's hard to speculate what's going on without a full picture (configuration, code and debug log). I suggest you turn on debug logging and follow the message(s) through your flow. If you can't figure it out, we can help, but we should probably switch to a chat - the Stack Overflow site does not permit long drawn out discussions.

EDIT:

As discussed in the chat you are arbitrarily creating new messages; the framework cannot know that these messages are related to the input.

Using spring integration aggregator for web services

1 Answers