使用“wrap”函数创建时出现 Spring RSocketRequester 问题

Question

当我像这样创建RSocketRequester 时它工作正常：

@Bean
RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) {
    return RSocketRequester.builder()
            .rsocketStrategies(rSocketStrategies)
            .connectTcp("localhost", 7000)
            .block();
}

但是当我尝试发送消息时它会抛出异常：

@Bean
RSocket rSocket() {
    return RSocketFactory
            .connect()
            .transport(TcpClientTransport.create(7000))
            .start()
            .block();
}

@Bean
RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) {
    return RSocketRequester.wrap(rSocket(), MediaType.APPLICATION_CBOR, MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()), rSocketStrategies);
}

io.rsocket.exceptions.ApplicationErrorException: 没有目标''的处理程序 在 io.rsocket.exceptions.Exceptions.from(Exceptions.java:45) ~[rsocket-core-1.0.0-RC5.jar:na] 抑制：reactor.core.publisher.FluxOnAssembly$OnAssemblyException： 在以下站点观察到错误： |_ 检查点 ⇢ 处理程序 com.issoft.rnd.ms2.controller.TestController#test4() [DispatcherHandler] |_ 检查点 ⇢ HTTP GET "/test4" [ExceptionHandlingWebHandler] 堆栈跟踪： 在 io.rsocket.exceptions.Exceptions.from(Exceptions.java:45) ~[rsocket-core-1.0.0-RC5.jar:na] 在 io.rsocket.RSocketRequester.handleFrame(RSocketRequester.java:556) ~[rsocket-core-1.0.0-RC5.jar:na] 在 io.rsocket.RSocketRequester.handleIncomingFrames(RSocketRequester.java:516) ~[rsocket-core-1.0.0-RC5.jar:na] 在 reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:242) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drainRegular(FluxGroupBy.java:554) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.drain(FluxGroupBy.java:630) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$UnicastGroupedFlux.subscribe(FluxGroupBy.java:696) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.Flux.subscribe(Flux.java:8134) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.MonoFlatMapMany$FlatMapManyMain.onNext(MonoFlatMapMany.java:188) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1592) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.MonoProcessor.onNext(MonoProcessor.java:317) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 io.rsocket.internal.ClientServerInputMultiplexer.lambda$new$1(ClientServerInputMultiplexer.java:116) ~[rsocket-core-1.0.0-RC5.jar:na] 在 reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$GroupByMain.drainLoop(FluxGroupBy.java:380) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$GroupByMain.drain(FluxGroupBy.java:316) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxGroupBy$GroupByMain.onNext(FluxGroupBy.java:201) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114) ~[reactor-core-3.3.0.RELEASE.jar:3.3.0.RELEASE] 在 reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:213) ~[reactor-netty-0.9.1.RELEASE.jar:0.9.1.RELEASE] 在 reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:346) ~[reactor-netty-0.9.1.RELEASE.jar:0.9.1.RELEASE] 在 reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:348) ~[reactor-netty-0.9.1.RELEASE.jar:0.9.1.RELEASE] 在 reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:93) ~[reactor-netty-0.9.1.RELEASE.jar:0.9.1.RELEASE] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:326) ~[netty-codec-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:300) ~[netty-codec-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1422) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:931) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:700) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:635) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:552) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:514) ~[netty-transport-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.util.concurrent.SingleThreadEventExecutor$6.run(SingleThreadEventExecutor.java:1050) ~[netty-common-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.43.Final.jar:4.1.43.Final] 在 io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.43.Final.jar:4.1.43.Final] 在 java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

发送功能：

public Mono<String> request(String route, String data, Class<String> clazz) {
    return rSocketRequester.route(route).data(data).retrieveMono(clazz);
}

服务器：

@Controller
@AllArgsConstructor
public class RTestController {

    private EchoService echoService;

    @MessageMapping("test")
    Mono<String> test(String value) {
        return echoService.echo(value);
    }
}

Spring Boot 版本： 2.2.1.发布

注意： Mime 类型取自第一个工作版本。

wrap 函数有什么问题？

Answer 1

添加dataMimeType 和metadataMimeType 有帮助。

@Bean
RSocket rSocket() {
    return RSocketFactory
            .connect()
            .dataMimeType(MediaType.APPLICATION_CBOR_VALUE)
            .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString())
            .transport(TcpClientTransport.create(7000))
            .start()
            .block();
}

使用 wrap 方法的最初原因是使用 LoadBalancedRSocketMono，它提供客户端负载平衡和重新连接 OOTB。

LoadBalancedRSocketMono.create(Flux.just(Collections.singleton(new RSocketSupplier(() ->
                        RSocketFactory
                                .connect()
                                .dataMimeType(MediaType.APPLICATION_CBOR_VALUE)
                                .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString())
                                .frameDecoder(PayloadDecoder.ZERO_COPY)
                                .transport(TcpClientTransport.create("localhost", 7000))
                                .start()
                                .doOnNext(rSocket -> {
                                    this.rSocketRequester = RSocketRequester.wrap(rSocket, MediaType.APPLICATION_CBOR, MimeTypeUtils.parseMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()), rSocketStrategies);
                                })
                )))
        ).subscribe();

Answer 2

TL;DR：这不是 Spring 的问题，而是您创建客户端 RSocket 连接的方式。

RSocketRequester.wrap 方法通常用于服务器端，一旦客户端发起连接并且服务器想要向对方发送请求。 Javadoc 说您可以在客户端使用它，这是一个完全有效的用例。

现在RSocketRequester.wrap 在其参数中接受有关预期数据mime 类型和元数据mime 类型的数据。这通常在建立连接时由两方协商，使用 RSocket SETUP/METADATA_PUSH 帧。一旦建立连接，它就完成了。 wrap 方法需要这些参数，因为我们无法从现有的 RSocket 中轻松提取这些信息。

在您的情况下，已使用 RSocketFactory 创建了连接，但未提供该信息（ClientRSocketFactory 类为此提供了许多方法）。所以我认为元数据无法在请求中正确编码，导致服务器无法读取路由信息。

除非您有非常具体的要求，否则我建议使用RSocketRequester.Builder API 创建客户端 RSocket，因为这样更容易正确处理。