RSocket与springboot整合

ゝ一世哀愁。 2023-02-14 08:00 18阅读 0赞

**一、依赖：**

<parent>
    	        <groupId>org.springframework.boot</groupId>
    	        <artifactId>spring-boot-starter-parent</artifactId>
    	        <version>2.3.0.RELEASE</version>
    	        <relativePath/> 
            </parent>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-rsocket</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-webflux</artifactId>
            </dependency>

**二、配置：**

package com.xxxx.faceadd.rsocket;
    
    import io.rsocket.RSocketFactory;
    import io.rsocket.SocketAcceptor;
    import io.rsocket.transport.ServerTransport;
    import io.rsocket.transport.netty.server.WebsocketRouteTransport;
    import org.springframework.boot.web.embedded.netty.NettyRouteProvider;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
    import org.springframework.web.reactive.config.EnableWebFlux;
    import reactor.netty.http.server.HttpServerRoutes;
    
    /** * Config */
    @EnableWebFlux
    @Configuration
    public class Config { 
    
        /** * RSocket 有自己的二进制协议，在浏览器端的实现需要使用 RSocket 提供的 JavaScript 客户端与服务器端交互。 * 在 Web 应用中使用 RSocket 提供的 NodeJS 模块 rsocket-websocket-client */
    
        @Bean
        RSocketWebSocketNettyRouteProvider rSocketWebSocketNettyRouteProvider(RSocketMessageHandler messageHandler) { 
            return new RSocketWebSocketNettyRouteProvider("/ws", messageHandler.responder());
        }
    
        static class RSocketWebSocketNettyRouteProvider implements NettyRouteProvider { 
    
            private final String mappingPath;
    
            private final SocketAcceptor socketAcceptor;
    
            RSocketWebSocketNettyRouteProvider(String mappingPath, SocketAcceptor  socketAcceptor) { 
                this.mappingPath = mappingPath;
                this.socketAcceptor = socketAcceptor;
            }
    
            @Override
            public HttpServerRoutes apply(HttpServerRoutes httpServerRoutes) { 
                ServerTransport.ConnectionAcceptor acceptor = RSocketFactory.receive()
                        .acceptor(this.socketAcceptor)
                        .toConnectionAcceptor();
                return httpServerRoutes.ws(this.mappingPath, WebsocketRouteTransport.newHandler(acceptor));
            }
        }
    
    }

package com.xxxx.faceadd.rsocket;
    
    import org.reactivestreams.Publisher;
    import org.springframework.messaging.handler.annotation.MessageMapping;
    import org.springframework.messaging.rsocket.RSocketRequester;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.GetMapping;
    import reactor.core.publisher.Flux;
    import reactor.core.publisher.Mono;
    
    import javax.annotation.Resource;
    
    @Controller
    public class ServerController { 
    
        @Resource
        private RSocketRequester rSocketRequester;
    
        /** * 客户端 * @return */
        @GetMapping(value = "/ping")
        public Publisher<String> ping(){ 
            return rSocketRequester.route("ping").data(new String()).retrieveMono(String.class);
        }
    
    
        /** * 客服端 * @param pong * @return */
        @MessageMapping("/pong")
        public Mono<String> pong(String pong) { 
            return Mono.just("ECHO >> " + pong);
        }
    
        /** * 客服端 * @param echo * @return */
        @MessageMapping("/echo")
        public Flux<String> echo(String echo) { 
            return Flux.fromStream(echo.codePoints().mapToObj(msg -> { 
                System.out.println("ECHO >> " + msg);
                return null;
            }));
        }
    
    }

package com.xxxx.faceadd.rsocket;
    
    import io.rsocket.AbstractRSocket;
    import io.rsocket.Payload;
    import io.rsocket.RSocket;
    import io.rsocket.RSocketFactory;
    import io.rsocket.transport.netty.client.TcpClientTransport;
    import io.rsocket.transport.netty.server.TcpServerTransport;
    import io.rsocket.util.DefaultPayload;
    import reactor.core.publisher.Mono;
    
    /** * RequestResponseExample 请求-响应流模式 */
    public class RequestResponseExample { 
    
        public static void main(String[] args) { 
            RSocketFactory.receive()
                    .acceptor((setupPayload, sendingSocket) -> Mono.just(
                            new AbstractRSocket() { 
                                @Override
                                public Mono<Payload> requestResponse(Payload payload) { 
                                    return Mono.just(DefaultPayload.create("ECHO >> " + payload.getDataUtf8()));
                                }
                            }
                    ))
                    // 指定传输层实现
                    .transport(TcpServerTransport.create("localhost", 7000))
                    // 启动
                    .start()
                    .subscribe();
    
    
            RSocket rSocket = RSocketFactory.connect()
                    // 指定传输层实现
                    .transport(TcpClientTransport.create("localhost", 7000))
                    // 启动客户端
                    .start()
                    .block();
    
            rSocket.requestResponse(DefaultPayload.create("hello"))
                    .map(Payload::getDataUtf8)
                    .doOnNext(System.out::println)
                    .block();
    
            rSocket.dispose();
    
        }
    
    }