TCP outbound gateway connection pool issue

[sameera]

Hi,

I'm experiencing an issue with the tcp outbound gateway. This is used to integrate my application with a third party application service running on the same server. The third party respond with an xml for my xml request. Below is the gateway configuration.

	

         <int-ip:tcp-connection-factory id="connectionFactory"
		deserializer="deserializer" host="${socket.host}" port="${socket.port}"
		pool-size="100" single-use="false"
		so-keep-alive="true" type="client"
		so-timeout="20000" />

	<int-ip:tcp-outbound-gateway id="gateway"
		request-channel="sendStringMessageChannel" reply-channel="recieveStringMessageChannel"
		connection-factory="connectionFactory" request-timeout="10000"
		reply-timeout="10000" />

The deserializer code executed in the reply is as follows. It checks the end tag of the xml response to close the stream and return the response to further processing.

public String deserialize(InputStream inputStream) throws IOException {

        BufferedReader br = null;
        StringBuilder sb = null;
        String str = null;

        try {

            br = new BufferedReader(new InputStreamReader(inputStream));
            sb = new StringBuilder(br.readLine());

            while (br.ready()) {

                str = br.readLine();
                sb.append(str);
                if (str != null && str.contains("</msg>")) {

                    break;
                }
            }

            return sb.toString();

        } catch (Exception e) {

            throw new RuntimeException(e);

        } finally {
            
            if (br != null) {
                br.close();
            }
        }
    }

The issue is when put under load and when the limit of the pool is reached it begins to throws following errors.

[ERROR] Failed to send or receive
org.springframework.integration.MessagingException: Failed to send or receive
at org.springframework.integration.ip.tcp.TcpOutboundGateway.handleRequestMessage(TcpOutboundGateway.java:120)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:97)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)

2012-10-19 14:20:55.989 TcpOutboundGateway [ERROR] Tcp Gateway exception
java.net.SocketException: Socket closed
         at java.net.SocketOutputStream.socketWrite(SocketOutputStream.java:116)
         at java.net.SocketOutputStream.write(SocketOutputStream.java:132)

When I check the threads using VisualVM I see this number of threads were created but those never get killed.

I'm really stuck at this point being unable to tweak it any further to fix the issue. Is there a flow in the configuration? or has any important setting missing or contradictory setting is used?

SI version is 2.1.0.

Thanks in advance for any help or insight into issues related to this.

Sameera

[Gary Russell]

You should not be closing the input stream in the deserializer.

If you only want to use a socket once for each request, set single-use to "true".

That said, we should be cleaning up the threads in this case; the pool-size is not really relevant for a client factory with single-use="false".

I'll open a JIRA ticket.

[sameera]

Thanks Gary for quick response. Its late in the night here. I'll update the deserializer code and run the test again. Hope you'll attach the Jira link here.

Thanks
Sameera

[Gary Russell]

I have run lots of tests with the deserializer improperly closing the connection and everything gets cleaned up properly; so I won't open a JIRA without more evidence that there is a problem on our end.

By the way, 2.1.0 is very old. 2.1.4 is the latest release.

[sameera]

I did try the suggestions and the upgrade as well. Still the issue seems not solved. I will put more effort to isolate the issue and report back. I had to move on - implemented a class and registered as a service activator to open a socket connection and close after processing - as we had to meet a deadline.

I'll update this thread if I find a issue with the gateway.

[yogya]

Hi,

Regarding the previous thread since the outbound gateway should only be used for relatively low-volume use thought of including outbound and inbound channel adapters.But I am getting the exception

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

Do I have to use header-enricher?

looking forward for your reply on this.

Thanks in advance

[Gary Russell]

The intermediate TCP sample shows one technique for retaining/reinstating the replyChannel for a message received from a channel adapter...

https://github.com/SpringSource/spri...rver-multiplex

[yogya]

Thanks a lot.I'll try with this.

[yogya]

Hi,

I tried with the example https://github.com/SpringSource/spri...rver-multiplex.

But was unable to get the response. I am also not clear on the correlation logic used in that example. Could you please explain to me on what this means?

correlation-strategy-expression="payload.substring(0,3)"
  
release-strategy-expression="size() == 2" />

<transformer input-channel="toTransformer.client"
expression="payload.get(1)"/>

I tried the same thing without using the aggregator. Since it gave me the org.springframework.integration.support.channel.Ch annelResolutionException: no output-channel or replyChannel header available exception, I used header-enricher. Here is my config file.

        <int:channel id="sendMessageChannel" />
	<int:channel id="sendXMLMessageChannel" />
	<int:channel id="sendStringMessageChannel" />
	<int:channel id="recieveStringMessageChannelIn" />

		
	<int:gateway id="jposGateway"
		service-interface="com.gateway.infrastructure.JPOSGateway">
		<int:method name="sendMessage" request-channel="sendMessageChannel" />
	</int:gateway>
	
	
	<int-xml:marshalling-transformer
		marshaller="marshaller" input-channel="sendMessageChannel"
		output-channel="sendXMLMessageChannel" result-type="StringResult" />

	<int:object-to-string-transformer id="transformerBytes2String"
		input-channel="sendXMLMessageChannel" output-channel="sendStringMessageChannel" />

	<int-ip:tcp-connection-factory id="client"
		deserializer="deserializer" type="client" host="${socket.host}"
		port="${socket.port}" single-use="false" so-timeout="20000"/>
		
		<int-ip:tcp-outbound-channel-adapter id="outAdapter.client"
		channel="sendStringMessageChannel"
		connection-factory="client" /> 

	<int-ip:tcp-inbound-channel-adapter id="inAdapter.client"
		channel="recieveStringMessageChannelIn"
		connection-factory="client"/>

	
	<bean id="deserializer" class="com.infrastructure.jpos.SocketInputStreamDeserializer" />
		
	<int-xml:unmarshalling-transformer
		unmarshaller="marshaller" input-channel="recieveStringMessageChannel" />
		
	<bean id="receive" class="org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel">
		<constructor-arg type="long" value="1000"/>
		</bean>

		
    <int:header-enricher input-channel="recieveStringMessageChannelIn"   outputchannel="recieveStringMessageChannel">
	<int:error-channel ref="receive"/>
     	<int:reply-channel ref="receive"/>
    </int:header-enricher>

This is my gateway method.

Isomsg sendMessage(Isomsg msg,
   @Header("Reference") String referenceId,
   @Header("requestType") String requestType);

But I was still unable to get a response.Could you please guide me on this?

[Gary Russell]

When you use one-way adapters (rather than a gateway) you lose the replyChannel header (it's a live object on which your gateway is waiting for a reply).

The aggregator in the sample shows one technique of retaining the header - send the original message to the aggregator (contains the header); send the reply as well; the aggregated message contains a list with both the request and the reply (and the header), In the sample, we drop the original request with the transformer.

In order for this to work, the aggregator needs some correlation data. With tcp, this has to be carried in the payload somewhere.

If you use an outbound-gateway instead, you don't need to worry about this because the framework can do the correlation for you.