WebFlux - 忽略 'cancel' 信号

Question

试图直接从 reactor 的文档和 spring webflux 文档中获得其他 SO 问答的答案，但不知何故我仍然不确定如何实现这个:(

假设我有如下简单的控制器：

@RestController
class DemoController {

    @GetMapping("/demo/{text}")
    public Flux<String> getDemo(@PathVariable String text) {
        return Flux.fromArray(text.split(""))
            .map(String::toUpperCase)
            .delayElements(Duration.ofSeconds(1L))
            .doOnNext(s -> System.out.println("S: " + s + " -> " + LocalDateTime.now().getSecond()))
            .doOnCancel(() -> System.out.println("Cancelled"));
    }
}

并且我希望处理忽略“取消”信号，该信号通常在客户端断开连接时出现。试图通过 nginx 的 proxy_ignore_client_abort on; 实现这一点，但无法正常工作。

我可以 .subscribe(...) 到一个自定义订阅者，在示例中忽略 cancel 信号，但我看到处理执行了两次 - 这很明显并且没有预料到。

另一种选择是，我可以 extends Flux<T> 并直接覆盖：

@Override
public void cancel() {
    System.out.println("Cancelled / ignored");
}

但这在某种程度上看起来更像是黑客攻击。

总结

有人可以建议，运行一些反应管道的“正确”方式是什么，通过控制器触发，可以忽略取消信号？

Answer 1

当前的 webflux 行为是完美的。但是，您的要求似乎有所不同，即使订阅者不再对收听数据感兴趣，您也总是希望发出。在这种情况下，只需将源设置为 HOT。

@RestController
class DemoController {

    @GetMapping("/demo/{text}")
    public Flux<String> getDemo(@PathVariable String text) {
        return Flux.fromArray(text.split(""))
                .map(String::toUpperCase)
                .delayElements(Duration.ofSeconds(1L))
                .doOnNext(s -> System.out.println("S: " + s + " -> " + LocalDateTime.now().getSecond()))
                .doOnCancel(() -> System.out.println("Cancelled"))
                .cache(); // hot source
    }
}

WebFlux - 忽略 'cancel' 信号

WebFlux - ignore 'cancel' signal

reactive-programming

project-reactor

spring-webflux

总结