reply-channel for tcp-inbound-gateway

[Gary Russell]

I haven't fully dissected your flow but transformers have to return a result

2011-02-08 21:55:22,751 (pool-1-thread-2) [ProductTransformer.transform]
ERROR - Error converting json to Products object, No content to map to Object due to end of input
2011-02-08 21:55:22,753 (pool-1-thread-2) [ProductTransformer.transform]
ERROR -
No content to map to Object due to end of input

...

org.springframework.integration.MessageHandlingExc eption: handler 'org.springframework.integration.transformer.Messa geTransformingHandler@29a220e6' requires a reply, but no reply was received

...

That's why you are getting an 'requires reply' error and why we never send it to the second recipient. Again, running it on a separate thread will make that process independent of your first leg.

Also, you need to add a <header-filter/> to remove the replyChannel from the first leg so we don't try to send any reply created by that leg back to the tcp gateway.

In future, please attach logs in a zip; thanks.

[johnjianfang]

Sorry. I didn't know that I can attach a zip file here.

The wired thing is why the transformer received an empty message because the client only sent one message with 3 products. Should never be empty.

Thanks,

John

[johnjianfang]

Did you mean whether to reply back is controlled by header meta data? Will other behaviors be controlled by the header meta data as well?

I changed the router to a publish-subcribe-channel, but got back the same result.

<publish-subscribe-channel id="productPublishChannel"/>

<service-activator input-channel="productPublishChannel" output-channel="replyChannel"
ref="productValidator" method="validate"/>

<beans:bean id="productSplitter" class="com.barnesandnoble.message.bus.ProductSplit ter"/>

<service-activator input-channel="productPublishChannel" output-channel="productsChannel"
ref="productSplitter" method="split"/>

I saw publish-subscribe-channel comes with a task-executor attribute, but could not find an example for it.

<publish-subscribe-channel id="pubsubChannel" task-executor="someExecutor"/>

Is the task-executor a thread or a thread pool. If it is a thread pool, will the thread pool be used for the current/single tcp request thread or shared for all tcp request threads?

Also, does the server tcp connection factory create tcp request threads from a thread pool? If it does, how to configure the thread pool?

Thanks,

John

[Gary Russell]

Yes, when an inbound gateway (any gateway, not just TCP) puts a message on the request-channel, it includes a temporary reply channel in the replyChannel header. This channel is used to send a reply to the client; even if you explicitly send the reply to the reply-channel, that gets bridged to the temporary reply channel.

Since you are sending the message to two subflows, it only makes sense for one of those subflows to send a reply so you need to expunge the replyChannel header from one of the flows using a <header-filter/>.

The reason you are seeing the same result with a pub-sub with no task executor is again because the first leg is failing. You can set ignore-failures to true, or use a task-executor to run the reply on a separate thread, but in any case, you must remove the replyChannel header from the other flow.

You can choose whatever task executor you want; the easiest is to use the <task: /> namespace, e.g.

	<task:executor id="executor" pool-size="10"/>

With the tcp connnection factory, you can specify an external task executor or, if you don't, we use a fixed thread pool task executor

			if (this.taskExecutor == null) {
				this.taskExecutor = Executors.newFixedThreadPool(this.poolSize);
			}

Where the pool size defaults to 5.

This is all documented in Table 16.1. Connection Factory Attributes.

[johnjianfang]

Thanks for the reply. But that does not explain why the socket was read multiple times on the server side. The client side only sent request once.

011-02-09 12:57:39,021 (pool-1-thread-1) [TcpNetServerConnectionFactory.run]
DEBUG - Accepted connection from 127.0.0.1
2011-02-09 12:57:39,025 (pool-1-thread-2) [TcpNetConnection.run]
DEBUG - Reading...
2011-02-09 12:57:39,025 (pool-1-thread-2) [ByteArrayCrLfSerializer.deserialize]
DEBUG - Available to read:0
2011-02-09 12:57:39,030 (pool-1-thread-2) [TcpNetConnection.run]
DEBUG - Message received [Payload=[B@6f03fcaa][Headers={timestamp=1297274259030, id=5481206f-897c-4887-bf74-45dc3f2d4363, ip_address=127.0.0.1, ip_connection_seq=1, ip_hostname=localhost.localdomain, ip_tcp_remote_port=57415, ip_connection_id=localhost.localdomain:57415:17741 97673}]
...
2011-02-09 12:57:39,112 (pool-1-thread-2) [ByteArrayCrLfSerializer.deserialize]
DEBUG - Available to read:2
2011-02-09 12:57:39,112 (pool-1-thread-2) [TcpNetConnection.run]
DEBUG - Message received [Payload=[B@53458dcb][Headers={timestamp=1297274259112, id=10f502df-6ac4-445c-8cad-9ce8bd87a47a, ip_address=127.0.0.1, ip_connection_seq=2, ip_hostname=localhost.localdomain, ip_tcp_remote_port=57415, ip_connection_id=localhost.localdomain:57415:17741 97673}]
...
2011-02-09 12:57:39,131 (pool-1-thread-2) [ByteArrayCrLfSerializer.deserialize]
DEBUG - Available to read:0
2011-02-09 12:57:39,132 (pool-1-thread-2) [TcpNetConnection.run]
ERROR - Read exception localhost.localdomain:57415:1774197673 SocketException:null:Connection reset

Does that mean the server will keep reading data from the socket and trying to process it even the input is empty? Sound wired to me because I assume the server will expect CRLF to indicate the end of a message and it should not process any empty input. Assume there is no exception, would the server keep processing the empty input as long as the request socket is open?

Thanks,

John

[johnjianfang]

Another confusing part is what tasks the executor will run, all the subscribers of the publish and subscribe channel? For example, in your case, you defined a thread pool with pool size 10.

You can choose whatever task executor you want; the easiest is to use the <task: /> namespace, e.g.

	<task:executor id="executor" pool-size="10"/>

Does that mean all the subscribers will be run on this thread pool?

Thanks,

John

[Gary Russell]

...Does that mean the server will keep reading data from the socket and trying to process it even the input is empty?

It depends on whether using-nio is true or false. Please read the reference documentation; if it's not clear there, please let us know.

To summarize...

If not using NIO, we have to dedicate a thread to each socket and we block on the read until there is new data, when data arrives we keep reading until we detect the end of the message, the message is then put on the channel and we wait for a reply; then we go back and wait for (read) the next message. In this case, the thread is never put back into the pool (until the socket closes).

If using NIO, threads are shared between sockets and a thread is used to read/assemble a message until it is complete; it then passes it to another thread that sends the message and waits for the reply. The initial thread is put back into the pool as soon as the message has been assembled; the second thread is put back into the pool when the reply is received. Reader threads and sender threads are continually being grabbed and returned to the pool.

NIO provides an event driven architecture so we know when new data is available on a socket; without NIO, we have no choice but to block on a read.

[Gary Russell]

...Does that mean all the subscribers will be run on this thread pool?

Channels, by default, are DirectChannels; that means the message consumer is run on the same thread as the producer. If you do an aysnc handoff (perhaps by using a pub-sub channel with a task-executor, or a QueueChannel), then the thread used for the hand-off will come from whatever task executor you configure for that purpose. If you configure the same task executor to be used in different places (e.g. poller, pub-sub channel, tcp connection factory, etc, etc) then, yes, they will share the same pool. But, there is nothing stopping you from configuring a different executor for each place.

[johnjianfang]

In my previous test, I did not use the nio. Then, I turned the use-nio to true, the empty input string "" is still coming and thus led to errors.

It depends on whether using-nio is true or false. Please read the reference documentation; if it's not clear there, please let us know.

I have not objection to whether block the thread for the socket, but am curious about why the empty input got passed to the message channel.
For the tcp-connection-factory, I didn't specify the serializer and deserializer, then it should use CRLF ones, right?

	<ip:tcp-connection-factory id="crLfServer" 
		type="server"
        pool-size="40"
        using-nio="true"
		port="8888"/>

For the empty string, it certainly does not have CRLF and the server should just keep checking if there is CRLF marked message coming and keep waiting for that. I just could not understand why the empty string got passed in. Or do I misunderstand anything here?

Thanks,

John

[Gary Russell]

I think this is your problem...

DEBUG: org.springframework.integration.transformer.Messag eTransformingHandler - org.springframework.integration.transformer.Messag eTransformingHandler@2e8aeed0 received message: [Payload={"products":[{"type":"BOOKS","ean":"2000003602803"},{"type": "ME RCH","ean":"2000003602888"},{"type":"WORKS","ean" : "2000003602999"}]}
][Headers={timestamp=1297220122633, id=f3f3ffbc-3f2f-401d-8c58-9228c0bc1ccb, errorChannel=org.springframework.integration.core. MessagingTemplate$TemporaryReplyChannel@46e06703,

Notice the newline at the end of the payload (between your final } and the ][Headers...)

Looks to me like your payload already has a CRLF at the end. The serializer on the client side is adding another CRLF and we are reading that as a second message...

...
2011-02-09 12:57:39,112 (pool-1-thread-2) [ByteArrayCrLfSerializer.deserialize]
DEBUG - Available to read:2
2011-02-09 12:57:39,112 (pool-1-thread-2) [TcpNetConnection.run]
DEBUG - Message received [Payload=[B@53458dcb][Headers={timestamp=1297274259112, id=10f502df-6ac4-445c-8cad-9ce8bd87a47a, ip_address=127.0.0.1, ip_connection_seq=2, ip_hostname=localhost.localdomain, ip_tcp_remote_port=57415, ip_connection_id=localhost.localdomain:57415:17741 97673}]
...

You need to strip off the CRLF from your payload or write a custom serializer that does not add CRLF on the client side (I just checked in a ByteArrayRawSerializer which would work) - you need to retain the existing deserializer on the server side.