Publish, Replay and Cache Operators

[benjchristensen]

(Redo of #260 after merging)

Added basic Publish (#15) and Replay (#71) operators to Observable. I have not done any of the overloads (particularly Replay which has 10+ in .Net.

I also added a new Cache operator as discussed by @johngmyers and I at #209.

Playing with Replay and ConnectableObservable it does not cater well to the typical use case of needing to just de-dupe calls (cache the responses) so this Cache operator can be thought of as an automatic version of Replay. It comes with the same risk as toList if used with infinite or very large sequences as you can not unsubscribe from it.

[jmhofer]

Still here (and twice below). :)

[cloudbees-pull-request-builder]

RxJava-pull-requests #134 FAILURE
Looks like there's a problem with this pull request

[mttkay]

Hi,

I noticed that calling Observable.replay() doesn't merely wrap the given observable into a ReplaySubject, but also combines this with OperationMulticast.

I have several questions around this.

1. Why do this? Aren't multicasting and caching/replaying both independent concepts in their own right?

2. When combined, it raises questions about how to deal with the (different) Subscription instances returned from replay().subscribe and replay().connect. Unfortunately there are no samples or unit/integration tests (that I'm aware of) which actually combine both operators, so looking at the tests didn't help. More precisely:

ConnectableObservable obs = sourceObservable.replay()
Subscription s1 = obs.subscribe(observer);
Subscription s2 = obs.connect();

s1 and s2 are entirely different subscriptions. Looking at the sources, s1 is a RepeatSubjectSubscription, which is put in an internal map and removed upon unsubscription. s1 however is an anonymous inner class instance in OperationMulticast which delegates to the subscription the sourceObservable would return normally.

Now, to which one do I hold on and unsubscribe from when I want to go through a disconnect/reconnect cycle? I actually tried a few combinations and none worked for me. I either ended up triggering the sourceObservable twice upon reconnection (which I don't want) or I end up not getting previous emissions replayed on my observer.

Any insight around this would be great.

[benjchristensen]

Have you read this? => http://northhorizon.net/2011/sharing-in-rx/ It helped me understand it quite a bit.

Here is a snippet from it:

Fortunately, we have Replay() for this. It literally has 15 overloads, which cover just about every permutation of windowing by count and/or time, and supplying an optional scheduler and/or selector. The parameterless call, however, just remembers everything. Replay is just like Publish in the sense that it also forwards a call to Multicast, but this time with a ReplaySubject.

The reason multicast is involved is because it gives a ConnectedObservable associated with a Subject. It supports multiple subscriptions to a single Observable which is exactly what replay and publish both want.

Anything with ConnectedObservable has the underlying source managed by the ConnectedObservable itself. Thus in your example it would be s2 which you then unsubscribe from when you want to shut down the source.

If you just want simple caching behavior then the cache operator may help since it does not require the ConnectedObservable interaction.

The ConnectedObservable behavior is important for use cases when the start/stop of the underlying sequence needs to be controlled independently of when child subscriptions subscribe/unsubscribe. The refcount operator is another example (we haven't yet implemented this one) of controlling this situation. I have found cache to be exactly what I need for most use cases (though we need to add overloads for time, count limits, etc) as I just want everything replayed as long as the reference is not garbage collected - but that is not how ConnectedObservable behaves.

Hope this helps.

[mttkay]

Thanks @benjchristensen, as always this was very helpful. I think what got me confused was the terminology. The fact that both replay() and cache() are implemented in terms of ReplaySubject is a little confusing at first (the name replay() clearly implies that items will be re-emitted, while cache() doesn't make that clear). The former wraps the ReplaySubject in a ConnectableObservable using OperationMulticast, while the latter simply forwards the subscription to ReplaySubject.

(Re)reading the article you posted (I did read it before, but that was before ReplaySubject had been released so it had slipped off my mind) and looking at the implementation, would you say this summary is correct:

• replay: Creates a connectable event stream where upon reconnection, events that had been in flight during the period of disconnection will be replayed onto the observer
• cache: Creates an event stream where upon subscription, events will be replayed onto the observer
• publish: Creates a connectable event stream where upon reconnection, new events will be received by the observer, but events that were in flight during the period of disconnection will not be replayed

[benjchristensen]

Sorry I lost this question in my backlog ...

Yes those summaries make sense to me. The cache one could be expanded further though:

Creates an event stream where upon subscription, all previously emitted events will be replayed onto the observer and future events if not completed.

[mttkay]

👍