使用 application/stream+json 时 WebFlux 不发送数据

Question

我有一个响应式核心 WebClient 可以发布到给定的端点。有效负载是Job 对象的通量，内容类型是 application/stream+json Flux jobFlux = Flux.just(new Job());

Mono<JsonNode> response = localEP.post().uri( "/dev/job" )
    .contentType(MediaType.APPLICATION_STREAM_JSON)
    .body( BodyInserters.fromObject(jobFlux))
    .retrieve()
    .bodyToMono( JsonNode.class );

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

  @PostMapping(path = "/dev/job", consumes = MediaType.APPLICATION_STREAM_JSON_VALUE)
  @ResponseStatus( HttpStatus.CREATED )
 public Mono<Void> loadJobs (@RequestBody Flux<Job> jobs) {
    return this.repository.create(jobs); 
 }

域类Job在实例化新对象时创建和标识：

  public Job() {
    UUID guid = UUID.randomUUID();
    id = guid.toString();
    title = "Random String";
  }

存储库目前只是一个存根：

@Repository
public class DemoJobRepository implements ReactiveRepository<Job> {
   private static Logger logger = LoggerFactory.getLogger(DemoJobRepository.class);
   private final List<Job> jobs = Lists.newArrayList();

@Override
public Mono<Void> create(Publisher<Job> jobStream) {
    return Flux.from(jobStream).doOnNext(jobs::add).then();
}

@Override
public Flux<Job> getAll() {
    return Flux.fromIterable(jobs);
}

@Override
public Mono<Job> findById(String id) {
    return null;
}
}

我没有看到客户端尝试发送请求正文。我们在客户端上调用了block 以获取结果，我看到客户端发送了请求，但是服务器端点总是看到一个空通量。请，非常感谢任何帮助。

Answer 1

对于响应式类型，nothing happens until you subscribe - 构建响应式管道不会执行它应该执行的操作，只有订阅它才会启动流程。

有几种订阅方式：

• 调用任何subscribe 变体。其中许多将 lambdas 作为参数，在处理完成或以错误结束时执行。使用空变量有点冒险，因为它只是启动执行，但你没有得到任何回调。从技术上讲，没有什么在等待它，因此 JVM 可以在处理完成之前退出

• 调用任何block/collect 方法。这不仅订阅而且返回预期值。

  这两种选择都不应该在返回反应类型的方法中完成，否则这将导致您的应用程序出现严重问题。