TcpOutboundGateway : Cannot correlate response - no pending reply for Cached

Question

I use spring integration for connect to tcp/ip socket server, I created mock server based on the telnet-mock https://github.com/maltempi/telnet-mock . and I can send and received message, but when I shut down mock server, in main application a cyclic error occurs and takes up all the CPU time:

ERROR 13942 --- [pool-4-thread-1] o.s.i.ip.tcp.TcpOutboundGateway : Cannot correlate response - no pending reply for Cached:localhost:3002:46550:f6234e17-c486-4506-82c8-a757a08ba73d.

How can I resolve this problem? My config class:

@EnableIntegration
@RequiredArgsConstructor
@Configuration
public class StpClientConfiguration {

private static final String REQUEST_CHANNEL = "toStp";
private static final String OUTPUT_CHANNEL = "resultToMap";
private static final String CRLF = "\\0";

private final ApplicationProperties applicationProperties;
private final ApplicationContext context;

private static String readUntil(InputStream inputStream, String stopWord) throws IOException {
    StringBuilder sb = new StringBuilder();
    BufferedReader buffer = new BufferedReader(new InputStreamReader(inputStream));
    int r;
    while ((r = buffer.read()) != -1) {
        char c = (char) r;
        sb.append(c);
        if (sb.toString().endsWith(stopWord)) {
            break;
        }
    }
    return sb.toString();
}

@Bean
public CachingClientConnectionFactory connectionFactory() {
    TcpNetClientConnectionFactory factory = new TcpNetClientConnectionFactory(
            applicationProperties.getHost(), applicationProperties.getPort());

    factory.setApplicationEventPublisher(this.context);
    factory.setTcpSocketSupport(new DefaultTcpSocketSupport());
    factory.setDeserializer((InputStream inputStream) -> readUntil(inputStream, CRLF));

    return new CachingClientConnectionFactory(factory, applicationProperties.getPoolSize());
}

/**
 * Creates the tcp gateway for service activation.
 *
 * @return the message handler
 */
@Bean
@ServiceActivator(inputChannel = REQUEST_CHANNEL)
public MessageHandler outboundGateway() {
    TcpOutboundGateway gateway = new TcpOutboundGateway();
    gateway.setConnectionFactory(connectionFactory());
    gateway.setOutputChannelName(OUTPUT_CHANNEL);

    return gateway;
}

@MessagingGateway(defaultRequestChannel = REQUEST_CHANNEL)
public interface RequestGateway {

    Map<String, String> send(String message);
}

@Bean
@Transformer(inputChannel = OUTPUT_CHANNEL)
public ObjectToMapTransformer objectToMapTransformer() {
    return new ObjectToMapTransformer();
}

}

Answer 1

Your deserializer looks suspicious; telnet messages are terminated with \r\n, not \\0.

Use the default deserializer for telnet (the default is a ByteArrayCrLfSerializer).

When the deserializer detects a normal end of stream (-1), between messages, it must throw a SoftEndOfStreamException to tell the framework the socket is closed. Your code keeps returning a zero length String,

/**
 * Used to communicate that a stream has closed, but between logical
 * messages.
 */
public class SoftEndOfStreamException extends IOException {

Or, as I said, use the default deserializer.