Spring Integration et socket de Serveur TCP-comment envoyer un message à un client?
j'essaie de créer un serveur au printemps qui écoute sur un port TCP et accepte les connexions. Je sais comment acheminer les demandes reçues à mon service, et il peut y répondre. Cependant je voudrais envoyer des messages à certains clients sans aucune demande reçue. Par exemple, parfois, je dois informer un client qu'il a reçu un message.
pour ce faire, je pense que j'ai besoin d'un moyen d'identifier les clients, par exemple en les laissant se connecter. Est-il possible d'avoir une "session" objet pour chaque connexion active dans laquelle je peux stocker les données de connexion?
Comment puis-je envoyer un message à un client qui s'est connecté avec le nom d'utilisateur X?
Est-ce possible au Printemps?
1 réponses
a Partir de la version 3.0; la cadres de maintenant émet des événements de connexion quand il y a des changements d'état de connexion. Vous pouvez capturer ces événements en utilisant un ApplicationListener
, ou en utilisant un <event:inbound-channel-adapter/>
.
TcpConnectionOpenEvent
contient connectionId
; vous pouvez envoyer des messages arbitraires à n'importe quelle connexion une fois que vous connaissez son id, en peuplant le IpHeaders.connectionId
en-tête (ip_connectionId
) dans un message et de l'envoyer à un <tcp:outbound-channel-adapter/>
.
si vous avez besoin de soutenir la demande/réponse ainsi que l'envoi messages arbitraires, vous devez utiliser une paire collaborative d'adaptateurs de canaux pour toute communication, pas une passerelle.
EDIT
Voici une simple application de démarrage...
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:
<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>