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 + "\n" + 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屬性