如何不依赖Reactor订阅时间

Question

我已经阅读了整个反应堆的文件，但我无法找到正确的模式，以解决以下问题。我有一个方法可以异步地执行某些操作。我以Flux的形式返回结果响应，消费者可以订阅它。

该方法有以下定义：

Flux<ResultMessage> sendRequest(RequestMessage message);

返回的通量是一个热通量，结果可以在任何给定的时间异步来。

sendRequest(message).subscribe(response->doSomethinWithResponse(response);

Flux<ResultMessage> sendRequest(RequestMessage message) {
   Flux<ResultMessage> result = incomingMessageStream
            .filter( resultMessage -> Objects.equals( resultMessage.getId(), message.getId() ) )
            .take( 2 );
// The message sending is done here...

    return result;
}

其中，incomingMessageStream是通过此通道的所有消息的flux。这个实现的一个问题是，消费者是在结果消息到来之后订阅的，它可能会错过其中的一些消息。

所以，我正在寻找的是一个解决方案，将允许消费者不依赖于订阅的时间。潜在的消费者可能根本不需要订阅产生的flux。我正在寻找一个通用的解决方案，但如果这是不可能的，您可以假设结果消息的数量不大于2。

Answer 1

过了一段时间,我创建了一个似乎有效的解决方案：

Flux<ResultMessage> sendRequest(RequestMessage message) {
  final int maxResponsesCount = 2;
  final Duration responseTimeout = Duration.ofSeconds( 10 );
  final Duration subscriptionTimeout = Duration.ofSeconds( 5 );

  // (1) 
  ConnectableFlux<ResultMessage> result = incomingMessageStream
      .ofType( ResultMessage.class )
      .filter( resultMessage ->Objects.equals(resultMessage.getId(), message.getId() ) )
      .take( maxResponsesCount )
      .timeout( responseTimeout )
      .replay( maxResponsesCount );
  Disposable connectionDisposable = result.connect();

  // (2)
  AtomicReference<Subscription> subscriptionForCancelSubscription = new AtomicReference<>();
  Mono.delay( subscriptionTimeout )
    .doOnSubscribe( subscriptionForCancelSubscription::set )
    .subscribe( x -> connectionDisposable.dispose() );

  // The message sending is done here...

  // (3)
  return result
    .doOnSubscribe(s ->subscriptionForCancelSubscription.get().cancel())
    .doFinally( signalType -> connectionDisposable.dispose() );
}

我正在使用一个ConnectableFlux立即连接到流，而不需要订阅，它被设置为使用reply（）方法存储所有消息，因此后面的任何订阅者都不会错过响应消息（1）。

可以执行的路径很少：

null

2.1.未返回任何消息

• 解决方案-为获取响应设置了超时(.timeout(responseTimeout))。在此之后，.dofinally(..)清理资源（1）（3）。

2.2.已返回一些响应消息

• 解决方案-与2.1相同。

null