在 Reactor Flux 中，onCancel 调用不会返回到调用方法答案

【问题标题】：In Reactor Flux an onCancel call is not returned to calling method在 Reactor Flux 中，onCancel 调用不会返回到调用方法
【发布时间】：2021-03-31 13:21:19
【问题描述】：

我有一组方法可以处理Flux<String> 所以

@GetMapping(value = "/web/{s}")
public Flux<String> getWeb(@PathVariable String s) {
    return myService.getFlux(s);
}

在service 类中：

public Flux<String> getFlux(String json) throws Exception {
    return handleRequest(json);
}

private Flux<String> handleRequest(String json) throws Exception {
    Many<String> sink = Sinks.many().multicast().directBestEffort();

    ... do stuff with sink ... 

    /* Handle flux completion, termination, error */
    Flux<String> webFlux = sink.asFlux()
            .doOnTerminate(() -> { disposeSink(id); })
            .doOnComplete(() -> { completeSink(id); })
            .doAfterTerminate(() -> { terminateSink(id); }) 
            .doOnCancel(() -> { cancelSink(id); })

    return webFlux;
}

当我让客户在完成之前缩短、取消、终止他们的请求时，doOnCancel 调用仅在它附加到初始调用方法时才有效，即

@GetMapping(value = "/web/{s}")
public Flux<String> getWeb(@PathVariable String s) {
    return myService.getFlux(s).doOnCancel(() -> Log.Info("Cancelled"));
}

这并不是我真正想写的地方，尽管考虑到它是面向客户的调用，这很有意义。为什么service 类中的doOnCancel 钩子没有返回给调用者？传递回调用方法的通量是一种发布/订阅自身的通量吗？！！

【问题讨论】：

标签： spring-webflux project-reactor

【解决方案1】：

这有点奇怪，但我不得不像这样直接返回带有钩子的sink.asFlux()：

private Flux<String> handleRequest(String json) throws Exception {
    Many<String> sink = Sinks.many().multicast().directBestEffort();

    ... do stuff with sink ... 

    /* Handle flux completion, termination, error */
    return sink.asFlux()
        .doOnTerminate(() -> { disposeSink(id); })
        .doOnComplete(() -> { completeSink(id); })
        .doAfterTerminate(() -> { terminateSink(id); }) 
        .doOnCancel(() -> { cancelSink(id); });
}

然后在面向客户端的类上，当客户端随机取消请求时，一切正常。我还把log() 放在那个类中

@GetMapping(value = "/web/{s}")
public Flux<String> getWeb(@PathVariable String s) {
    return myService.getFlux(s).log();
}

所以基本上没有直接的Flux 初始化，只需转换一次sink.asFlux()，然后将return 备份到堆栈中。

【讨论】：