作者：Josh Long

Spring Integration 4.1 已釋出，其中包含大量出色的新功能！我最喜歡的功能之一？與 Spring 4 WebSocket 支援的智慧整合。現在，您可以組合一個其最終目的地是 WebSocket 客戶端的整合流。還支援充當 WebSocket 服務的客戶端。

為了編譯它，您需要 Java 8（我們在此大量使用 lambda 表示式）以及以下 Maven 依賴項：

• groupId：org.springframework.integration，artifactId：spring-integration-java-dsl，version： 1.0.0.RC1。
• groupId：org.springframework.integration，artifactId：spring-integration-websocket，version： 4.1.0.RELEASE。
• groupId：org.springframework.boot，artifactId：spring-boot-starter-websocket，version： 1.2.0.RC1。

為了解析這些依賴項，您需要 snapshot 和 milestone Maven 儲存庫。

所有監聽 /names 的客戶端都將接收發送到 requestChannel 通道的訊息。Spring 4 MessageChannel 是一個命名通道，更多或更少類似於 java.util.Queue<T>。此示例使用 Spring Integration Java 配置 DSL 並在新的 Spring Integration 4.1 Web Socket 支援之上。

package demo ;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.*;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.websocket.ServerWebSocketContainer;
import org.springframework.integration.websocket.outbound.WebSocketOutboundMessageHandler;
import org.springframework.messaging.*;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.Executors;
import java.util.stream.Collectors;

/**
 * @author Artem Bilan
 * @author Josh Long
 */
@Configuration
@ComponentScan
@EnableAutoConfiguration
@RestController
public class Application {

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

    @Bean
    ServerWebSocketContainer serverWebSocketContainer() {
        return new ServerWebSocketContainer("/names").withSockJs();
    }

    @Bean
    MessageHandler webSocketOutboundAdapter() {
        return new WebSocketOutboundMessageHandler(serverWebSocketContainer());
    }

    @Bean(name = "webSocketFlow.input")
    MessageChannel requestChannel() {
        return new DirectChannel();
    }

    @Bean
    IntegrationFlow webSocketFlow() {
        return f -> {
            Function<Message , Object> splitter = m -> serverWebSocketContainer()
                    .getSessions()
                    .keySet()
                    .stream()
                    .map(s -> MessageBuilder.fromMessage(m)
                            .setHeader(SimpMessageHeaderAccessor.SESSION_ID_HEADER, s)
                            .build())
                    .collect(Collectors.toList());
            f.split( Message.class, splitter)
                    .channel(c -> c.executor(Executors.newCachedThreadPool()))
                    .handle(webSocketOutboundAdapter());
        };
    }

    @RequestMapping("/hi/{name}")
    public void send(@PathVariable String name) {
        requestChannel().send(MessageBuilder.withPayload(name).build());
    }
}

IntegrationFlow 非常簡單。對於接收到的每條訊息，複製它，然後透過新增一個帶有 SimpMessageHeaderAccessor.SESSION_ID_HEADER 的標頭將其定址給每個正在監聽的 WebSocket 會話，然後將其傳送到出站 webSocketOutboundAdapter，該介面卡將其傳遞給每個正在監聽的客戶端。要檢視其工作原理，請在一個瀏覽器視窗中開啟 https://:8080/，然後在另一個視窗中開啟 https://:8080/hi/Spring。這裡有一個簡單的客戶端 在此技術提示的程式碼儲存庫中進行了演示。

關於如何在 Spring Integration 4.1 文件的 Web Socket 元件 中使用 Web Socket，有很棒的文件。在 Spring Integration 示例目錄 中還有一個更具啟發性的示例。