java Spring Integration 和 TCP 服务器套接字 - 如何向客户端发送消息？

Spring Integration and TCP server socket - how can I send a message to a client?

提问by Aron Lorincz

I'm trying to create a server in Spring that's listening on a TCP port and accepts connections. I know how to route incoming requests to my service, and it can respond to those. However I would like to send messages to certain clients without any request received. For example, sometimes I have to inform a client about that it has got a message.

我正在尝试在 Spring 中创建一个侦听 TCP 端口并接受连接的服务器。我知道如何将传入请求路由到我的服务，并且它可以响应这些请求。但是，我想在没有收到任何请求的情况下向某些客户发送消息。例如，有时我必须通知客户它收到了一条消息。

To do this, I think I need a way to identify the clients, e.g. by letting them log in. Is there a way to have a "session" object for each active connection in which I can store login data?

为此，我想我需要一种方法来识别客户端，例如让他们登录。有没有办法为每个活动连接创建一个“会话”对象，我可以在其中存储登录数据？

How could I send a message to a client which has logged in with username X?

如何向使用用户名 X 登录的客户端发送消息？

Is this possible in Spring at all?

这在春天有可能吗？

采纳答案by Gary Russell

Starting with version 3.0; the frameworks now emits connection events when there are connection state changes. You can capture these events using an ApplicationListener, or using an <event:inbound-channel-adapter/>.

从 3.0 版开始；当连接状态发生变化时，框架现在会发出连接事件。您可以使用ApplicationListener或 使用捕获这些事件<event:inbound-channel-adapter/>。

The TcpConnectionOpenEventcontains a connectionId; you can send arbitrary messages to any connection once you know its id, by populating the IpHeaders.connectionIdheader (ip_connectionId) in a message and sending it to a <tcp:outbound-channel-adapter/>.

该TcpConnectionOpenEvent包含一个connectionId; 一旦您知道其 ID，您就可以向任何连接发送任意消息，方法是在消息中填充IpHeaders.connectionId标头 ( ip_connectionId) 并将其发送到<tcp:outbound-channel-adapter/>.

If you need to support request/reply as well as sending arbitrary messages, you need to use a collaborating pair of channel adapters for all communication, not a gateway.

如果您需要支持请求/回复以及发送任意消息，您需要使用一对协作的通道适配器进行所有通信，而不是网关。

EDIT

编辑

Here's a simple Boot app...

这是一个简单的启动应用程序...

package com.example;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.Socket;

import javax.net.SocketFactory;

import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
import org.springframework.integration.ip.tcp.connection.TcpConnectionOpenEvent;
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpServerConnectionFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

@SpringBootApplication
public class So25102101Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(So25102101Application.class)
                .web(false)
                .run(args);
        int port = context.getBean(TcpServerConnectionFactory.class).getPort();
        Socket socket = SocketFactory.getDefault().createSocket("localhost", port);
        BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
        String line = reader.readLine();
        System.out.println(line);
        context.close();
    }

    @Bean
    public TcpReceivingChannelAdapter server(TcpNetServerConnectionFactory cf) {
        TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter();
        adapter.setConnectionFactory(cf);
        adapter.setOutputChannel(inputChannel());
        return adapter;
    }

    @Bean
    public MessageChannel inputChannel() {
        return new QueueChannel();
    }

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

    @Bean
    public TcpNetServerConnectionFactory cf() {
        return new TcpNetServerConnectionFactory(0);
    }

    @Bean
    public IntegrationFlow outbound() {
        return IntegrationFlows.from(outputChannel())
                .handle(sender())
                .get();
    }

    @Bean
    public MessageHandler sender() {
        TcpSendingMessageHandler tcpSendingMessageHandler = new TcpSendingMessageHandler();
        tcpSendingMessageHandler.setConnectionFactory(cf());
        return tcpSendingMessageHandler;
    }

    @Bean
    public ApplicationListener<TcpConnectionOpenEvent> listener() {
        return new ApplicationListener<TcpConnectionOpenEvent>() {

            @Override
            public void onApplicationEvent(TcpConnectionOpenEvent event) {
                outputChannel().send(MessageBuilder.withPayload("foo")
                        .setHeader(IpHeaders.CONNECTION_ID, event.getConnectionId())
                        .build());
            }

        };
    }

}

pom deps:

pom deps：

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-integration</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-ip</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>