Spring集成TCP服务器不接收消息

Question

我正在尝试创建一个 TCP 服务器，该服务器在端口 5002 上接受来自外部程序的消息。但是，它没有从外部程序接收消息。

@Bean
public TcpReceivingChannelAdapter inbound(AbstractServerConnectionFactory cf) {
   TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
   adapter.setConnectionFactory(cf);
   adapter.setOutputChannel(tcpIn());
   return adapter;
 }

@Bean
public MessageChannel tcpIn() {
    return new DirectChannel();
}

@Bean
@Transformer(inputChannel = "tcpIn", outputChannel = "serviceChannel")
public ObjectToStringTransformer transformer() {
    return new ObjectToStringTransformer();
}

@ServiceActivator(inputChannel = "serviceChannel")
public void messageToService(String in) {
    // Message received
}

@Bean
public AbstractServerConnectionFactory serverConnectionFactory() {
    TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(5002);
    tcpNetServerConnectionFactory.setSoTimeout(5000);
    tcpNetServerConnectionFactory.setMapper(new TimeoutMapper());
    return tcpNetServerConnectionFactory;
}

为了验证我的 TCP 服务器是否正常工作，我像这样使用 telnet，程序确实收到了文本“hello”。

telnet 192.168.1.2 5002
Trying 192.168.1.2...
Connected to 192.168.1.2.
Escape character is '^]'.
hello

设置wireshark我可以看到计算机正在5002端口接收来自外部程序（我期待的）的消息。为什么我的程序无法接收这些消息？

最终解决方案更新：

由于有效载荷没有停止线，我必须按照@Artem Bilan 的描述实现我自己的反序列化器。我使用“~”字符来表示客户端的行尾。

@Bean
public AbstractServerConnectionFactory serverConnectionFactory() {
    TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(tcpPort);
    tcpNetServerConnectionFactory.setSoTimeout(0);
  tcpNetServerConnectionFactory.setDeserializer(endOfLineSerializer());
    tcpNetServerConnectionFactory.setSerializer(endOfLineSerializer());
    tcpNetServerConnectionFactory.setMapper(new TimeoutMapper());
    return tcpNetServerConnectionFactory;
}

我实现的示例序列化程序：

public class EndOfLineSerializer extends AbstractPooledBufferByteArraySerializer {

private static final char MANUAL_STOP_LINE = '~';
private static final char AUTO_STOP_LINE = '\t';
private static final byte[] CRLF = "\r\n".getBytes();

/**
 * Reads the data in the inputStream to a byte[]. Data must be terminated
 * by a single byte. Throws a {@link SoftEndOfStreamException} if the stream
 * is closed immediately after the terminator (i.e. no data is in the process of
 * being read).
 */
@Override
protected byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
    int n = 0;
    int bite;

    try {
        while (true) {

            try {
                bite = inputStream.read();
            } catch (SocketTimeoutException e) {
                bite = -1;
            }

            if (bite < 0) {
                // Payload complete
                break;
            }

            if ((n > 0 && bite == '\n' && buffer[n - 1] == '\r') || bite == this.MANUAL_STOP_LINE || bite == this.AUTO_STOP_LINE) {
                break;
            }

            buffer[n++] = (byte) bite;
            if (n >= this.maxMessageSize) {
                throw new IOException("Terminator not found before max message length: " + this.maxMessageSize);
            }
        }
        return copyToSizedArray(buffer, n);
    } catch (IOException e) {
        publishEvent(e, buffer, n);
        throw e;
    } catch (RuntimeException e) {
        publishEvent(e, buffer, n);
        throw e;
    }
}

/**
 * Writes the byte[] to the stream and appends the CRLF.
 */
@Override
public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
    outputStream.write(bytes);
    outputStream.write(this.CRLF);
    }
}

Answer 1

TcpNetServerConnectionFactory 默认使用ByteArrayCrLfSerializer，这就是消息分隔符：

private static final byte[] CRLF = "\r\n".getBytes();

因此，您应该确保您的客户最后发送带有正确符号的消息。

有许多开箱即用的序列化程序供您选择：

https://docs.spring.io/spring-integration/docs/5.0.3.RELEASE/reference/html/ip.html#tcp-connection-factories

或者您可以实现自己的 Deserializer 并注入到 serverConnectionFactory bean 定义中。

Answer 2

见the documentation;向下滚动到

TCP 是一种流协议；这意味着必须为通过 TCP 传输的数据提供某种结构，以便接收方可以将数据划分为离散的消息。连接工厂配置为使用（反）序列化程序在消息有效负载和通过 TCP 发送的位之间进行转换。这是通过分别为入站和出站消息提供反序列化器和序列化器来实现的。提供了许多标准（反）序列化程序。

并阅读有关标准反序列化程序的信息。根据您的配置，标准解串器正在等待终止 \r\n (CRLF)。

Telnet 附加 CRLF，这就是它起作用的原因。