作者：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 4.1 WebSocket 支持之上使用了Spring Integration Java 配置 DSL。以下是示例：

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很简单。对于传入的每条消息，将其复制并将其发送到每个正在侦听的WebSocket会话，方法是添加一个具有SimpMessageHeaderAccessor.SESSION_ID_HEADER的标头，然后将其发送到输出webSocketOutboundAdapter，该适配器会将其传递给每个正在侦听的客户端。要查看其工作情况，请在一个浏览器窗口中打开https://127.0.0.1:8080/，然后在另一个窗口中打开https://127.0.0.1:8080/hi/Spring。有一个简单的客户端在此技术提示的代码库中演示。

有关如何在Spring Integration 4.1 文档中使用 WebSocket 组件的详细文档。在Spring Integration 示例目录中也有一个更具启发性的示例。