TCP客户端向外部服务器发送消息，不需要回复消息答案

【问题标题】：TCP client send Message to external server, reply message is not expectedTCP客户端向外部服务器发送消息，不需要回复消息
【发布时间】：2019-02-14 21:36:39
【问题描述】：

我正在通过 Spring Integration 设置 TCP 客户端，发送带有字符串的消息作为有效负载，不期望返回。也许序列化器/反序列化器工作不正常？抱歉，我正在学习 Spring 集成。

我可以通过 oepnssl 连接到外部 TCP 服务器：

---
# DC API Test System: microstrategy
sessions.list.
.
response
,status_code,1
,status_message,Unrecognised operation
,time,2019-02-15 07:08:08 (+1000)
.

我需要发送的命令是“sessions.list\n.\n”。

现在我构建了一个 tcp 客户端尝试连接到服务器：

spring-context.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:int="http://www.springframework.org/schema/integration"
    xmlns:int-ip="http://www.springframework.org/schema/integration/ip"
    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
        http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip-5.1.xsd
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-5.1.xsd">

<bean id="integrationConversionService"
          class="org.springframework.context.support.ConversionServiceFactoryBean">
        <property name="converters">
            <list>
                <bean class="com.microstrategy.example.ByteArrayToStringConverter"/>
            </list>
        </property>
</bean> 


<bean id="customeSerilizerDeserlizer" class="com.microstrategy.example.CustomSerializerDeserializer" />

<int:gateway service-interface="com.microstrategy.example.SimpleGateway"
    default-request-channel="output"
    default-reply-channel="reply"/>

<int:channel id="output"/>
<int:channel id="reply" datatype="java.lang.String"/>

<int-ip:tcp-connection-factory
    id="clientFactory"
    type="client"
    host="server"
    port="15099"
    serializer="customeSerilizerDeserlizer"
    single-use="true"
    so-timeout="10000"/>

<int-ip:tcp-outbound-gateway 
    request-channel="output"
    reply-channel="reply"
    connection-factory="clientFactory"
    request-timeout="10000"
    reply-timeout="10000"/>

</beans>

所以在this repo 之后，字符串应该转换为 byte[]。

我使用与 repo 完全相同的转换器，所以我只是复制这里以节省您的时间：

import java.io.UnsupportedEncodingException;

import org.springframework.core.convert.converter.Converter;

public class ByteArrayToStringConverter implements Converter<byte[], String> {

    private String charSet = "UTF-8";

    public String convert(byte[] bytes) {
        try {
            return new String(bytes, this.charSet);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            System.out.println("caught excepton in converter");
            return new String(bytes);
        }
    }

    /**
     * @return the charSet
     */
    public String getCharSet() {
        return charSet;
    }

    /**
     * @param charSet the charSet to set
     */
    public void setCharSet(String charSet) {
        this.charSet = charSet;
    }

}

public interface SimpleGateway {
    public String send(Message message);
}

我做了一个自定义序列化器：

package com.microstrategy.example;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.springframework.integration.ip.tcp.serializer.AbstractByteArraySerializer;

public class CustomSerializerDeserializer extends AbstractByteArraySerializer {
    @Override
    public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
        outputStream.write(bytes); 
    }

    @Override
    public byte[] deserialize(InputStream inputStream) throws IOException {
        // TODO Auto-generated method stub
        return null;
    }

}

我的主要功能：

Message<String> message = MessageBuilder.withPayload("sessions.list").build();
String replyMessage = simpleGateway.send(message);
Message<String> message2 = MessageBuilder.withPayload(".").build();
String replyMessage2 = simpleGateway.send(message2);
System.out.println(replyMessage2);

replyMessage 是

# DC API Test System: microstrategy

我似乎通过发送消息成功连接到服务器，但该消息未被服务器正确识别。任何有用的建议将不胜感激，谢谢！

更新 1：

我将输出添加到序列化器：

public class CustomSerializerDeserializer extends AbstractByteArraySerializer {
    @Override
    public void serialize(byte[] bytes, OutputStream outputStream) throws IOException {
        System.out.println("inside serialize");
        System.out.println(System.currentTimeMillis());


        String string = new String(bytes);
        System.out.println("byte[] in serialize is " + string);
        outputStream.write(bytes); 
    }

    @Override
    public byte[] deserialize(InputStream inputStream) throws IOException {
        // TODO Auto-generated method stub
        System.out.println("inside deserialize");
        System.out.println(System.currentTimeMillis());

        return null;
    }

}

inside serialize
1550182834431
byte[] in serialize is sessions.list
.

# DC API Test System: microstrategy
2019-02-14 17:21:35.185  INFO 91620 --- [       Thread-1] o.s.i.endpoint.EventDrive

输出显示 byte[] 似乎是正确的，那么为什么服务器没有按预期返回？

更新2：我更改了main函数（已更新），因为框架会在每条消息的末尾添加“\n”。对吗？

输出是

inside serialize
1550184564485
byte[] in serialize is sessions.list
inside serialize
1550184565003
byte[] in serialize is .
2019-02-14 17:49:35.013 ERROR 91740 --- [           main] o.s.i.ip.tcp.TcpOutboundGateway          : Tcp Gateway exception

org.springframework.integration.MessageTimeoutException: Timed out waiting for response

没有回应？

更新 3：我可以通过发送空消息来打开连接。但为什么其他消息不起作用？

Message<String> message = MessageBuilder.withPayload("").build();
String replyMessage = simpleGateway.send(message);
System.out.println(replyMessage);

有什么帮助吗？

这是我在解决问题之前的更新，已被管理员删除：

我现在收到了来自服务器的响应，但它带有错误：

Cannot correlate response - no pending reply for server:15099:49469:0fdce5c4-432f-4ce4-b878-2e08d0e96419
inside serialize
1550189909340
byte[] in serialize is sessions.list
.

GenericMessage [payload=byte[35], headers={ip_tcp_remotePort=15099, ip_connectionId=server:15099:49550:a3bc44fa-7d36-483c-a1b8-f91eea62d839, ip_localInetAddress=/10.21.66.115, ip_address=217.78.6.17, id=3a6ff696-f12f-6328-da1a-5d613d37a4b2, ip_hostname=server, timestamp=1550189909764}]
2019-02-14 19:18:29.850 ERROR 92282 --- [pool-1-thread-1] o.s.i.ip.tcp.TcpOutboundGateway          : Cannot correlate response - no pending reply for server:15099:49550:a3bc44fa-7d36-483c-a1b8-f91eea62d839
2019-02-14 19:18:29.851 ERROR 92282 --- [pool-1-thread-1] o.s.i.ip.tcp.TcpOutboundGateway          : Cannot correlate response - no pending reply for server:15099:49550:a3bc44fa-7d36-483c-a1b8-f91eea62d839
2019-02-14 19:18:29.851 ERROR 92282 --- [pool-1-thread-1] o.s.i.ip.tcp.TcpOutboundGateway          : Cannot correlate response - no pending reply for server:15099:49550:a3bc44fa-7d36-483c-a1b8-f91eea62d839
2019-02-14 19:18:29.851 ERROR 92282 --- [pool-1-thread-1] o.s.i.ip.tcp.TcpOutboundGateway          : Cannot correlate response - no pending reply for server:15099:49550:a3bc44fa-7d36-483c-a1b8-f91eea62d839
2019-02-14 19:18:29.852 ERROR 92282 --- [pool-1-thread-1] o.s.i.ip.tcp.TcpOutboundGateway          : Cannot correlate response - no pending reply for server:15099:49550:a3bc44fa-7d36-483c-a1b8-f91eea62d839
2019-02-14 19:18:29.852 ERROR 92282 --- [pool-1-thread-1] o.s.i.ip.tcp.TcpOutboundGateway

主要功能是

message = new GenericMessage<String>("sessions.list\n.\n");
replyMessage = simpleGateway.send(message);
System.out.println(replyMessage);

我试图删除最后一个“\n”

message = new GenericMessage<String>("sessions.list\n.");

它不起作用，出现超时异常。如何删除这些“无法关联响应”错误？

更新 1：

我认为服务器响应了几行消息：

sessions.list
.
response
,status_code,0
,status_message,OK
,time,2019-02-16 00:10:49 (+1000)
sessions
.

我需要捕获“.”之前的所有响应。

【问题讨论】：

标签： tcp spring-integration tcpclient

【解决方案1】：

如果您不期待回复，您应该使用出站通道适配器而不是网关。

【讨论】：

【解决方案2】：

终于解决了。我将发布我的解决方案供其他人参考。

我的 TCP 客户端发送以“.”结尾的多行命令，并期望多行响应也以“.”结尾从外部服务器。所以我需要编写自定义序列化器和反序列化器。默认的 CRLF 序列化器/反序列化器不符合我的情况。

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;


import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;

public class CustomSerializerDeserializer implements Serializer<String>, Deserializer<String> {

    @Override
    public String deserialize(InputStream inputStream) throws IOException {
        // TODO Auto-generated method stub
        StringBuilder builder = new StringBuilder();
        int c;
        while (true) {
            c = inputStream.read();
            builder.append((char)c);
            if ((char)c == '.') {
                break;
            }
        }

        return builder.toString();
    }

    @Override
    public void serialize(String object, OutputStream outputStream) throws IOException {
        // TODO Auto-generated method stub
        outputStream.write(object.getBytes());
        outputStream.flush();   
    }
}

xml配置是

<int-ip:tcp-connection-factory
    id="clientFactory"
    type="client"
    host="server"
    port="15099"
    ssl-context-support="sslContext"
    serializer="customeSerilizerDeserlizer"
    deserializer="customeSerilizerDeserlizer"
    single-use="true"
    so-timeout="10000"/>

【讨论】：