Spring集成网关与适配器

Question

这里排着队

@Component
@MessagingGateway(defaultRequestChannel = "toTcp.input")
public interface TcpClientGateway {
    byte[] send(String data, @Header("host") String host, @Header("port") int port);
}

private MessageChannel createNewSubflow(Message<?> message) {
        String host = (String) message.getHeaders().get("host");
        Integer port = (Integer) message.getHeaders().get("port");
        Assert.state(host != null && port != null, "host and/or port header missing");
        String hostPort = host + port;

        TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory(host, port);
        TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
        handler.setConnectionFactory(cf);
        IntegrationFlow flow = f -> f.handle(handler);
        IntegrationFlowContext.IntegrationFlowRegistration flowRegistration =
                this.flowContext.registration(flow)
                        .addBean(cf)
                        .id(hostPort + ".flow")
                        .register();
        MessageChannel inputChannel = flowRegistration.getInputChannel();
        this.subFlows.put(hostPort, inputChannel);
        return inputChannel;
    }

但我用

private MessageChannel createNewSubflow(Message<?> message) {
    String host = (String) message.getHeaders().get("host");
    Integer port = (Integer) message.getHeaders().get("port");
    Assert.state(host != null && port != null, "host and/or port header missing");
    String hostPort = host + port;

    TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory(host, port);
    cf.setLeaveOpen(true);
    //cf.setSingleUse(true);

    ByteArrayCrLfSerializer  byteArrayCrLfSerializer =new ByteArrayCrLfSerializer();
    byteArrayCrLfSerializer.setMaxMessageSize(1048576);

    cf.setSerializer(byteArrayCrLfSerializer);
    cf.setDeserializer(byteArrayCrLfSerializer);

    TcpOutboundGateway tcpOutboundGateway = new TcpOutboundGateway();
    tcpOutboundGateway.setConnectionFactory(cf);

    IntegrationFlow flow = f -> f.handle(tcpOutboundGateway);

    IntegrationFlowContext.IntegrationFlowRegistration flowRegistration =
        this.flowContext.registration(flow)
            .addBean(cf)
            .id(hostPort + ".flow")
            .register();
    MessageChannel inputChannel = flowRegistration.getInputChannel();

    this.subFlows.put(hostPort, inputChannel);
    return inputChannel;
}

使用请求/响应体系结构。它确实工作得很好，因为它提供了动态路由，手动创建tcp客户端。

Answer 1

你只需要注册两个流；一个用于输入；一个用于输出--问题是将响应与应答相关联，并将任意消息路由到网关之外的某个地方。

我在这个分支上更新了这个用例的示例。

您可以看到该分支上一次提交中的更改；大多数更改都是为了模拟服务器端。