Flux.switchOnNext variant that switches when the next publisher _emits_ rather than when it _is emitted_

user_592139 0
Viewed 123

Reactor has the switchOnNext operator which mirrors a sequence of publishers, cancelling the previous subscription whenever a new publisher becomes available:

For my use case I need a variation on this theme, where instead of cancelling the first publisher before subscribing to the next, I continue to mirror the sequence of publisher 1 until the point when publisher 2 emits its first item, and only then make the switch, as in this marble diagram (for anyone who finds this question later this is not a diagram of an existing operator from the Reactor docs, it's one I've just sketched myself):

I appreciate that in the general case this could potentially involve the operator maintaining an unbounded number of subscriptions waiting for any one of them to emit before cancelling the others, but for my use case I know that the initial flux-of-fluxes is finite (so I don't necessarily need the fully general publisher-of-publishers solution, something that works for a finite list of N publishers would be sufficient).

Can anyone see a clever combination of the existing operators that would implement this behaviour or do I need to write it from first principles?

project-reactor
1 Answers

Interesting problem! I think something like this might work:

@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));
}

The important part is that we convert the sequences into hot publishers (meaning that resubscribing them doesn't cause another subscription, rather than we share the initial subscription). Then we use scan to emit a tuple containing the previous and the next, and finally we just use a regular switchMap to observe both (note how the first will stop when the second emits, due to takeUntilOther).

Note that the grace period is important because switchMap will first cancel and then subscribe the next, so if there isn't any grace period it would cause the current hot publisher to fully stop and start from scratch, which is not what we want.

user_25971430
Related