标题翻译
How to create a Flux including historical and future items?
问题
我有一个应用程序，它接收事件（使用Spring应用程序监听器），我想创建一个Flux来表示一些历史事件（可能从持久性存储中读取），以及一个无限的流式传入事件。
目标是提供一个公共方法，返回所有已经被观察到的事件的Flux，即初始历史事件的串联，所有已观察到的事件以及所有未来事件。以下服务应该提供以stream()方法形式的API。
@Servicepublic class EventService implements ApplicationListener<Event> {private Flux<Integer> eventStream = getHistoricalEvents();private Flux<Integer> getHistoricalEvents() {return Flux.just(1,2,3,4,5);}@Overridepublic void onApplicationEvent(Event event) {eventStream = eventStream.concatWithValues(event.hashCode());}public Flux<Integer> stream(){return eventStream;}}
这样可以实现前两个目标，即客户端接收到初始的一组历史事件以及迄今已观察到的所有事件。
如何添加未来事件？理想情况下，我正在寻找一个能够在语法上实现所有这些的解决方案，因为我对响应式编程还不熟悉。
英文翻译
I have an application that receives events (using Spring Application Listeners) and I want to create a Flux of some historical events (that might be read from persistence) and an infinite stream of incoming events.
The goal is to provide a public method that returns a Flux of all events that have ever been observed, i.e., the concatenation of the initial historical events, all events observed and all future events. The following service is supposed to provide the API in terms of the method stream().
@Servicepublic class EventService implements ApplicationListener<Event>{private Flux<Integer> eventStream = getHistoricalEvents();private Flux<Integer> getHistoricalEvents() {return Flux.just(1,2,3,4,5);}@Overridepublic void onApplicationEvent(Event event) {eventStream = eventStream.concatWithValues(event.hashCode());}public Flux<Integer> stream(){return eventStream;}}
This works to achieve the first two goals, i.e., clients receive the initial set of historical events and all events that have been observed so far.
How can I add future events? Ideally, I am looking for a solution that achieves all that idiomatically as I am new to reactive programming.
答案1
得分: 0
你已经找到了如何与现有的持久化历史事件进行concat的方法。至于所有未来事件，听起来你很可能需要Flux.cache()：
> 将这个Flux转变为热源，并缓存最近发出的信号以供进一步订阅者使用。将保留一个无限量的onNext信号。完成和错误也将被重新播放。
英文翻译
You've already worked out how to concat with existing persisted, historical events. As for all future events, it sounds very much like you need Flux.cache():
> Turn this Flux into a hot source and cache last emitted signals for further Subscriber. Will retain an unbounded volume of onNext signals. Completion and Error will also be replayed.
专注分享java语言的经验与见解，让所有开发者获益！
评论