使用 RSocket-Java for Spring Rsocket Server 的 rsocket 路由元数据

Question

如何设置路由元数据（在服务器使用 Spring Boot Rsocket 时仅使用 RSocket-Java 的负载中。

Flux<Payload> s = connection.flatMapMany(requester -> requester.requestStream(DefaultPayload.create("Some Message")))

服务器正在使用@MessageMapping("/route")

Answer 1

互动类型

使用 @MessageMapping 的 SpringBoot 上的 RSocket 交互类型是根据注释方法的签名决定的（更多信息在 中）

假设它有签名：

@MessageMapping("/route")
Flux<String> getStreamOfStrings(String message) {...}

基于 spring 文档交互类型的基数 table 是 Request-Stream。

RSocket 客户端

RSocket java 客户端需要为元数据指定 mime 类型：

RSocket rsocketClient = RSocketConnector.create()
    //metadata header needs to be specified
    .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString())
    // value of spring.rsocket.server.port eg 7000
    .connect(TcpClientTransport.create(7000))
    .block();

数据

数据将是简单的字符串：

ByteBuf data = ByteBufAllocator.DEFAULT.buffer().writeBytes("request msg".getBytes());

元数据

RSocket 中的路由定义为 metadata extension and needs to be sent together with data to specify routing. Here is example how it can be created (see other classes in package io.rsocket.metadata)

CompositeByteBuf metadata = ByteBufAllocator.DEFAULT.compositeBuffer();
RoutingMetadata routingMetadata = TaggingMetadataCodec.createRoutingMetadata(ByteBufAllocator.DEFAULT, List.of("/route"));
CompositeMetadataCodec.encodeAndAddMetadata(metadata,
        ByteBufAllocator.DEFAULT,
        WellKnownMimeType.MESSAGE_RSOCKET_ROUTING,
        routingMetadata.getContent());

请求流请求

已创建数据和元数据，因此您可以使用以下方式执行 requestSteam：

rsocketClient.requestStream(DefaultPayload.create(data, metadata))
    .map(Payload::getDataUtf8)
    .toIterable()
    .forEach(System.out::println);