通量switchOnNext变体，在下一个发布服务器发出而不是发出时进行切换_

Question

反应器具有SwitchOnNext运算符，它镜像一系列发布者，每当新的发布者可用时取消先前的订阅：

对于我的用例，我需要一个关于此主题的变体，在订阅下一个发布服务器之前，我不会取消第一个发布服务器，而是继续镜像发布服务器1的序列，直到发布服务器2发出其第一个项目，然后才进行切换，正如这张大理石图（对于后来发现这个问题的人来说，这不是Reactor文件中现有操作员的图，这是我自己刚刚画的）：

我理解，在一般情况下，这可能涉及运营商维护无限数量的订阅，等待其中任何一个发出，然后取消其他订阅，但对于我的用例，我知道通量的初始流量是有限的（因此我不一定需要完全通用的发布者解决方案，对于N个发布者的有限列表就足够了）。

有人能看到现有运营商的巧妙组合来实现这种行为吗？或者我需要从第一原理开始写吗？

Answer 1

有趣的问题！我认为这样做可能会奏效：

@Test
void switchOnNextEmit() {
  Duration gracePeriod = Duration.ofSeconds(2);

  Flux.concat(
          Mono.just(sequence("a", 1)),
          Mono.just(sequence("b", 3)).delaySubscription(Duration.ofSeconds(5)),
          Mono.just(sequence("c", 10)).delaySubscription(Duration.ofSeconds(10)))
      .map(seq -> seq.publish().refCount(1, gracePeriod))
      .scan(
          Tuples.of(Flux.<String>never(), Flux.<String>never()),
          (acc, next) -> Tuples.of(acc.getT2().takeUntilOther(next), next))
      .switchMap(t -> Flux.merge(t.getT1(), t.getT2()))
      .doOnNext(it -> System.out.println("Result: " + it))
      .then()
      .block();
}

private static Flux<String> sequence(String name, int interval) {
  return Flux.interval(Duration.ofSeconds(interval))
      .map(i -> name + i)
      .doOnSubscribe(__ -> System.out.println("Subscribe: " + name))
      .doOnCancel(() -> System.out.println("Cancel: " + name));
}

重要的是，我们将序列转换为热门发布服务器（这意味着重新订阅它们不会导致另一个订阅，而不是共享初始订阅）。然后，我们使用扫描来发出一个包含前一个和下一个的元组，最后我们只需使用常规的开关映射来观察这两个元组（请注意，由于takeUntilOther的原因，第一个元组在第二个元组发出时将如何停止）。

请注意，宽限期很重要，因为switchMap将首先取消，然后订阅下一个，因此如果没有任何宽限期，将导致当前热门发布服务器完全停止并从头开始，这不是我们想要的。