Styjun

In Spring Boot 2 with Reactor, I am attempting to merge two Flux hot sources. However, the merge only ever seems to report the first of the two Flux parameters in merge. How do I get the merge to recognize the second Flux.

In the example below, the System.err in B-2 doesn't even print when outgoing1a is the first parameter. If I make outgoing2 the first, then A-2 doesn't print.

Below is the full example;

package com.example.demo;

import java.time.Duration;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class Weather 
String city;
Integer temperature;

public Weather(String city, Integer temperature) 
 this.city = city;
 this.temperature = temperature;


@Override
public String toString() 
 return "Weather [city=" + city + ", temperature=" + temperature + "]";


public static void main(String[] args) 

 BlockingQueue<Weather> queue = new LinkedBlockingQueue<>();
 BlockingQueue<Weather> queue2 = new LinkedBlockingQueue<>();

 // Assume Spring @Repository "A-1"
 new Thread(() -> 
 for (int d = 1; d < 1000; d += 1) 
 for (String s: new String[] "LDN", "NYC", "PAR", "ZUR") 
 queue.add(new Weather(s, d));
 try Thread.sleep(250); catch (InterruptedException e) 
 
 
 ).start(); 

 // Assume Spring @Repository "B-1"
 new Thread(() -> 
 for (int d = 1; d < 1000; d += 1) 
 for (String s: new String[] "MOS", "TLV") 
 queue2.add(new Weather(s, d));
 try Thread.sleep(1000); catch (InterruptedException e) 
 
 
 ).start();

 // Assume Spring @Service "A-2" = real-time LDN, NYC, PAR, ZUR
 Flux<Weather> outgoing1 = Flux.<Weather>create(
 sink -> 
 for (int i = 0; i < 1000; i++) 
 try 
 sink.next(queue.take());
 System.err.println("1 " + queue.size());
 catch (InterruptedException e) 
 e.printStackTrace();
 
 
 sink.complete();
 
 ).publishOn(Schedulers.newSingle("outgoing-1"));

 // Assume Spring @Service "B-2" = real-time MOS, TLV
 Flux<Weather> outgoing2 = Flux.<Weather>create(
 sink -> 
 for (int i = 0; i < 1000; i++) 
 try 
 sink.next(queue2.take());
 System.err.println("2 " + queue2.size());
 catch (InterruptedException e) 
 e.printStackTrace();
 
 
 sink.complete();
 
 ).publishOn(Schedulers.newSingle("outgoing-2"));

 // Assume Spring @Service "A-3" = 5 second summary of LDN, NYC, PAR, ZUR
 Flux<Weather> outgoing1a = Flux.from(outgoing1) 
 .groupBy(c -> c.city)
 .flatMap(g -> g
 .sample(Duration.ofSeconds(5))
 )
 .log("C");

 // Assume Spring @Service "C" - merges "A-3" and "B-2"
 // only prints outgoing1a
 Flux.merge(outgoing1a, outgoing2).subscribe(System.out::println); 

 // only prints outgoing2
 //Flux.merge(outgoing2, outgoing1a).subscribe(System.out::println); 

There are a few things at play here.

1. Note the following recommendation of the .merge operator...

Note that merge is tailored to work with asynchronous sources or finite sources. When dealing with an infinite source that doesn't already publish on a dedicated Scheduler, you must isolate that source in its own Scheduler, as merge would otherwise attempt to drain it before subscribing to another source.

2. Your outbound Fluxes use .publishOn, but that only affects operators chained after the .publishOn operator. i.e. It doesn't affect anything before the .publishOn. Specifically, it does not affect the thread on which the code in lambda passed to Flux.create executes. You can see this if you add .log() before the .publishOn in each of the outbound Fluxes.




3. Your lambda passed to Flux.create calls a blocking method (queue.take).

Since you call subscribe(...) on the merged Flux in the main thread, your lambda passed to Flux.create executes in the main thread, and blocks it.

The easiest fix is to use .subscribeOn instead of .publishOn so that your code in the lambda passed to Flux.create operates on a different thread (other than main). This will prevent the main thread from blocking, and allow the merged output from both outbound streams to be interleaved.