RSocketGraphQlClient订阅响应映射POJO失败问题求助

用RSocket和Spring Boot实现GraphQL订阅客户端，服务端及订阅映射功能正常（已通过rsc命令行工具成功获取数据），但使用RSocketGraphQlClient时，无法将响应字段映射到自定义POJO（SecurityQl），返回的Scala Map.Map1对象无法转换。现有示例均为简单String返回类型，求解决方法或参考文档。

相关代码 #

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]

解决方法 #

1. 修正响应解析逻辑 #

堆栈信息显示响应数据是外层Map包含getSecuritiesSubscription键，对应值为内部数据Map。直接调用response.field()会报错，因为该方法期望当前层级存在目标字段，需先获取整个data对象再提取内部数据：

// 替换原field提取代码
Map<String, Object> dataMap = response.getData();
SecurityQl data1 = objectMapper.convertValue(dataMap.get("getSecuritiesSubscription"), SecurityQl.class);

需确保项目中配置了Jackson的ObjectMapper实例，Spring Boot默认会自动装配。

2. 使用包装类适配响应结构 #

定义包含顶层查询字段的包装类，直接将整个响应data转换为该类实例，再提取目标POJO：

// 定义包装类
public class SecuritySubscriptionResponse {
    private SecurityQl getSecuritiesSubscription;
    // 生成getter和setter
}

// 代码中使用
SecuritySubscriptionResponse responseWrapper = response.toEntity(SecuritySubscriptionResponse.class);
SecurityQl data1 = responseWrapper.getSecuritiesSubscription();

3. 确保POJO字段映射正确 #

检查SecurityQl类的字段名与GraphQL返回字段完全匹配，若存在命名差异，使用@JsonProperty注解指定映射：

public class SecurityQl {
    @JsonProperty("secId")
    private String secId;
    @JsonProperty("rowStatusType")
    private String rowStatusType;
    @JsonProperty("label")
    private String label;
    // 生成getter和setter
}

4. 处理Scala Map兼容问题 #

由于返回的是Scala Map，需添加Jackson Scala模块依赖并配置：

<!-- Maven依赖 -->
<dependency>
    <groupId>com.fasterxml.jackson.module</groupId>
    <artifactId>jackson-module-scala_2.13</artifactId>
</dependency>

Spring Boot环境下通过配置类注册模块：

@Configuration
public class JacksonConfig {
    @Bean
    public ObjectMapper objectMapper() {
        ObjectMapper mapper = new ObjectMapper();
        mapper.registerModule(new DefaultScalaModule());
        return mapper;
    }
}

内容的提问来源于stack exchange，提问作者juggernaut