弹簧集成的请求/响应套接字

Question

我有一个弹簧引导服务器应用程序，我需要创建一个ssl套接字并让客户端连接到它。 我希望能够向客户端发送一条消息（例如，在向我的服务器发送REST API请求时），并且还读取它发回的响应。

我设法到目前为止做的是：

1. 创建套接字，并允许客户端连接（从TcpEvent保存连接ID）
2. 根据要求
发送插座上的消息
3. 客户端接收请求并发送回一个响应

在这点 - 我不能够读取他们的答复（我看到他们通过发送Wireshark的回复）。即使我配置了使用与TcpSendingMessageHandler相同的连接工厂的TcpReceivingChannelAdapter。

从那时起 - 我可以发送请求，但客户端不会收到它们......我怀疑这是因为他的第一个响应没有在我身边处理（我无法访问客户端代码验证）。

这里是我的代码：

Config.java

@EnableIntegration 
@IntegrationComponentScan 
@Configuration 
public class SocketConfiguration implements ApplicationListener<TcpConnectionEvent> { 

    private static final org.slf4j.Logger log = LoggerFactory.getLogger("SocketConfiguration"); 

    @Bean 
    public AbstractServerConnectionFactory AbstractServerConnectionFactory() { 
     TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(40003); 
     DefaultTcpNetSSLSocketFactorySupport tcpNetSSLSocketFactory = tcpSocketFactorySupport(); 
     tcpNetServerConnectionFactory.setTcpSocketFactorySupport(tcpNetSSLSocketFactory); 
     return tcpNetServerConnectionFactory; 
    } 


    @Bean 
    public DefaultTcpNetSSLSocketFactorySupport tcpSocketFactorySupport() { 
     DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("keystore.jks", 
      "trustStore.jks", "123456", "123456"); 
     sslContextSupport.setProtocol("TLSv1.2"); 
     DefaultTcpNetSSLSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(sslContextSupport); 
     return tcpSocketFactorySupport; 
    } 

    @Bean 
    public static MessageChannel getResponseChannel() { 
     DirectChannel directChannel = new DirectChannel(); 
     directChannel.setComponentName("getResponseChannel"); 
     directChannel.setLoggingEnabled(true); 
     return directChannel; 
    } 

    @Bean 
    public static MessageChannel getInputMessageChannel() { 
     DirectChannel directChannel = new DirectChannel(); 
     directChannel.setComponentName("inputMessageChannel"); 
     directChannel.setLoggingEnabled(true); 
     return directChannel; 
    } 

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

    @Bean 
    public TcpReceivingChannelAdapter in(AbstractServerConnectionFactory connectionFactory) { 
     TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); 
     adapter.setOutputChannel(getInputMessageChannel()); 
     adapter.setConnectionFactory(connectionFactory); 
     adapter.setSendTimeout(5000); 
     return adapter; 
    } 

    @ServiceActivator(inputChannel="toClientChannel") 
    @Bean 
    public TcpSendingMessageHandler out(AbstractServerConnectionFactory connectionFactory) { 
     TcpSendingMessageHandler tcpSendingMessageHandler = new TcpSendingMessageHandler(); 
     tcpSendingMessageHandler.setConnectionFactory(connectionFactory); 
     tcpSendingMessageHandler.setLoggingEnabled(true); 
     return tcpSendingMessageHandler; 
    } 

    @Transformer(inputChannel = "invokeChannel", outputChannel = "toClientChannel") 
    public Message<String> headerBeforeSend(String message) throws Exception { 
     log.debug("send message to socket: {}", message); 
     Map.Entry<String, TcpConnection> connectionEntry = GetConnectionEntry(); 
     log.debug("connection id is: {}", connectionEntry.getKey()); 
     return MessageBuilder.withPayload(message) 
     .setHeader(IpHeaders.CONNECTION_ID,connectionEntry.getKey()) 
     .build(); 
    } 

    private static ConcurrentHashMap<String, TcpConnection> tcpConnections = new ConcurrentHashMap<>(); 

    @Override 
    public void onApplicationEvent(TcpConnectionEvent tcpEvent) { 
     TcpConnection source = (TcpConnection) tcpEvent.getSource(); 
     if (tcpEvent instanceof TcpConnectionOpenEvent) { 
     log.info("Socket Opened " + source.getConnectionId()); 
     tcpConnections.put(tcpEvent.getConnectionId(), source); 
     } else if (tcpEvent instanceof TcpConnectionCloseEvent) { 
     log.info("Socket Closed " + source.getConnectionId()); 
     if(tcpConnections.containsKey(source.getConnectionId())) 
      tcpConnections.remove(source.getConnectionId()); 
     } else if (tcpEvent instanceof TcpConnectionExceptionEvent) { 
     log.error("Error {}",tcpEvent.getCause().getMessage()); 
     if(tcpConnections.containsKey(source.getConnectionId())) 
      tcpConnections.remove(source.getConnectionId()); 
     } 
    } 
} 

Controller.java

@RestController 
@ControllerAdvice 
@RequestMapping("/socket") 
public class SocketController { 
    private final Logger log = LoggerFactory.getLogger(getClass()); 

    @Inject 
    MessageChannel invokeChannel; 

    @LogAspect 
    @PostMapping 
    public ResponseEntity sendMessage(@RequestBody SendMessageRequest request) throws Exception { 
     log.debug("Message is {}",request.get_message()); 
     String msg = "Some test message"; 
     MessagingTemplate template = new MessagingTemplate(); 
     template.send(invokeChannel, new GenericMessage<>(msg)); 
     return new ResponseEntity(HttpStatus.OK); 
    } 
} 

能否请您看看我的配置和引导我正确的设置？

感谢

Answer 1

您应该确保您发送和接收消息正确序列化和反序列化。默认情况下它是ByteArrayCrlfSerializer这是基于\r\n消息终止：https://docs.spring.io/spring-integration/docs/4.3.11.RELEASE/reference/html/ip.html#connection-factories