WebFlux WebSocket学习

1. 引入相应webflux包
2. 实现自定义的请求处理类WebSocketHandler
3. 配置url映射关系及WebSocketHandlerAdapter
4. 通过页面进行测试

Server

要创建WebSocket服务器，您可以先创建一个MyWebSocketHandler实现WebSocketHandler。以下示例显示了如何执行此操作：
消息通过getPayloadAsText()获取内容后，再次获取则为空

package com.example.webflux.handler;

import com.example.webflux.service.TokenService;
import org.eclipse.jetty.util.MultiMap;
import org.eclipse.jetty.util.UrlEncoded;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.cors.reactive.CorsConfigurationSource;
import org.springframework.web.reactive.socket.*;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class MyWebSocketHandler implements WebSocketHandler, CorsConfigurationSource {

    @Autowired
    private TokenService tokenService;

    /**
     * 实现自定义的请求处理类WebSocketHandler
     */
    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // 在生产环境中，需对url中的参数进行检验，如token，不符合要求的连接的直接关闭
        HandshakeInfo handshakeInfo = session.getHandshakeInfo();
        if (handshakeInfo.getUri().getQuery() == null) {
            return session.close(CloseStatus.REQUIRED_EXTENSION);
        } else {
            // 对参数进行解析，在些使用的是jetty-util包
            MultiMap<String> values = new MultiMap<String>();
            UrlEncoded.decodeTo(handshakeInfo.getUri().getQuery(), values, "UTF-8");
            String token = values.getString("token");
            boolean isValidate = tokenService.validate(token);
            if (!isValidate) {
                return session.close();
            }
        }
        Flux<WebSocketMessage> output = session
                .receive() //访问入站消息流。
                .doOnNext(message ->{
                    //对每条消息执行一些操作
                    //对于嵌套的异步操作，您可能需要调用message.retain()使用池化数据缓冲区的基础服务器
                    WebSocketMessage retain = message.retain();
                })
                .concatMap(mapper -> {
                    //执行使用消息内容的嵌套异步操作
                    String msg = mapper.getPayloadAsText();
                    System.out.println("mapper: " + msg);
                    return Flux.just(msg);
                })
                .map(value -> {
                    // 创建出站消息，生成组合流
                    System.out.println("value: " + value);
                    return session.textMessage("Echo " + value);
                });
        return session.send(output);
    }

    /**
     * 直接使自定义的WebSocketHandler实现CorsConfigurationSource接口，并返回一个CorsConfiguration
     */
    @Override
    public CorsConfiguration getCorsConfiguration(ServerWebExchange exchange) {
        CorsConfiguration corsConfiguration = new CorsConfiguration();
        corsConfiguration.addAllowedOrigin("*");
        return corsConfiguration;
    }
}

TokenService: 校验参数，在生产环境需要对每个连接进行校验，符合要求的才允许连接

package com.example.webflux.service;

import org.springframework.stereotype.Service;

@Service
public class TokenService {
    // demo演示，在引只对长度做校验
    public boolean validate(String token) {
        if (token.length() > 5) {
            return true;
        }
        return false;
    }
}

WebConfig: 服务器配置

package com.example.webflux.config;

import com.example.webflux.handler.MyWebSocketHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.HandlerMapping;
import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class WebConfig {

    @Bean
    public MyWebSocketHandler getMyWebsocketHandler() {
        return new MyWebSocketHandler();
    }

    @Bean
    public HandlerMapping handlerMapping() {
        // 对相应的URL进行添加处理器
        Map<String, WebSocketHandler> map = new HashMap<>();
        map.put("/hello", getMyWebsocketHandler());
        SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
        mapping.setUrlMap(map);
        mapping.setOrder(-1);
        return mapping;
    }

    @Bean
    public WebSocketHandlerAdapter handlerAdapter() {
        return new WebSocketHandlerAdapter();
    }
}

index.html: 测试代码，内容如下：

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<textarea id="msgBoxs"></textarea><br>
待发送消息：<input type="text" id="msg"><input type="button" id="sendBtn" onclick="send()" value="发送">
<script type="application/javascript">
    var msgBoxs = document.getElementById("msgBoxs")
    var msgBox = document.getElementById("msg")
    document.cookie="token2=John Doe";
    var ws = new WebSocket("ws://localhost:9000/hello?token=aabb123&demo=1&userType=0")
    ws.onopen = function (evt) {
        console.log("Connection open ...");
        ws.send("Hello WebSocket!");
    }

    ws.onmessage = function (evt) {
        console.log("Received Message: ", evt.data)
        var msgs = msgBoxs.value
        msgBoxs.innerText = msgs + "
" + evt.data
        msgBoxs.scrollTop = msgBoxs.scrollHeight;
    }

    ws.onclose = function (evt) {
        console.log("Connect closed.");
    }



    function send() {
        var msg = msgBox.value
        ws.send(msg)
        msgBox.value = ""
    }
</script>
</body>
</html>

CORS

1. 直接使自定义的WebSocketHandler实现CorsConfigurationSource接口，并返回一个CorsConfiguration

public class MyWebSocketHandler implements WebSocketHandler, CorsConfigurationSource {


    /**
     * 实现自定义的请求处理类WebSocketHandler
     */
    @Override
    public Mono<Void> handle(WebSocketSession session) {
       ...
    }

    /**
     * 直接使自定义的WebSocketHandler实现CorsConfigurationSource接口，并返回一个CorsConfiguration
     */
    @Override
    public CorsConfiguration getCorsConfiguration(ServerWebExchange exchange) {
        CorsConfiguration corsConfiguration = new CorsConfiguration();
        corsConfiguration.addAllowedOrigin("*");
        return corsConfiguration;
    }
}

2. 可以在SimpleUrlHandler上设置corsConfigurations属性