service-activator not getting message payload?

[thisisCP]

I'm coding an application in which I need to send a request over a TCP socket and get a response back from a server asynchronously. With my current implimentation, I'm able to send the request. On the server side, I can see from the logs that request is received and a response is even sent back without any errors. But when I try to just dump the response to the console, nothing happens.

Just not sure what I'm missing here...

Here's my config:

<context:property-placeholder
		location="classpath:config.properties"/>

	<!-- CLIENT SIDE -->
	
	<bean id="rawSerializer" class="org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer" />
	<bean id="requestUtils" class="com.util.RequestUtils"/>
	
	<int:channel 
		id="initialRequest"
		datatype="java.lang.String"/>
		
	<int:channel 
		id="txAdded"
		datatype="java.lang.String"/>
		
	<int:channel 
		id="finalRequest"
		datatype="java.lang.String"/>	
		
	<int:gateway
		id="requestHandler"
		service-interface="com.util.RequestHandler"
		default-request-channel="initialRequest"/>
		
	<int:transformer
		id="addTXcontrol"
		input-channel="initialRequest"
		output-channel="txAdded"
		ref="isRequestUtils"
		method="addTX"/>
		
	<int:transformer
		id="addPrefix"
		input-channel="txAdded"
		output-channel="finalRequest"
		ref="requestUtils"
		method="addPrefix"/>
	
 	<int-ip:tcp-outbound-channel-adapter
		id="message"
		channel="finalRequest"
		connection-factory="requestCF"/>
		
	<int-ip:tcp-connection-factory
		id="requestCF"
		type="client" 
		host="${server.address}" 
		port="${server.request.port}"
		so-keep-alive="true"
		single-use="false"
		serializer="rawSerializer"/>


	<!-- SERVER SIDE -->
	
	<bean id="byteArrayStxEtxSerializer" class="org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer" />
	<bean id="responseUtils" class="com.util.ResponseUtils"/>
	
	<int:channel
		id="serverResponse"
		datatype="java.lang.String"/>
		
	<int-ip:tcp-connection-factory
		id="serverListener"
		type="server"
		host="localhost"
		port="${server.response.port}"
		so-keep-alive="true"
		single-use="false"
		deserializer="byteArrayStxEtxSerializer"/>
		
	<int-ip:tcp-inbound-channel-adapter
		id="inboundResponse"
		channel="serverResponse"
		connection-factory="serverListener"/>
		
	<int:service-activator
		input-channel="serverResponse"
		ref="responseUtils"
		method="showResponse"/>

My main:

public static void main(String[] args) {
		ApplicationContext context = new ClassPathXmlApplicationContext("/spring-config/connection-config.xml");
		
		final RequestHandler requestHandler = context.getBean(RequestHandler.class);
		final AbstractServerConnectionFactory serverListener = context.getBean(AbstractServerConnectionFactory.class);
		
		requestHandler.send("Test message");
	}

ResponseUtils:

public class ResponseUtils {

	public String showResponse(String payload) {
		System.out.println(payload);
		return payload;
	}
	
}

[Gary Russell]

This configuration is incomplete - you have no output channel on the responseUtils service activator and you are using an inbound-channel-adapter. I would expect this to throw an exception.

In any case, I see nothing on the client side to receive the response.

You either need to use gateways instead of adapters (for request/response) or use collaborating inbound and outbound channel adapters on both client and server. This is discussed in the reference manual and the tcp-client-server-multiplex sample.

It's always a good idea to turn on DEBUG logging for org.springframework.integration so you can see messages flowing between components.

[thisisCP]

I decided to use adapters because the request/response will be decoupled. We only need to send requests to the remote server. There will be no acknowledgement the request was received. The response will actually be persisted to a database table and then some other process will look up the response in the database. We're doing this in case the port that receives response from the remote server goes down (but not the connection that sends requests). That server will queue up all the responses until the connection is re-established. At that point, all the queued up responses will be sent.Eventually I'd add the persist logic in. But initially, I just want to confirm I get the response.

[Gary Russell]

Right, that makes sense (a unidirectional flow), but you have been talking about requests and responses.

But, right now, your showResponse returns a value, and there's no where to send it.

I am confused by your comment

response is even sent back without any errors

Sent back where?

Perhaps if you can post a DEBUG log from the client and the server, and explain what you mean by this quote, I can understand better what you are saying?

[thisisCP]

Sorry, I'm probably not using the correct terminology.By request, I meant a message sent to the initialRequest channel by

requestHandler.send("Test message");

and eventually out to the remote server through the tcp-outbound-channel-adapter.By response, I meant a message received on the serverResponse channel through the tcp-inbound-channel-adapter.For example, say the request message contains a person_id and the response message would contain the person's demographic info.So if I run the code as is on my localhost and do:

requestHandler.send("1234");

I can see on the remote server's incoming request logs that it received a person_id:

Connected from 10.127.211.92:9300 to 10.10.140.76:1197rcv:^DCTXmsg:rcv:                      -10000000701 9305^C^BSTXmsg:rcv:1234^CProcessingrcvd app,989449,03:15:08 PM

If I then check the server log for outgoing responses, I see it sends back a response message:

Connected from 10.127.211.92:2237 to 10.10.140.76:9305APP:989449 :start app,989449,03:15:10 PMsnd:^BJAMES    T    KIRK       USS ENTERPRISE ^Cfinished app,989449,03:15:10 PM

So, I'm just trying to get a hold of that response message that contains the person's demo info.

[thisisCP]

BTW, that last post was formatted, but for some reason, the Enhanced Interface - Full WYSIWYG Editing option isn't taking, so all I end up with is a plain text box with no formatting

[Gary Russell]

Maybe I am misunderstanding your app - I thought your "CLIENT SIDE" config was talking to your "SERVER SIDE" config.


It sounds like this is talking to some other server?


If that's the case, and the responses come back on the same socket, you need to just use the clent connection factory and inject it into both the inbound and outbound adapter.

However, if the server is connecting back to the client on a different socket, you would use two connection factories - but you should see all the connection activity stuff in the logs (under debug).

[thisisCP]

Yes, the client side and server side are not taking to each other, but a remote server. I'm running the application on my localhost, and have it configured such that the client side sends messages to port 9300 of the remote server. When the remote server sends a message back, it's configured to send responses back to port 9305 of my localhost. Checking the DEBUG logs for spring-integration, it looks like the application is getting the message, but it's just too long. I read in the documentation I need to set the maxMessageSize property, but I'm not sure where to do that. Also, is there a max maxMessageSize?

INFO  DefaultConfiguringBeanFactoryPostProcessor:120 No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.INFO  DefaultConfiguringBeanFactoryPostProcessor:146 No bean named 'taskScheduler' has been explicitly defined. Therefore, a default ThreadPoolTaskScheduler will be created.INFO  GatewayProxyFactoryBean$MethodInvocationGateway:87 started requestHandlerINFO  GatewayProxyFactoryBean:87 started requestHandlerDEBUG ServiceActivatingHandler:134 Unable to attempt conversion of Message payload types. Component 'null' has no explicit ConversionService reference, and there is no 'integrationConversionService' bean within the context.INFO  EventDrivenConsumer:83 Adding {transformer:addTXcontrol} as a subscriber to the 'initialRequest' channelINFO  DirectChannel:69 Channel 'initialRequest' has 1 subscriber(s).INFO  EventDrivenConsumer:87 started addTXcontrolINFO  EventDrivenConsumer:83 Adding {transformer:addPrefix} as a subscriber to the 'txAdded' channelINFO  DirectChannel:69 Channel 'txAdded' has 1 subscriber(s).INFO  EventDrivenConsumer:87 started addPrefixINFO  TcpNetClientConnectionFactory:425 started requestCFINFO  TcpNetClientConnectionFactory:425 started requestCFINFO  EventDrivenConsumer:83 Adding {ip:tcp-outbound-channel-adapter:message} as a subscriber to the 'finalRequest' channelINFO  DirectChannel:69 Channel 'finalRequest' has 1 subscriber(s).INFO  EventDrivenConsumer:87 started messageINFO  EventDrivenConsumer:83 Adding {service-activator} as a subscriber to the 'isResponse' channelINFO  DirectChannel:69 Channel 'isResponse' has 1 subscriber(s).INFO  EventDrivenConsumer:87 started org.springframework.integration.config.ConsumerEndpointFactoryBean#0INFO  EventDrivenConsumer:83 Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channelINFO  PublishSubscribeChannel:69 Channel 'errorChannel' has 1 subscriber(s).INFO  EventDrivenConsumer:87 started _org.springframework.integration.errorLoggerINFO  TcpNetServerConnectionFactory:425 started isListenerINFO  TcpNetServerConnectionFactory:425 started isListenerINFO  TcpReceivingChannelAdapter:87 started inboundResponseINFO  TcpNetServerConnectionFactory:76 Listening on port 9305DEBUG DirectChannel:224 preSend on channel 'initialRequest', message: [Payload=1234][Headers={timestamp=1359722177269, id=c1bf41c5-2ec6-45cb-9a6a-a034c4f581ec, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG MessageTransformingHandler:67 org.springframework.integration.transformer.MessageTransformingHandler@aeea66 received message: [Payload=1234][Headers={timestamp=1359722177269, id=c1bf41c5-2ec6-45cb-9a6a-a034c4f581ec, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG MessageTransformingHandler:197 handler 'org.springframework.integration.transformer.MessageTransformingHandler@aeea66' sending reply Message: [Payload=1234][Headers={timestamp=1359722177301, id=f548a488-48bd-4981-b9be-f3062f897b55, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG DirectChannel:224 preSend on channel 'txAdded', message: [Payload=1234][Headers={timestamp=1359722177301, id=f548a488-48bd-4981-b9be-f3062f897b55, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG MessageTransformingHandler:67 org.springframework.integration.transformer.MessageTransformingHandler@c0f1ec received message: [Payload=1234][Headers={timestamp=1359722177301, id=f548a488-48bd-4981-b9be-f3062f897b55, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG MessageTransformingHandler:197 handler 'org.springframework.integration.transformer.MessageTransformingHandler@c0f1ec' sending reply Message: [Payload=                      -10000000701 93051234][Headers={timestamp=1359722177301, id=43cf2b26-f7d8-44f4-b357-c4ace3953e2c, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG DirectChannel:224 preSend on channel 'finalRequest', message: [Payload=                      -10000000701 93051234][Headers={timestamp=1359722177301, id=43cf2b26-f7d8-44f4-b357-c4ace3953e2c, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG TcpSendingMessageHandler:67 org.springframework.integration.ip.tcp.TcpSendingMessageHandler#0 received message: [Payload=                      -10000000701 93051234][Headers={timestamp=1359722177301, id=43cf2b26-f7d8-44f4-b357-c4ace3953e2c, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG TcpNetClientConnectionFactory:61 Opening new socket connection to 10.127.211.92:9300DEBUG TcpSendingMessageHandler:147 Got Connection VMCAP2ISDEVMED:9300:cbf4c3a9-4ef3-4367-a336-288927ccf7e3DEBUG TcpNetConnection:96 TcpListener exiting - no listener and not single useDEBUG TcpNetConnection:94 Message sent [Payload=                      -10000000701 93051234][Headers={timestamp=1359722177301, id=43cf2b26-f7d8-44f4-b357-c4ace3953e2c, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG DirectChannel:237 postSend (sent=true) on channel 'finalRequest', message: [Payload=                      -10000000701 93051234][Headers={timestamp=1359722177301, id=43cf2b26-f7d8-44f4-b357-c4ace3953e2c, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG DirectChannel:237 postSend (sent=true) on channel 'txAdded', message: [Payload=1234][Headers={timestamp=1359722177301, id=f548a488-48bd-4981-b9be-f3062f897b55, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG DirectChannel:237 postSend (sent=true) on channel 'initialRequest', message: [Payload=1234][Headers={timestamp=1359722177269, id=c1bf41c5-2ec6-45cb-9a6a-a034c4f581ec, errorChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8, replyChannel=org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel@1b0bdc8}]DEBUG TcpNetServerConnectionFactory:100 Accepted connection from 10.127.211.92DEBUG TcpNetConnection:101 Reading...DEBUG TcpNetConnection:118 Read exception VMCAP2ISDEVMED:3138:b6f850f8-d597-4837-a02d-6adc6551b038 IOException:null:ETX not found before max message length: 2048DEBUG TcpNetServerConnectionFactory:100 Accepted connection from 10.127.211.92DEBUG TcpNetConnection:101 Reading...

[thisisCP]

Actually, I got the application to the point where it get the response message now (as seen in the DEBUG log) and where it also prints it to the console.I changed the config to this:

But I get this message at the end:

ERROR TcpNetConnection:155 Exception sending meeeage: [Payload=[B@1724a9d][Headers={timestamp=1359726232483, id=50d71c3f-c71c-493d-89f2-d38c52b0ac82, ip_tcp_remotePort=3247, ip_address=10.127.211.92, ip_hostname=VMCAP2ISDEVMED, ip_connectionId=VMCAP2ISDEVMED:3247:30784d75-5afd-4e28-878e-87b72324d25a}]org.springframework.integration.MessageDeliveryException: Dispatcher has no subscribers for channel toPersist.	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:82)	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128)	at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:288)	at org.springframework.integration.core.MessagingTemplate.send(MessagingTemplate.java:149)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendMessage(AbstractReplyProducingMessageHandler.java:216)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendReplyMessage(AbstractReplyProducingMessageHandler.java:200)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.produceReply(AbstractReplyProducingMessageHandler.java:165)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleResult(AbstractReplyProducingMessageHandler.java:159)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:141)	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:115)	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:102)	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128)	at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:288)	at org.springframework.integration.core.MessagingTemplate.send(MessagingTemplate.java:149)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendMessage(AbstractReplyProducingMessageHandler.java:216)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.sendReplyMessage(AbstractReplyProducingMessageHandler.java:200)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.produceReply(AbstractReplyProducingMessageHandler.java:165)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleResult(AbstractReplyProducingMessageHandler.java:159)	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:141)	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:115)	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:102)	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128)	at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:288)	at org.springframework.integration.core.MessagingTemplate.send(MessagingTemplate.java:149)	at org.springframework.integration.endpoint.MessageProducerSupport.sendMessage(MessageProducerSupport.java:92)	at org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter.onMessage(TcpReceivingChannelAdapter.java:74)	at org.springframework.integration.ip.tcp.connection.TcpNetConnection.run(TcpNetConnection.java:144)	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(Unknown Source)	at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)	at java.lang.Thread.run(Unknown Source)Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:109)	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:102)	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)	... 36 more

How do I specify there's nothing left to do after I get the response message?

[Gary Russell]

<bean id="byteArrayStxEtxSerializer" class="org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializer">
	<property name="maxMessageSize" value="10000"/>
</bean>

The default is 2048.

This is there to protect the application from OOM conditions - if the other side keeps sending data without the appropriate delimiter.

However, this should be been logged as an ERROR - you should not have had to log at DEBUG to detect this condition; I have opened a JIRA issue: https://jira.springsource.org/browse/INT-2911.