TcpOutboundGateway below high concurrency request not match response

[slmouse]

In what version(s) of Spring Integration are you seeing this issue?

5.4.13

Describe the bug

A clear and concise description of what the bug is.
below high concurrency TcpOutboundGateway class have Cannot correlate response - no pending reply for exception
It seems other thread removed AsyncReply from pendingReplies with connectionId,because the all thread use the same connectionId, so connectionId maybe not used as a unique identifier in pendingReplies sence.

I tried to change uuid instance connectionId put pendingReplies ,but when put uuid to header in method handleRequestMessage,
the header in method onMessage nonexistent uuid,

To Reproduce

high concurrency call

Expected behavior

below high concurrency request not match response

[garyrussell]

This situation is well documented here: https://docs.spring.io/spring-integration/docs/current/reference/html/ip.html#ip-gateways

TCP has no standard way to provide request/reply correlation in the TCP stream so each socket can only be used for one request at-a-time.

There are several solutions, including:

• use a new connection for each request
• use the CachingClientConnectionFactory which maintains a pool of connections which are reused when available
• use collaborating channel adapters and provide your own correlation mechanism

[slmouse]

This situation is well documented here: https://docs.spring.io/spring-integration/docs/current/reference/html/ip.html#ip-gateways

TCP has no standard way to provide request/reply correlation in the TCP stream so each socket can only be used for one request at-a-time.

There are several solutions, including:

• use a new connection for each request
• use the CachingClientConnectionFactory which maintains a pool of connections which are reused when available
• use collaborating channel adapters and provide your own correlation mechanism

Hi
The tcp server(other company) does not support too many links and frequent creation new connection ,and not support add something with header,so I only use one connection
Can give me more advice, tks

[garyrussell]

There are no other alternatives.

The CachingClientConnectionFactory allows you to limit the number of cached connections, so it doesn’t have to be “too many” connections, just enough to support your workload.

[slmouse]

In my scene payload contain unique identifier.
may i extends TcpOutboundGateway get unique identifier from payload replace connectionId put/get from pendingReplies
in method handleRequestMessage pendingReplies.put(connectionId->unique identifier , reply);
in method onMessage AsyncReply reply = this.pendingReplies.get(connectionId->unique identifier);
Do you think this plan is OK?

[garyrussell]

It should, but the gateway is not designed to be extended that way; you would probably have to make a copy and add your modifications.

Or, you can just use collaborating channel adapters (see the documentation I pointed you to), and store the pending replies in your own component.

That would avoid copying any code.

[artembilan]

See this sample for more info: https://github.com/spring-projects/spring-integration-samples/tree/main/intermediate/tcp-client-server-multiplex

[slmouse]

I add IntegrationFlow The number of inconsistencies falls,But there will be a small number of them
code below ,the code should be the same follow artembilan example

  public IntegrationFlow integrationFlow() {
        return IntegrationFlows.from(DataSenderApplication.BYTES_CHANNEL)
                .enrichHeaders(headerEnrichedSpec -> headerEnrichedSpec.header("correlationId", UUID.randomUUID()))
                .handle(tcpMessageHandler)
                .aggregate(aggregatorSpec -> aggregatorSpec
                        .correlationStrategy(message -> message.getHeaders().get("correlationId"))
                        .releaseStrategy(messageGroup -> messageGroup.size() >= 1)
                        .expireGroupsUponCompletion(true)
                        .sendPartialResultOnExpiry(true)
                )
                .channel(replyChannel())
                .get();
    }

[artembilan]

This one cannot work:

                .handle(tcpMessageHandler)
                .aggregate(aggregatorSpec -> aggregatorSpec

The TcpSendingMessageHandler is one-way: it does not produce a reply to be sent to the next endpoint in the flow.

You are missing a publishSubscribeChannel() for those two: https://docs.spring.io/spring-integration/docs/current/reference/html/core.html#channel-implementations-publishsubscribechannel

I'm moving this one to Discussions, since it does not reflect any problems in the project.

[slmouse]

tcpMessageHandler in this is MessageHandler(TcpOutboundGateway)
You mean I need to add publishSubscribeChannel between from(DataSenderApplication.BYTES_CHANNEL) and replyChannel?