Skip to content

Commit f17d3ff

Browse files
authored
1.x: expand and explain cache() in more detail (ReactiveX#4026)
1 parent 76517d9 commit f17d3ff

1 file changed

Lines changed: 61 additions & 19 deletions

File tree

‎src/main/java/rx/Observable.java‎

Lines changed: 61 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -4027,27 +4027,48 @@ public final <B> Observable<List<T>> buffer(Observable<B> boundary, int initialC
40274027
}
40284028

40294029
/**
4030-
* Caches the emissions from the source Observable and replays them in order to any subsequent Subscribers.
4031-
* This method has similar behavior to {@link #replay} except that this auto-subscribes to the source
4032-
* Observable rather than returning a {@link ConnectableObservable} for which you must call
4033-
* {@code connect} to activate the subscription.
4030+
* Returns an Observable that subscribes to this Observable lazily, caches all of its events
4031+
* and replays them, in the same order as received, to all the downstream subscribers.
40344032
* <p>
40354033
* <img width="640" height="410" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/cache.png" alt="">
40364034
* <p>
40374035
* This is useful when you want an Observable to cache responses and you can't control the
40384036
* subscribe/unsubscribe behavior of all the {@link Subscriber}s.
40394037
* <p>
4040-
* When you call {@code cache}, it does not yet subscribe to the source Observable and so does not yet
4041-
* begin caching items. This only happens when the first Subscriber calls the resulting Observable's
4042-
* {@code subscribe} method.
4038+
* The operator subscribes only when the first downstream subscriber subscribes and maintains
4039+
* a single subscription towards this Observable. In contrast, the operator family of {@link #replay()}
4040+
* that return a {@link ConnectableObservable} require an explicit call to {@link ConnectableObservable#connect()}.
40434041
* <p>
40444042
* <em>Note:</em> You sacrifice the ability to unsubscribe from the origin when you use the {@code cache}
40454043
* Observer so be careful not to use this Observer on Observables that emit an infinite or very large number
4046-
* of items that will use up memory.
4044+
* of items that will use up memory.
4045+
* A possible workaround is to apply `takeUntil` with a predicate or
4046+
* another source before (and perhaps after) the application of cache().
4047+
* <pre><code>
4048+
* AtomicBoolean shouldStop = new AtomicBoolean();
4049+
*
4050+
* source.takeUntil(v -&gt; shouldStop.get())
4051+
* .cache()
4052+
* .takeUntil(v -&gt; shouldStop.get())
4053+
* .subscribe(...);
4054+
* </code></pre>
4055+
* Since the operator doesn't allow clearing the cached values either, the possible workaround is
4056+
* to forget all references to it via {@link #onTerminateDetach()} applied along with the previous
4057+
* workaround:
4058+
* <pre><code>
4059+
* AtomicBoolean shouldStop = new AtomicBoolean();
4060+
*
4061+
* source.takeUntil(v -&gt; shouldStop.get())
4062+
* .onTerminateDetach()
4063+
* .cache()
4064+
* .takeUntil(v -&gt; shouldStop.get())
4065+
* .onTerminateDetach()
4066+
* .subscribe(...);
4067+
* </code></pre>
40474068
* <dl>
40484069
* <dt><b>Backpressure Support:</b></dt>
4049-
* <dd>This operator does not support upstream backpressure as it is purposefully requesting and caching
4050-
* everything emitted.</dd>
4070+
* <dd>The operator consumes this Observable in an unbounded fashion but respects the backpressure
4071+
* of each downstream Subscriber individually.</dd>
40514072
* <dt><b>Scheduler:</b></dt>
40524073
* <dd>{@code cache} does not operate by default on a particular {@link Scheduler}.</dd>
40534074
* </dl>
@@ -4070,27 +4091,48 @@ public final Observable<T> cache(int initialCapacity) {
40704091
}
40714092

40724093
/**
4073-
* Caches emissions from the source Observable and replays them in order to any subsequent Subscribers.
4074-
* This method has similar behavior to {@link #replay} except that this auto-subscribes to the source
4075-
* Observable rather than returning a {@link ConnectableObservable} for which you must call
4076-
* {@code connect} to activate the subscription.
4094+
* Returns an Observable that subscribes to this Observable lazily, caches all of its events
4095+
* and replays them, in the same order as received, to all the downstream subscribers.
40774096
* <p>
40784097
* <img width="640" height="410" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/cache.png" alt="">
40794098
* <p>
40804099
* This is useful when you want an Observable to cache responses and you can't control the
40814100
* subscribe/unsubscribe behavior of all the {@link Subscriber}s.
40824101
* <p>
4083-
* When you call {@code cache}, it does not yet subscribe to the source Observable and so does not yet
4084-
* begin caching items. This only happens when the first Subscriber calls the resulting Observable's
4085-
* {@code subscribe} method.
4102+
* The operator subscribes only when the first downstream subscriber subscribes and maintains
4103+
* a single subscription towards this Observable. In contrast, the operator family of {@link #replay()}
4104+
* that return a {@link ConnectableObservable} require an explicit call to {@link ConnectableObservable#connect()}.
40864105
* <p>
40874106
* <em>Note:</em> You sacrifice the ability to unsubscribe from the origin when you use the {@code cache}
40884107
* Observer so be careful not to use this Observer on Observables that emit an infinite or very large number
40894108
* of items that will use up memory.
4109+
* A possible workaround is to apply `takeUntil` with a predicate or
4110+
* another source before (and perhaps after) the application of cache().
4111+
* <pre><code>
4112+
* AtomicBoolean shouldStop = new AtomicBoolean();
4113+
*
4114+
* source.takeUntil(v -&gt; shouldStop.get())
4115+
* .cache()
4116+
* .takeUntil(v -&gt; shouldStop.get())
4117+
* .subscribe(...);
4118+
* </code></pre>
4119+
* Since the operator doesn't allow clearing the cached values either, the possible workaround is
4120+
* to forget all references to it via {@link #onTerminateDetach()} applied along with the previous
4121+
* workaround:
4122+
* <pre><code>
4123+
* AtomicBoolean shouldStop = new AtomicBoolean();
4124+
*
4125+
* source.takeUntil(v -&gt; shouldStop.get())
4126+
* .onTerminateDetach()
4127+
* .cache()
4128+
* .takeUntil(v -&gt; shouldStop.get())
4129+
* .onTerminateDetach()
4130+
* .subscribe(...);
4131+
* </code></pre>
40904132
* <dl>
40914133
* <dt><b>Backpressure Support:</b></dt>
4092-
* <dd>This operator does not support upstream backpressure as it is purposefully requesting and caching
4093-
* everything emitted.</dd>
4134+
* <dd>The operator consumes this Observable in an unbounded fashion but respects the backpressure
4135+
* of each downstream Subscriber individually.</dd>
40944136
* <dt><b>Scheduler:</b></dt>
40954137
* <dd>{@code cache} does not operate by default on a particular {@link Scheduler}.</dd>
40964138
* </dl>

0 commit comments

Comments
 (0)