reactor/reactor-core

Hello Reactor team.

Static composition can be done with combine() method, but I want a way to manage flux composition dynamically.
I would like to use it for a IoT project where the UI can listen one or more sensor depending what the user want.
The result flux is given to a web socket.
Subscribe or unsubscribe to a sensor flux is done with a rest service.

I would like to know if DirectProcessor is a good way to create a Flux multiplexer from others Flux ?
I create a SimulatorModule to test how I can combine different flux in one flux with a DirectProcessor.
It seams to work fine (in reality a sensor produce 1 data every 3 minutes - backpressure is not a pb now).

I also tried with FluxSink from DirectProcessor#sink() instead DirectProcessor#onNext() but it seams to be a way less performant !
What are the difference between these two way to feed the DirectProcessos ?

I tried with EmitterProcessor, it seams to be less performance, is it the case ?
Is EmitterProcessor,the only way to manage backpressure ?

public class SimulatorModule {

    private DirectProcessor<Triple> dp;
    private HashMap<String, Disposable> producers = new HashMap<>();

    public SimulatorModule() {
        dp = DirectProcessor.create();
        long start = Instant.now().toEpochMilli();
    }

    public void simulate(String name, long interval) {
        removeSimulator(name);
        Disposable disposable = Flux.interval(Duration.ofMillis(interval))
                .map(e -> new Triple(name, System.currentTimeMillis(), 1.0))
                .subscribe(e -> dp.onNext(e));
        producers.put(name, disposable);
    }

    public void removeSimulator(String name) {
        Disposable disposable1 = producers.get(name);
        if (disposable1 != null) {
            disposable1.dispose();
            producers.remove(name);
        }
    }

    public Flux<Triple> flux() {
        return dp;
    }

    public static class Triple {
        public final String name;
        public final long ts;
        public final double value;

        public Triple(String name, long ts, double value) {
            this.name = name;
            this.ts = ts;
            this.value = value;
        }
    }
}

I call it with :

        simulatorModule.flux()
                .filter(e -> e.name.equals("F1"))
                .subscribe(e -> System.out.println("1 " + e.name +  " -  " +Thread.currentThread().getName())); 
        simulatorModule.flux()
                .filter(e -> e.name.equals("F2"))
                .subscribe(e -> System.out.println("2 " + e.name +  " -  " +Thread.currentThread().getName())); 
        simulatorModule.simulate("F1", 1000);
        simulatorModule.simulate("F2", 1200);

Thanks in advance for your help !