In Reactor Sinks.Many() what is equivalent to EmitterProcessor onCancel

user_1964277 0
Viewed 3689

Per here I used to have code

EmitterProcessor<String> emitter = EmitterProcessor.create();
FluxSink<String> sink = emitter.sink(FluxSink.OverflowStrategy.LATEST);

sink.onCancel(() -> {
  cancelSink(id, request);
});

and when for example with rSocket a browser opened a session and asked for some data, calling the EmitterProcessor when a client shut down their browser the publisher like

Flux<String> out = Flux
    .from(emitter
    .log(log.getName()));

would know that the Flux subscriber was cancelled (when a browser was closed) and that would call the onCancel handle.

With Sinks.Many() I have implemented

Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();

sink.asFlux().doOnCancel(() -> {
    cancelSink(id, request);
});

Flux<String> out = Flux
    .from(sink.asFlux()
    .log(log.getName()));

and the strings are published via a flux to the browser, but when the client closes the session there is no longer the onCancel to handle some tidying up.

It looks like this was discussed here and also here but I don't understand the solutions. What is it please?

project-reactor
1 Answers

sink.asFlux().doOnCancel(...) and sink.asFlux() are two different instances. You're not reusing the one where you have set up cancel handling logic, and that is why you don't observe the cancelSink cleanup on your out variable.

Do something more like:

Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();

Flux<String> fluxWithCancelSupport = sink.asFlux().doOnCancel(() -> {
    cancelSink(id, request);
});

Flux<String> out = fluxWithCancelSupport
    .log(log.getName()));

(PS: you don't need the Flux.from(sink.asFlux()) since the later already gives you a Flux).

user_11134860
Related