WebFlux - 忽略“取消”信号

Question

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

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

@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
    }
}