### Question:
Part2CreationTransformationTermination { @Complexity(HARD) public static Flux<IceCreamBall> fillIceCreamWaffleBowl( Flux<IceCreamType> clientPreferences, Flux<IceCreamBall> vanillaIceCreamStream, Flux<IceCreamBall> chocolateIceCreamStream ) { throw new RuntimeException("Not implemented yet"); }  @Complexity(EASY) static Mono<List<String>> collectAllItemsToList(Flux<String> source); @Complexity(EASY) static String lastElementFromSource(Flux<String> source); @Complexity(EASY) static Publisher<String> mergeSeveralSources(Publisher<String>... sources); @Complexity(EASY) static Publisher<String> fromFirstEmitted(Publisher<String>... sources); @Complexity(EASY) static Publisher<GroupedFlux<Character, String>> groupWordsByFirstLatter(Flux<String> words); @Complexity(MEDIUM) static Mono<String> executeLazyTerminationOperationAndSendHello(Flux<String> source); @Complexity(MEDIUM) static Publisher<String> zipSeveralSources(Publisher<String> prefix,
                                                      Publisher<String> word,
                                                      Publisher<String> suffix); @Complexity(HARD) static Publisher<String> combineSeveralSources(Publisher<String> prefix,
                                                          Publisher<String> word,
                                                          Publisher<String> suffix); @Complexity(HARD) static Flux<IceCreamBall> fillIceCreamWaffleBowl(
            Flux<IceCreamType> clientPreferences,
            Flux<IceCreamBall> vanillaIceCreamStream,
            Flux<IceCreamBall> chocolateIceCreamStream
    );  }### Answer:
@Test public void switchBetweenSourcesTest() { StepVerifier .withVirtualTime(() -> fillIceCreamWaffleBowl( Flux.just(IceCreamType.VANILLA, IceCreamType.VANILLA, IceCreamType.CHOCOLATE, IceCreamType.VANILLA, IceCreamType.CHOCOLATE, IceCreamType.CHOCOLATE).delayElements(Duration.ofSeconds(1)), Flux.interval(Duration.ofMillis(500)).map(i -> ball("A" + i)).onBackpressureDrop().publish(1).autoConnect(0), Flux.interval(Duration.ofMillis(200)).map(i -> ball("B" + i)).onBackpressureDrop().publish(1).autoConnect(0) ).map(Object::toString)) .expectSubscription() .expectNoEvent(Duration.ofSeconds(1)) .thenAwait(Duration.ofSeconds(1)) .expectNext("A1", "A2") .thenAwait(Duration.ofSeconds(1)) .expectNext("A3", "A4") .thenAwait(Duration.ofSeconds(1)) .expectNext("B14", "B15", "B16", "B17", "B18") .thenAwait(Duration.ofSeconds(1)) .expectNext("A5", "A7", "A8") .thenAwait(Duration.ofSeconds(1)) .expectNext("B19", "B24", "B25", "B26", "B27", "B28") .thenAwait(Duration.ofSeconds(1)) .expectNext("B29", "B30", "B31", "B32", "B33") .thenCancel() .verify(); }