RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
Observable<Data> signal1 = hotSignal.filter(data -> data.id == 1).startWith(emptyData);
Observable<Data> signal2 = hotSignal.filter(data -> data.id == 2).startWith(emptyData);
Observable<Data> signal3 = hotSignal.filter(data -> data.id == 3).startWith(emptyData);
Observable<String> latestData = Observable.combineLatest(
signal1, signal2, signal3,
(data1, data2, data3) -> mergeData(data1, data2, data3)
);
latestData.sample(1, TimeUnit.SECONDS).subscribe(mergedData -> System.out.println(mergedData));
here is a gist to demonstrate: https://gist.github.com/weefbellington/952414cb7076976bfd49
Observable first = Observable.create(new Observable.OnSubscribe<Object>() {
@Override
public void call(Subscriber<? super Object> subscriber) {
subscriber.add(Subscriptions.create(() -> {
log(“first unsubscribe”);
}));
}
}).doOnUnsubscribe(() -> log(“first”));
Observable second = Observable.create(…).doOnUnsubscribe(() -> log(“second”));
Observable result = first.mergeWith(second).doOnUnsubscribe(() -> loge(“result”));
Subscription subscription = result.subscribe(…);
//…
subscription.unsubscribe();
but now I can't delete it
disregard the above comment
@artem-zinnatullin you're right, but if the number of streams is finite and the IDs are known,
combineLatest is a pretty natural approach
final Func0<HashMap<Integer, String>> initialState = () -> new HashMap<>();
final HashMap<Integer, String> output = new HashMap<>();
source.window(1, TimeUnit.SECONDS)
.flatMap(observable -> observable.collect(
initialState,
(mostRecent, data) -> mostRecent.put(data.id, data.content))
)
.map(input -> {
output.putAll(input);
return output;
}).subscribe(reduced -> System.out.println(output));
^ original gist updated: https://gist.github.com/weefbellington/952414cb7076976bfd49
that final
.map is kind of kludgy, I'm pondering a better way to produce the output dictionary
Also i'm not sure what's going on in the final map. You seem to add back all previous data (in case some id's where unused last second). Is that truly by the spec? If so, i would think there is no need to use window, just
scan the original source instead of collect, and sample the result.
Also
reduced == output, so you can just used reduced in subscribe.
It doesn't look like
reducedis the same as output to me, either. The output of each window is a Map. The final step is merging the map emitted by the current window with the map from the previous window.
@weefbellington
The map ends with
scan emit all intermediate values also. Not sure if that bode well if you are emitting the same mutable datatype each time.The map ends with
return output;. If i understand that correctly that means that in .subscribe(reduced -> ...) , reduced is the output variable returned by map.
@Dorus thanks for pointing that out!
obviously it would be toBlocking under the covers
There are ways to use it ofcourse, but it's never "natural".
@Dorus updated the previous gist to use scan, thanks for the input:
source.scan(new HashMap<>(), (output, data) -> {
output.put(data.id, data.content);
return output;
}).sample(1, TimeUnit.SECONDS).subscribe(output -> System.out.println(output));
@pakoito @artem-zinnatullin ^ in case you are still interested in this puzzle :)
@weefbellington The only thing to keep in mind is that
output.put has side effects, and you are emitting the same collection every time, since sample might allows the previous operators to run concurrent, output might receive new put calls during the onNext call in subscribe. Might be a good idea to use ConcurrentHashMap or clone the collection somewhere.
Your previous example only updated the collection once every second, so this was less problematic.
I went through the second example and it seems to fit the bill. I can also add an extra signal to have an invalidation policy for data that's not fresh :)
like every 5 seconds add a signal that vacates those positions whose timestamped value is older than 5 seconds
@Dorus good point. I might be a little worried about the efficiency of cloning a new HashMap with each emission. I know that languages like Haskell that don't have mutable data structures are really good at creating a lot of them at once and then recollecting the memory, but I don't know how Java would handle it. So I'd probably go with the ConcurrentHashMap
Even worse on Dalvick/ART tho.
me or you
Sharing is Cheap
If you are certain that an object will never change, then sharing said object becomes a simple matter of providing a reference to it. In Java to do so often requires a lot of defensive copying. Along this vein, because we can freely share references for immutable objects we can likewise intern them for free.