Spring Boot Integration-在连接初始化时发送问候

Question

null

@SpringBootApplication
@EnableIntegration
public class ExampleApp {

    public static void main(String[] args) {
        SpringApplication.run(ExampleApp.class, args);
    }

    // Create listener on port 1234
    @Bean
    public AbstractServerConnectionFactory serverConnectionFactory() {
        TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(1234);
        return tcpNetServerConnectionFactory;
    }

    // Inbound channel
    @Bean
    public MessageChannel requestChannel() {
        return new DirectChannel();
    }

    // Outbound channel
    @Bean
    public MessageChannel replyChannel() {
        return new DirectChannel();
    }

    // Inbound gateway
    @Bean
    public TcpInboundGateway tcpInboundGateway(AbstractServerConnectionFactory serverConnectionFactory, MessageChannel requestChannel, MessageChannel replyChannel) {
        TcpInboundGateway tcpInboundGateway = new TcpInboundGateway();
        tcpInboundGateway.setConnectionFactory(serverConnectionFactory);
        tcpInboundGateway.setRequestChannel(requestChannel);
        tcpInboundGateway.setReplyChannel(replyChannel);
        return tcpInboundGateway;
    }

    // Send reply for incoming message -> working
    @ServiceActivator(inputChannel = "requestChannel", outputChannel = "replyChannel")
    public Message<String> processMessage(Message<String> message) {
        Message reply = MessageBuilder
                .withPayload("OK")
                .setHeader(IpHeaders.CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class))
                .build();
        return reply;
    }

    // Send greeting -> not working
    @Bean
    public ApplicationListener<TcpConnectionEvent> listener(MessageChannel replyChannel) {
        return tcpConnectionEvent -> {
            if (tcpConnectionEvent instanceof TcpConnectionOpenEvent) {
                Message<String> message = MessageBuilder
                        .withPayload("Hello client")
                        .setHeader(IpHeaders.CONNECTION_ID, tcpConnectionEvent.getConnectionId())
                        .build();
                replyChannel.send(message);
            }
        };
    }
}

nc -C localhost 1234

连接已建立，但日志中出现以下错误：

未能发布TcpConnectionOpenEvent[Source=TCPNetConnection:LocalHost:37656:1234:187CFBC2-7E5D-4F4E-97DE-1A3B55A4E264],[Factory=ServerConnectionFactory,ConnectionId=LocalHost:37656:1234:187CFBC2-7E5D-4F4E-97DE-1A3B55A4E264]打开：Dispatcher没有通道“Application.ReplyChannel”的订阅服务器

如果我向服务器发送一个字符串，他会按预期回复“OK”。

多亏了加里·罗素的评论，我找到了解决办法。入站网关必须“拆分为”入站/出站通道适配器对。以下是完整的工作示例：

@SpringBootApplication
@EnableIntegration
public class ExampleApp {

    public static void main(String[] args) {
        SpringApplication.run(ExampleApp.class, args);
    }

    // Create listener on port 1234
    @Bean
    public AbstractServerConnectionFactory serverConnectionFactory() {
        TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(1234);
        return tcpNetServerConnectionFactory;
    }

    // Inbound channel
    @Bean
    public MessageChannel requestChannel() {
        return new DirectChannel();
    }

    // Outbound channel
    @Bean
    public MessageChannel replyChannel() {
        return new DirectChannel();
    }

    // Inbound channel adapter
    @Bean
    public TcpReceivingChannelAdapter receivingChannelAdapter(AbstractServerConnectionFactory serverConnectionFactory, MessageChannel requestChannel) {
        TcpReceivingChannelAdapter tcpReceivingChannelAdapter = new TcpReceivingChannelAdapter();
        tcpReceivingChannelAdapter.setConnectionFactory(serverConnectionFactory);
        tcpReceivingChannelAdapter.setOutputChannel(requestChannel);
        return tcpReceivingChannelAdapter;
    }

    // Outbound channel adapter
    @Bean
    @ServiceActivator(inputChannel = "replyChannel")
    public TcpSendingMessageHandler tcpSendingMessageHandler(AbstractServerConnectionFactory serverConnectionFactory) {
        TcpSendingMessageHandler tcpSendingMessageHandler = new TcpSendingMessageHandler();
        tcpSendingMessageHandler.setConnectionFactory(serverConnectionFactory);
        return tcpSendingMessageHandler;
    }

    // Send reply for incoming message -> working
    @ServiceActivator(inputChannel = "requestChannel", outputChannel = "replyChannel")
    public Message<String> processMessage(Message<String> message) {
        Message<String> reply = MessageBuilder
                .withPayload("OK")
                .setHeader(IpHeaders.CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class))
                .build();
        return reply;
    }

    // Send greeting -> now working
    @Bean
    public ApplicationListener<TcpConnectionEvent> listener(MessageChannel replyChannel) {
        return tcpConnectionEvent -> {
            if (tcpConnectionEvent instanceof TcpConnectionOpenEvent) {
                Message<String> message = MessageBuilder
                        .withPayload("Hello client")
                        .setHeader(IpHeaders.CONNECTION_ID, tcpConnectionEvent.getConnectionId())
                        .build();
                replyChannel.send(message);
            }
        };
    }
}

现在，客户端在建立连接时得到问候语“Hello client”，在每个发送消息时得到答复“OK”。

Answer 1

ReplyChannel.Send（消息）；

你不能那样做；当第一个请求进来时，应答通道被连接起来。

在任何情况下，您都不能使用这样的网关，应答通道是用于对请求的应答，而不是用于发送任意消息。