在 Reactor Sinks.Many() 中相当于 EmitterProcessor onCancel

Question

每here我曾经有代码

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

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

例如，当使用rSocket 时，浏览器打开会话并请求一些数据，当客户端关闭浏览器时调用EmitterProcessor publisher 之类的

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

会知道 Flux 订阅者被取消（当浏览器关闭时），并且会调用 onCancel 句柄。

Sinks.Many() 我已经实现了

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

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

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

字符串通过通量发布到浏览器，但是当客户端关闭会话时，不再有 onCancel 来处理一些整理工作。

看起来这是discussed here 和here，但我不明白解决方案。请问是什么？

Answer 1

sink.asFlux().doOnCancel(...) 和 sink.asFlux() 是两个不同的实例。您没有重复使用已设置取消处理逻辑的那个，这就是为什么您没有观察到 cancelSink 对 out 变量的清理。

做一些类似的事情：

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

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

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

（PS：你不需要Flux.from(sink.asFlux())，因为后者已经给了你Flux）。