### Question:
Part6HotTransformationAndProcession { @Complexity(MEDIUM) public static Flux<String> processEachSubscriberOnSeparateThread(Flux<String> coldSource) { throw new RuntimeException("Not implemented"); }  @Complexity(MEDIUM) static Publisher<String> transformToHot(Flux<String> coldSource); @Complexity(MEDIUM) static Publisher<String> replayLast3ElementsInHotFashion(Flux<String> coldSource); @Complexity(MEDIUM) static Publisher<String> transformToHotUsingProcessor(Flux<String> coldSource); @Complexity(MEDIUM) static Flux<String> processEachSubscriberOnSeparateThread(Flux<String> coldSource);  }### Answer:
@Test public void processEachSubscriberOnSeparateThreadTest() { UnicastProcessor<String> source = UnicastProcessor.create(); ReplayProcessor<String> consumer1 = ReplayProcessor.create(10); ReplayProcessor<String> consumer2 = ReplayProcessor.create(10); Thread[] forConsumers = new Thread[2]; Flux<String> publisher = processEachSubscriberOnSeparateThread(source); publisher .doOnComplete(() -> forConsumers[0] = Thread.currentThread()) .subscribe(consumer1); source.onNext("A"); source.onNext("B"); source.onNext("C"); publisher .doOnComplete(() -> forConsumers[1] = Thread.currentThread()) .subscribe(consumer2); source.onNext("D"); source.onNext("E"); source.onNext("F"); source.onComplete(); StepVerifier.create(consumer1) .expectSubscription() .expectNext("A", "B", "C", "D", "E", "F") .verifyComplete(); StepVerifier.create(consumer2) .expectSubscription() .expectNext("D", "E", "F") .verifyComplete(); Assert.assertTrue( "Expected execution on different Threads", !forConsumers[0].equals(forConsumers[1]) ); }