Graphql Spring boot 订阅客户端

Question

我正在尝试使用 Rsocket 和 spring boot 为 graphql 实现一个简单的订阅客户端。我的服务工作正常，我知道我的订阅映射工作正常，因为我可以使用 rsc 从命令行检索它。但是当我尝试获取订阅并使用 RSocketGraphQlClient 时，它无法从响应中映射字段。它返回 Scala Map.Map1 对象，它无法转换为我的对象的实体（简单的 Pojo 映射响应）。我能找到的所有示例都使用返回类型 String 并且非常简单，因此无法找到更全面的内容。任何帮助将不胜感激，或者如果你能指出我很好的参考文档来为 Graphql 编写 rsocket 订阅客户端。据我所知，我可能不会在我的返回对象上强制模式，但我无法弄清楚如果是这样的话。下面是相关代码：

var rsocketDoc = """
                subscription {
                    getSecuritiesSubscription {
                        secId
                        rowStatusType
                        label
                    }
                }
                """;
var d = rSocketGraphQlClient
            .document(rsocketDoc)
            .executeSubscription()
            .mapNotNull(response -> {
                if (!response.isValid()) {
                    log.info(response.getData());
                    throw new RuntimeException(response.getErrors().toString());
                }
                final SecurityQl data1 = response.field("getSecuritiesSubscription").toEntity(SecurityQl.class);
                    log.info("data " + data1);
                    return data1;
                });
log.info(d.toString());
d.subscribe(System.out::println);

堆栈跟踪：

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalArgumentException: Invalid path [getSecuritiesSubscription], data: Map(getSecuritiesSubscription -> Map(secId -> 111111, rowStatusType -> null, label -> Blah Blah Blah))
Caused by: java.lang.IllegalArgumentException: Invalid path [getSecuritiesSubscription], data: Map(getSecuritiesSubscription -> Map(secId -> 111111, rowStatusType -> null, label -> Blah Blah Blah))
    at org.springframework.util.Assert.isTrue(Assert.java:140) ~[spring-core-6.0.2.jar:6.0.2]
    at org.springframework.graphql.support.AbstractGraphQlResponse$DefaultResponseField.initFieldValue(AbstractGraphQlResponse.java:123) ~[spring-graphql-1.1.0.jar:1.1.0]
    at org.springframework.graphql.support.AbstractGraphQlResponse$DefaultResponseField.<init>(AbstractGraphQlResponse.java:71) ~[spring-graphql-1.1.0.jar:1.1.0]
    at org.springframework.graphql.support.AbstractGraphQlResponse.field(AbstractGraphQlResponse.java:46) ~[spring-graphql-1.1.0.jar:1.1.0]
    at org.springframework.graphql.client.DefaultClientGraphQlResponse.field(DefaultClientGraphQlResponse.java:66) ~[spring-graphql-1.1.0.jar:1.1.0]
    at com.tfgam.stream.processor.controller.SecurityController.lambda$getSecuritySub$2(SecurityController.java:141) ~[classes/:na]
    at reactor.core.publisher.Flux.lambda$mapNotNull$24(Flux.java:6428) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:112) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:250) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.5.0.jar:3.5.0]
    at io.rsocket.core.ResolvingOperator$DeferredResolution.onNext(ResolvingOperator.java:513) ~[rsocket-core-1.1.3.jar:na]
    at io.rsocket.core.RequestStreamRequesterFlux.handlePayload(RequestStreamRequesterFlux.java:330) ~[rsocket-core-1.1.3.jar:na]
    at io.rsocket.core.ReassemblyUtils.handleNextSupport(ReassemblyUtils.java:94) ~[rsocket-core-1.1.3.jar:na]
    at io.rsocket.core.RequestStreamRequesterFlux.handleNext(RequestStreamRequesterFlux.java:407) ~[rsocket-core-1.1.3.jar:na]
    at io.rsocket.core.RSocketRequester.handleFrame(RSocketRequester.java:255) ~[rsocket-core-1.1.3.jar:na]
    at io.rsocket.core.RSocketRequester.handleIncomingFrames(RSocketRequester.java:211) ~[rsocket-core-1.1.3.jar:na]
    at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160) ~[reactor-core-3.5.0.jar:3.5.0]
    at io.rsocket.core.ClientServerInputMultiplexer$InternalDuplexConnection.onNext(ClientServerInputMultiplexer.java:248) ~[rsocket-core-1.1.3.jar:na]
    at io.rsocket.core.ClientServerInputMultiplexer.onNext(ClientServerInputMultiplexer.java:129) ~[rsocket-core-1.1.3.jar:na]
    at io.rsocket.core.ClientServerInputMultiplexer.onNext(ClientServerInputMultiplexer.java:48) ~[rsocket-core-1.1.3.jar:na]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:122) ~[reactor-core-3.5.0.jar:3.5.0]
    at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:377) ~[reactor-netty-core-1.1.0.jar:1.1.0]
    at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:411) ~[reactor-netty-core-1.1.0.jar:1.1.0]
    at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:113) ~[reactor-netty-core-1.1.0.jar:1.1.0]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:346) ~[netty-codec-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:333) ~[netty-codec-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:454) ~[netty-codec-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:290) ~[netty-codec-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:444) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562) ~[netty-transport-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[netty-common-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.85.Final.jar:4.1.85.Final]
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.85.Final.jar:4.1.85.Final]
    at java.base/java.lang.Thread.run(Thread.java:1589) ~[na:na]

Answer 1

很容易将结果映射到您的 gql 类型。

1. 注入objectMapper。
2. 将 data 值（在响应映射中）转换为您的 gql 类型。

   我的 Spring Graphql 示例项目中有一个示例（用 Kotlin 编写）。

   client.document(subscriptionQuery)
       .executeSubscription()
       .map { response: ClientGraphQlResponse ->
            objectMapper.convertValue(
                response.getData<Map<String, Any>>()!!["commentAdded"],
                Comment::class.java
            )
        }
        .subscribe { comments.add(it) }
   

   检查使用 RSocketGraphQlClient 的 the complete integration tests in my example project。