如何使用 Spring Boot RSocket 服务器支持 WebSocket 传输？

Question

TLDR：配置暴露支持 WebSocket 传输的 RSocket 接口的 Spring Boot 应用程序需要什么？

---

我正在同时学习 RSocket 和 Spring Boot，所以请多多包涵。

在我的努力中，我已经能够构建一个非常简单和人为的 Spring Boot 应用程序实现，它使用第二个使用 RSocket 作为协议的 Spring Boot 应用程序提供/公开的 API，但是，我只能在以下情况下实现这一点使用TcpClientTransport.

从我的角度来看，WebsocketTransport它更有可能被使用并且对客户端-> 服务器架构更有用，但是，我还没有找到任何关于如何正确配置使用 WebSocket 作为接受 RSocket 消息的 Spring Boot 应用程序的工作示例或文档运输。

奇怪的是，在我的测试中，我的消费者（客户端）似乎确实建立了到服务器/生产者的 WebSocket 连接，但是，“握手”似乎挂起并且连接从未完全建立。我已经对 JavaScript 库（rsocket-websocket-client、rsocket-rpc-core 等）和 Java 库（io.rsocket.transport.netty.client.WebsocketClientTransport）进行了测试，并且服务器似乎表现出相同行为不管。

重申一下，使用 TCPTransport 我能够连接到服务器并调用请求就好了，但是在使用WebsocketTransport连接时从未建立。

一个旨在通过WebsocketClientTransport过去消费spring-boot-starter-rsocket作为依赖项支持 RSocket 的 Spring Boot 应用程序需要什么？

服务器

---

pom.xml

...

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.2.0.M5</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

...

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>

    <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>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
</dependencies>

...

应用程序属性

spring.rsocket.server.port=8081
management.endpoints.enabled-by-default=true
management.endpoints.web.exposure.include=*

SpringBootRSocketServerApplication.java

spring.rsocket.server.port=8081
management.endpoints.enabled-by-default=true
management.endpoints.web.exposure.include=*

用户套接字控制器

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class SpringBootRSocketServerApplication {

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

启动日志

 :: Spring Boot ::             (v2.2.0.M5)

2019-09-08 21:40:02,986 INFO  [main] org.springframework.boot.StartupInfoLogger: Starting SpringBootRSocketServerApplication on REDACTED with PID 22540 (REDACTED)
2019-09-08 21:40:02,988 INFO  [main] org.springframework.boot.SpringApplication: No active profile set, falling back to default profiles: default
2019-09-08 21:40:04,103 INFO  [main] org.springframework.boot.actuate.endpoint.web.EndpointLinksResolver: Exposing 14 endpoint(s) beneath base path '/actuator'
2019-09-08 21:40:04,475 INFO  [main] org.springframework.boot.rsocket.netty.NettyRSocketServer: Netty RSocket started on port(s): 8081
2019-09-08 21:40:04,494 INFO  [main] org.springframework.boot.web.embedded.netty.NettyWebServer: Netty started on port(s): 8080
2019-09-08 21:40:04,498 INFO  [main] org.springframework.boot.StartupInfoLogger: Started SpringBootRSocketServerApplication in 1.807 seconds (JVM running for 2.883)

消费者/客户

---

客户端配置文件

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;
import java.util.Random;
import java.util.stream.Stream;

@Slf4j
@Controller
public class UserRSocketController {

    @Autowired
    private UserRepository userRepository;

    @MessageMapping("usersList")
    public Mono<List<User>> usersList() {
        log.info("Handling usersList request.");
        return Mono.just(this.userRepository.getUsers());
    }

    @MessageMapping("usersStream")
    Flux<User> usersStream(UserStreamRequest request) {
        log.info("Handling request for usersStream.");
        List<User> users = userRepository.getUsers();
        Stream<User> userStream = Stream.generate(() -> {
            Random rand = new Random();
            return users.get(rand.nextInt(users.size()));
        });
        return Flux.fromStream(userStream).delayElements(Duration.ofSeconds(1));
    }

    @MessageMapping("userById")
    public Mono<User> userById(GetUserByIdRequest request) {
        log.info("Handling request for userById id: {}.", request.getId());
        return Mono.just(this.userRepository.getUserById(request.getId()));
    };
}

启动日志

 :: Spring Boot ::             (v2.2.0.M5)

2019-09-08 21:40:52,331 INFO  [main] org.springframework.boot.StartupInfoLogger: Starting SpringBootRsocketConsumerApplication on REDACTED with PID 18904 (REDACTED)
2019-09-08 21:40:52,334 INFO  [main] org.springframework.boot.SpringApplication: No active profile set, falling back to default profiles: default

Answer 1

您只需要两件事就可以让 RSocket 应用程序使用 websocket 传输公开端点：

首先，您需要 webflux 和 rsocket 依赖项，因为您可能还需要提供网页和静态资源：

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-rsocket</artifactId>
    </dependency>

然后你需要在你的application.properties文件中相应地配置 RSocket 服务器：

#server.port=8080 this is already the default
spring.rsocket.server.transport=websocket
spring.rsocket.server.mapping-path=/rsocket

您可以在有关 RSocket的Spring Boot 参考文档中找到更多相关信息。

websocket 客户端现在可以连接到ws://localhost:8080/rsocket.

请注意，从当前的 2.2.0 SNAPSHOT 开始，RSocket 协议已经发展，并且 rsocket-js 库目前正在迎头赶上，尤其是在元数据支持方面。您还可以在此处找到工作示例。

在 Java 客户端方面，Spring Boot 为您提供了一个RSocketRequester.Builder已经配置和定制的编解码器和拦截器，以满足您的需求：

@Component
public class MyService {

    private final RSocketRequester rsocketRequester;

    public MyService(RSocketRequester.Builder builder) {
        this.rsocketRequester = builder
                .connectWebSocket(URI.create("ws://localhost:8080/rsocket"))
                .block();
    }
}