使用 application/stream+json 时 WebFlux 不发送数据 [英] WebFlux not sending data when using application/stream+json

问题描述

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

I have a reactive core WebClient to post to a given endpoint. The payload is a flux of Job objects and the content-type is 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.

On the server end I have tried both a Spring Controller style and Spring Web Reactive FunctionHandler to process the payload of the above call with a payload that is a 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在实例化新对象时创建和id:

The domain class Job creates and id when a new object is instantiated:

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

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

The repository currently is just a stub:

@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 来获取结果，我看到客户端发送了请求，但是服务器端点总是看到一个空的流量.非常感谢您的帮助.

I don't see the client attempting to send the requestbody. We I called block on the client to get the result, I see the client send the request, however the server endpoint always sees an empty flux. Please, any help is very appreciate.

推荐答案

使用响应式类型，在您订阅之前不会发生任何事情 - 构建反应式管道不会执行它应该做的事情，只有订阅它才会启动该过程.

With reactive types, nothing happens until you subscribe - building the reactive pipeline will not execute what's it's supposed to do, only subscribing to it will start the process.

有几种订阅方式:

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

• calling any of the subscribe variants. Many of them take lambdas as parameters that are executed when the processing is done, or ended with an error. Using the empty variant is a bit risky because it just launches the execution but you get no callback whatsoever. Technically, nothing is waiting on it so the JVM could exit before the processing is done

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

calling any of the block/collect methods. This not only subscribes but also returns the expected value(s).

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

Both of those choices should never be done within a method that returns a reactive type, otherwise this will lead to serious problems in your application.

这篇关于使用 application/stream+json 时 WebFlux 不发送数据的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！