Spring Reactive WebFlux在使用Application/Stream+JSON时报告空流量

Question

我有一个反应核心WebClient可以发布到给定的endpoint。有效负载是JsonNodes的流量，内容类型是application/stream+json

JsonNode response = localEP.post().uri( "/createItem" )
    .contentType(MediaType.APPLICATION_STREAM_JSON)
    .body( BodyInserters.fromPublisher(itemData, JsonNode.class ))
    .retrieve()
    .bodyToMono( JsonNode.class )
    .block();

在服务器端，我尝试了Spring Controller样式和Spring Web Reactive FunctionHandler来处理上述调用的负载，负载是一个流量。

  @PostMapping(path = "/dev/jobad/dynamo", consumes = MediaType.APPLICATION_STREAM_JSON_VALUE)
  @ResponseStatus(HttpStatus.CREATED)
    public Flux<JsonNode> loadItems (@RequestBody Flux<JsonNode> items) {
       items.subscribe(storage::add);
       JsonNode response = new ObjectMapper().createObjectNode().put( "shady", "shade" );
      return Flux.just( response );
    }

返回到客户机总是ok，但是服务器报告通量的内容为空。如果我将(@requestbody flux items 更改为(@requestbody JsonNode items)，那么有效负载将很好地接收到。WebClient日志似乎表明它已经将数据写入线路并处理了响应。但是，主体似乎是空的

Reactor.core.Exceptions$ErrorCallbackNotImplemented: org.springframework.web.server.ServerWebInputException: Response status 400 with reason "Request body is missing: public reactor.core.publisher.Flux<com.fasterxml.jackson.databind.JsonNode> com.talroo.rest.JobResource.loadJobs(reactor.core.publisher.Flux<com.fasterxml.jackson.databind.JsonNode>)"
Caused by: org.springframework.web.server.ServerWebInputException: 
Response status 400 with reason "Request body is missing: public 
reactor.core.publisher.Flux<com.fasterxml.jackson.databind.JsonNode>

我需要做什么才能将post的请求主体作为一个Flux来处理？

Answer 1

首先，我不认为Spring正式支持直接从控制器读/写JacksonJsonNode实例。您的应用程序应该请求一个域对象或类似map 之类的东西。

在Jackson的模型中，jsonNode表示JSON树中的任何节点--由于它是一棵树，您可以期望得到一个flux节点，但显然能够得到根节点--这解释了您看到的行为。

所以我认为您的应用程序应该更多地依赖于更高级的类，并让Jackson为您反序列化它们。

注意，您的控制器实现也违反了一些规则：

• 不应在返回反应类型的方法中调用阻塞运算符，如block
（您的控制器不是破坏此类型，而是关闭此类型）
• 您不应破坏反应流水线，将请求的读取和响应的写入解耦；在您的控制器有机会读取整个请求之前，HTTP交换可能会被关闭。调用subscribe就是这样做的。