Reactor/WebFlux实现一个反应式http新闻自动收录器

我有一个非常简单的请求,但我无法在不泄漏资源的情况下完成它。

我想返回类型为的响应 application/stream+json ,以某人发布的新闻事件为特色。我不想使用Websockets,不是因为我不喜欢它们,我只是想知道如何使用流。

为此,我需要返回 Flux<News> 从我的restcontroller,只要有人发布任何消息,就会不断收到消息。

我的尝试是创建一个发布者:

public class UpdatePublisher<T> implements Publisher<T> {

    private List<Subscriber<? super T>> subscribers = new ArrayList<>();

    @Override
    public void subscribe(Subscriber<? super T> s) {
        subscribers.add(s);
    }

    public void pushUpdate(T message) {
        subscribers.forEach(s -> s.onNext(message));
    }

}

还有一个简单的新闻对象:

public class News {
    String message;
    // Constructor, getters, some properties omitted for readability...
}

和发布新闻的端点分别获取新闻流

// ...

private UpdatePublisher<String> updatePublisher = new UpdatePublisher<>();

@GetMapping(value = "/news/ticker", produces = "application/stream+json")
public Flux<News> getUpdateStream() {
     return Flux.from(updatePublisher).map(News::new);
}

@PutMapping("/news")
public void putNews(@RequestBody News news) {
    updatePublisher.pushUpdate(news.getMessage());
}

这是可行的,但我无法取消订阅或再次访问任何给定的订阅-因此,一旦客户端断开连接 updatePublisher 只会继续推进越来越多的死区通道,因为我无法呼叫 onCompleted() 订阅上的处理程序。

TL;DL:

可以将消息从不同的线程推送到可能的无止境流量上,并且仍然可以根据需要终止流量,而不依赖于对等异常重置或类似的内容吗?