Split a Flux into a Mono head and Flux tail

I want to split a Flux into two parts: A Mono for the 1st element (the head), and a Flux for everything else (the tail).
The base Flux should not be re-subscribed to in the course of this process.

Example of what DOESN'T work:

final Flux<Integer> baseFlux = Flux.range(0, 3).log(); final Mono<Integer> head = baseFlux.next(); final Flux<Integer> tail = baseFlux.skip(1L); assertThat(head.block()).isEqualTo(0); assertThat(tail.collectList().block()).isEqualTo(Arrays.asList(1, 2)); 

The log for this looks something like the following, and as you can see, the base Flux will be re-subscribed to twice:

[main] DEBUG reactor.util.Loggers$LoggerFactory - Using Slf4j logging framework [main] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription) [main] INFO reactor.Flux.Range.1 - | request(unbounded) [main] INFO reactor.Flux.Range.1 - | onNext(0) [main] INFO reactor.Flux.Range.1 - | cancel() [main] INFO reactor.Flux.Range.1 - | onSubscribe([Synchronous Fuseable] FluxRange.RangeSubscription) [main] INFO reactor.Flux.Range.1 - | request(unbounded) [main] INFO reactor.Flux.Range.1 - | onNext(0) [main] INFO reactor.Flux.Range.1 - | onNext(1) [main] INFO reactor.Flux.Range.1 - | onNext(2) [main] INFO reactor.Flux.Range.1 - | onComplete() [main] INFO reactor.Flux.Range.1 - | request(1) 

My actual case is that my base Flux contains the lines of a CSV file, with the first line being the header of the file, which is needed to parse all subsequent lines. The base Flux can only be subscribed to once, as it is based on an InputStream

The only somewhat related resource I found for this is this question, but I found this to be somewhat unfitting for my needs.

1 Answer

Thanks to a suggestion offered in the comments, I was able to devise the following solution:

final Flux<Integer> baseFlux = Flux.range(0, 3).log(); final Flux<? extends Tuple2<? extends Integer, Integer>> zipped = baseFlux .switchOnFirst((signal, flux) -> (signal.hasValue() ? Flux.zip(Flux.just(signal.get()).repeat(), flux.skip(1L)) : Flux.empty())); final List<? extends Tuple2<? extends Integer, Integer>> list = zipped.collectList().block(); assertThat(list.stream().map(Tuple2::getT1)).isEqualTo(Arrays.asList(0, 0)); assertThat(list.stream().map(Tuple2::getT2)).isEqualTo(Arrays.asList(1, 2)); 

It transforms the base Flux after the 1st element, by zipping this element repeated with the tail of the original flux. And it only subscribes to the baseFlux once.

I'm not sure this is the best solution, as it will create a lot of Tuple2 objects which will be GC'd eventually, compared to a solution which would have a stateful ("hot") flux based on the baseFlux, which keeps the original subscription alive.