TCP client Gateway - but need to handle unsolicited TCP server responses

[matt_koss]

That's excellent, thanks for your help.

Matt

[matt_koss]

I just found another problem related to the above mentioned connection exceptions.

When using a pair of collaborative adapters, I am trying to route such errors back into the aggregator, so that they result in a correct exception in the caller thread.
This is proving more difficult than I thought.

I set the errorChannel on the gateway, so such errors are routed to the transformer :

	<int:gateway id="barxServerGateway"
				 service-interface="MyServerGateway"
				 default-request-channel="requestObject" errorChannel="errorObject"/>


	<int:transformer input-channel="errorObject" ref="myErrorManager"
					 method="handleException" output-channel="toAggregator"/>

	<!-- Aggregator matches incoming responses to the blocked requests and releases them -->
	<int:aggregator input-channel="toAggregator" output-channel="aggregatedResponseObject"
					expire-groups-upon-completion="true" release-strategy="customCorrelator"
					correlation-strategy="customCorrelator"/>

	<!-- Extract the second object - the response -->
	<int:transformer input-channel="aggregatedResponseObject" expression="payload.get(1)"/>

Transformer then enriches them with the correct correlation ID, like this :

	public Object handleException(Message<?> message)
	{
		Throwable throwable = (Throwable) message.getPayload();

		log.error("Got exception", throwable.getMessage());

		if (throwable instanceof MessagingException)
		{
			if ("Failed to create connection".equals(throwable.getMessage()))
			{
				loggedIn = false;
			}
		}

		MyCustomException exception = new MyCustomException ("Error during communication.");

		// construct a new message, passing in headers from the original one
		Map<String, Object> messageHeaderMap = new HashMap<String, Object>();
		messageHeaderMap.putAll(message.getHeaders());

		// copy the "messageId" header, so that the aggregator can correctly correlate
		MessageHeaders failedMessageHeaders =
				((MessagingException) throwable).getFailedMessage().getHeaders();
		messageHeaderMap.put("messageId", failedMessageHeaders.get("messageId"));
		// messageHeaderMap.put("errorChannel", failedMessageHeaders.get("errorChannel"));
		// messageHeaderMap.put("replyChannel", failedMessageHeaders.get("replyChannel"));

		Message<?> result = new ErrorMessage(exception, messageHeaderMap);

		return result;
	}

Note that the aggregator is using the "messageId" header as the correlation key.
I tried to populate just messageId field, that complains about replyChannel not present.
If I populate also errorChannel and replyChannel (from the failed message), this results in exception NOT routed to the caller.

[Gary Russell]

What you are trying to do won't work. The problem is that once the gateway gets the exception, and sends the error message, it is no longer waiting on the "old" replyChannel - it's waiting on a new replyChannel in the ErrorMessage header.

Think of the error-channel flow as the "catch" part of a "try/catch" block - you can't jump back into the try from the catch.

You can't send the new replyChannel to the aggregator because it will drop that header (because it doesn't know which one is the right one to use).

So, what you need is another "try/catch" block in the flow (within the outer "try").

This is achieved by adding another gateway after the first, with the error channel...

<int:gateway id="barxServerGateway"
				 service-interface="MyServerGateway"
				 default-request-channel="toErrorCatchingGw" />
<int:service-activator input-channel="toErrorCatchingGw" ref="errorCatchingGw" />
<int:gateway id="errorCatchingGw" error-channel="errorObject" default-request-channel="requestObject" />

This way, the error is handled on this intermediate gateway and the reply is returned normally to the inbound gateway.

[matt_koss]

Thanks, I now got further with this.
Errors seem to be routing as they should.

With this set up however, I am getting a number of errors during startup :

01-24-2013 17:27:43 - [DEBUG] [main] MessagingMethodInvokerHelper - Method [public final void $Proxy44.addAdvisor(int,org.springframework.aop.Advisor) throws org.springframework.aop.framework.AopConfigException] is not eligible for Message handling.
java.lang.IllegalArgumentException: Found more than one parameter type candidate: [int] and [org.springframework.aop.Advisor]

2 more such exceptions appears, related to the methods addAdvice and replaceAdvisor.

They don't seem to affect the set up afterwards, but I wonder what could be causing them...

[Gary Russell]

That message is benign - it's because the service activator code is trying to find the appropriate method to invoke on the gateway (which, with no service-interface is RequestReplyExchanger). It finds the method ok, but during the search, it hits a method that's not eligible for consideration and logs the condition under DEBUG.

We should probably change this log to TRACE level (rather then DEBUG), or at least suppress the full stack trace under DEBUG.

You can eliminate the message by configuring your log4j to set logs for org.springframework.integration.util.MessagingMeth odInvokerHelper to INFO, or WARN.

[matt_koss]

Yes, I have done that now, thanks.
I only spotted after my post that is was DEBUG level (got confused as the stack trace was was being logged too).

I very much appreciate your quick responses.

[Gary Russell]

No problem - I opened a JIRA ticket... https://jira.springsource.org/browse/INT-2900

[matt_koss]

BTW, in the error scenario above, with regard to the original request message that is in the aggregator... What happens to that ?
Error seems to come back to the caller, so does the request message expire inside the aggregator independently ?

[Gary Russell]

The aggregator releases both messages in the released message (with the payload being a List<?>). Your transformer (get(1)) transforms the List<?> to the second payload and the first is simply discarded.

This is true whether the second message is the response, or the ErrorMessage.

[matt_koss]

That part I understand. However, in case of the above error, as it is not routed back into the aggregator, how can it "release" the group ?
I can confirm, that during the debugging, in case of the connectivity error, the exception / ErrorMessage does not come into the aggregator and my release strategy does not get invoked :

	@ReleaseStrategy
	public boolean isReadyToRelease(List<Message<?>> messages)
	{
		return messages.size() == 2;
	}