Package io.reactivex.rxjava3.internal.operators.observable

Class ObservableReplay<T>

java.lang.Object

io.reactivex.rxjava3.core.Observable<T>

io.reactivex.rxjava3.observables.ConnectableObservable<T>

io.reactivex.rxjava3.internal.operators.observable.ObservableReplay<T>

All Implemented Interfaces:

ObservableSource<T>, HasUpstreamObservableSource<T>

public final class ObservableReplay<T>
extends ConnectableObservable<T>
implements HasUpstreamObservableSource<T>

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
(package private) static class	ObservableReplay.BoundedReplayBuffer<T>	Base class for bounded buffering with options to specify an enter and leave transforms and custom truncation behavior.
(package private) static interface	ObservableReplay.BufferSupplier<T>
(package private) static class	ObservableReplay.DisposeConsumer<R>
(package private) static class	ObservableReplay.InnerDisposable<T>	A Disposable that manages the disposed state of a child Observer in thread-safe manner.
(package private) static class	ObservableReplay.MulticastReplay<R,U>
(package private) static class	ObservableReplay.Node	Represents a node in a bounded replay buffer's linked list.
(package private) static interface	ObservableReplay.ReplayBuffer<T>	The interface for interacting with various buffering logic.
(package private) static class	ObservableReplay.ReplayBufferSupplier<T>
(package private) static class	ObservableReplay.ReplayObserver<T>
(package private) static class	ObservableReplay.ReplaySource<T>
(package private) static class	ObservableReplay.ScheduledReplaySupplier<T>
(package private) static class	ObservableReplay.SizeAndTimeBoundReplayBuffer<T>	Size and time bound replay buffer.
(package private) static class	ObservableReplay.SizeBoundReplayBuffer<T>	A bounded replay buffer implementation with size limit only.
(package private) static class	ObservableReplay.UnBoundedFactory
(package private) static class	ObservableReplay.UnboundedReplayBuffer<T>	Holds an unbounded list of events.

Field Summary

Fields

Modifier and Type	Field	Description
(package private) ObservableReplay.BufferSupplier<T>	bufferFactory	A factory that creates the appropriate buffer for the ReplayObserver.
(package private) java.util.concurrent.atomic.AtomicReference<ObservableReplay.ReplayObserver<T>>	current	Holds the current subscriber that is, will be or just was subscribed to the source observable.
(package private) static ObservableReplay.BufferSupplier	DEFAULT_UNBOUNDED_FACTORY
(package private) ObservableSource<T>	onSubscribe
(package private) ObservableSource<T>	source	The source observable.

Constructor Summary

Constructors

Modifier	Constructor	Description
private	ObservableReplay(ObservableSource<T> onSubscribe, ObservableSource<T> source, java.util.concurrent.atomic.AtomicReference<ObservableReplay.ReplayObserver<T>> current, ObservableReplay.BufferSupplier<T> bufferFactory)

Method Summary

Modifier and Type	Method	Description
void	connect(Consumer<? super Disposable> connection)	Instructs the ConnectableObservable to begin emitting the items from its underlying Observable to its Observers.
static <T> ConnectableObservable<T>	create(ObservableSource<T> source, int bufferSize, boolean eagerTruncate)	Creates a replaying ConnectableObservable with a size bound buffer.
static <T> ConnectableObservable<T>	create(ObservableSource<T> source, long maxAge, java.util.concurrent.TimeUnit unit, Scheduler scheduler, boolean eagerTruncate)	Creates a replaying ConnectableObservable with a time bound buffer.
static <T> ConnectableObservable<T>	create(ObservableSource<T> source, long maxAge, java.util.concurrent.TimeUnit unit, Scheduler scheduler, int bufferSize, boolean eagerTruncate)	Creates a replaying ConnectableObservable with a size and time bound buffer.
(package private) static <T> ConnectableObservable<T>	create(ObservableSource<T> source, ObservableReplay.BufferSupplier<T> bufferFactory)	Creates a OperatorReplay instance to replay values of the given source observable.
static <T> ConnectableObservable<T>	createFrom(ObservableSource<? extends T> source)	Creates a replaying ConnectableObservable with an unbounded buffer.
static <U,R> Observable<R>	multicastSelector(Supplier<? extends ConnectableObservable<U>> connectableFactory, Function<? super Observable<U>,? extends ObservableSource<R>> selector)	Given a connectable observable factory, it multicasts over the generated ConnectableObservable via a selector function.
void	reset()	Resets this ConnectableObservable into its fresh state if it has terminated or has been disposed.
ObservableSource<T>	source()	Returns the upstream source of this Observable.
protected void	subscribeActual(Observer<? super T> observer)	Operator implementations (both source and intermediate) should implement this method that performs the necessary business logic and handles the incoming Observers.

Methods inherited from class io.reactivex.rxjava3.observables.ConnectableObservable

autoConnect, autoConnect, autoConnect, connect, refCount, refCount, refCount, refCount, refCount, refCount

Methods inherited from class io.reactivex.rxjava3.core.Observable

all, amb, ambArray, ambWith, any, blockingFirst, blockingFirst, blockingForEach, blockingForEach, blockingIterable, blockingIterable, blockingLast, blockingLast, blockingLatest, blockingMostRecent, blockingNext, blockingSingle, blockingSingle, blockingStream, blockingStream, blockingSubscribe, blockingSubscribe, blockingSubscribe, blockingSubscribe, blockingSubscribe, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, buffer, bufferSize, cache, cacheWithInitialCapacity, cast, collect, collect, collectInto, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatest, combineLatestArray, combineLatestArray, combineLatestArrayDelayError, combineLatestArrayDelayError, combineLatestDelayError, combineLatestDelayError, compose, concat, concat, concat, concat, concat, concat, concatArray, concatArrayDelayError, concatArrayEager, concatArrayEager, concatArrayEagerDelayError, concatArrayEagerDelayError, concatDelayError, concatDelayError, concatDelayError, concatEager, concatEager, concatEager, concatEager, concatEagerDelayError, concatEagerDelayError, concatEagerDelayError, concatEagerDelayError, concatMap, concatMap, concatMap, concatMapCompletable, concatMapCompletable, concatMapCompletableDelayError, concatMapCompletableDelayError, concatMapCompletableDelayError, concatMapDelayError, concatMapDelayError, concatMapDelayError, concatMapEager, concatMapEager, concatMapEagerDelayError, concatMapEagerDelayError, concatMapIterable, concatMapMaybe, concatMapMaybe, concatMapMaybeDelayError, concatMapMaybeDelayError, concatMapMaybeDelayError, concatMapSingle, concatMapSingle, concatMapSingleDelayError, concatMapSingleDelayError, concatMapSingleDelayError, concatMapStream, concatWith, concatWith, concatWith, concatWith, contains, count, create, debounce, debounce, debounce, debounce, defaultIfEmpty, defer, delay, delay, delay, delay, delay, delay, delaySubscription, delaySubscription, delaySubscription, dematerialize, distinct, distinct, distinct, distinctUntilChanged, distinctUntilChanged, distinctUntilChanged, doAfterNext, doAfterTerminate, doFinally, doOnComplete, doOnDispose, doOnEach, doOnEach, doOnError, doOnLifecycle, doOnNext, doOnSubscribe, doOnTerminate, elementAt, elementAt, elementAtOrError, empty, error, error, filter, first, firstElement, firstOrError, firstOrErrorStage, firstStage, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMap, flatMapCompletable, flatMapCompletable, flatMapIterable, flatMapIterable, flatMapMaybe, flatMapMaybe, flatMapSingle, flatMapSingle, flatMapStream, forEach, forEachWhile, forEachWhile, forEachWhile, fromAction, fromArray, fromCallable, fromCompletable, fromCompletionStage, fromFuture, fromFuture, fromIterable, fromMaybe, fromOptional, fromPublisher, fromRunnable, fromSingle, fromStream, fromSupplier, generate, generate, generate, generate, generate, groupBy, groupBy, groupBy, groupBy, groupBy, groupJoin, hide, ignoreElements, interval, interval, interval, interval, intervalRange, intervalRange, isEmpty, join, just, just, just, just, just, just, just, just, just, just, last, lastElement, lastOrError, lastOrErrorStage, lastStage, lift, map, mapOptional, materialize, merge, merge, merge, merge, merge, merge, merge, merge, mergeArray, mergeArray, mergeArrayDelayError, mergeArrayDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeDelayError, mergeWith, mergeWith, mergeWith, mergeWith, never, observeOn, observeOn, observeOn, ofType, onErrorComplete, onErrorComplete, onErrorResumeNext, onErrorResumeWith, onErrorReturn, onErrorReturnItem, onTerminateDetach, publish, publish, range, rangeLong, reduce, reduce, reduceWith, repeat, repeat, repeatUntil, repeatWhen, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, replay, retry, retry, retry, retry, retry, retryUntil, retryWhen, safeSubscribe, sample, sample, sample, sample, sample, sample, sample, scan, scan, scanWith, sequenceEqual, sequenceEqual, sequenceEqual, sequenceEqual, serialize, share, single, singleElement, singleOrError, singleOrErrorStage, singleStage, skip, skip, skip, skipLast, skipLast, skipLast, skipLast, skipLast, skipLast, skipUntil, skipWhile, sorted, sorted, startWith, startWith, startWith, startWith, startWithArray, startWithItem, startWithIterable, subscribe, subscribe, subscribe, subscribe, subscribe, subscribe, subscribeOn, subscribeWith, switchIfEmpty, switchMap, switchMap, switchMapCompletable, switchMapCompletableDelayError, switchMapDelayError, switchMapDelayError, switchMapMaybe, switchMapMaybeDelayError, switchMapSingle, switchMapSingleDelayError, switchOnNext, switchOnNext, switchOnNextDelayError, switchOnNextDelayError, take, take, take, takeLast, takeLast, takeLast, takeLast, takeLast, takeLast, takeLast, takeLast, takeLast, takeUntil, takeUntil, takeWhile, test, test, throttleFirst, throttleFirst, throttleFirst, throttleLast, throttleLast, throttleLast, throttleLatest, throttleLatest, throttleLatest, throttleLatest, throttleLatest, throttleWithTimeout, throttleWithTimeout, throttleWithTimeout, timeInterval, timeInterval, timeInterval, timeInterval, timeout, timeout, timeout, timeout, timeout, timeout, timeout, timeout, timer, timer, timestamp, timestamp, timestamp, timestamp, to, toFlowable, toFuture, toList, toList, toList, toMap, toMap, toMap, toMultimap, toMultimap, toMultimap, toMultimap, toSortedList, toSortedList, toSortedList, toSortedList, unsafeCreate, unsubscribeOn, using, using, window, window, window, window, window, window, window, window, window, window, window, window, window, window, window, window, window, withLatestFrom, withLatestFrom, withLatestFrom, withLatestFrom, withLatestFrom, withLatestFrom, wrap, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zip, zipArray, zipWith, zipWith, zipWith, zipWith

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

source

final ObservableSource<T> source

The source observable.

current

final java.util.concurrent.atomic.AtomicReference<ObservableReplay.ReplayObserver<T>> current

Holds the current subscriber that is, will be or just was subscribed to the source observable.

bufferFactory

final ObservableReplay.BufferSupplier<T> bufferFactory

A factory that creates the appropriate buffer for the ReplayObserver.

onSubscribe

final ObservableSource<T> onSubscribe

DEFAULT_UNBOUNDED_FACTORY

static final ObservableReplay.BufferSupplier DEFAULT_UNBOUNDED_FACTORY

Constructor Detail

ObservableReplay

private ObservableReplay(ObservableSource<T> onSubscribe,
                         ObservableSource<T> source,
                         java.util.concurrent.atomic.AtomicReference<ObservableReplay.ReplayObserver<T>> current,
                         ObservableReplay.BufferSupplier<T> bufferFactory)

Method Detail

multicastSelector

public static <U,R> Observable<R> multicastSelector(Supplier<? extends ConnectableObservable<U>> connectableFactory,
                                                          Function<? super Observable<U>,? extends ObservableSource<R>> selector)

Given a connectable observable factory, it multicasts over the generated ConnectableObservable via a selector function.

Type Parameters:

U - the value type of the ConnectableObservable

R - the result value type

Parameters:

connectableFactory - the factory that returns a ConnectableObservable for each individual subscriber

selector - the function that receives an Observable and should return another Observable that will be subscribed to

Returns:

the new Observable instance

createFrom

public static <T> ConnectableObservable<T> createFrom(ObservableSource<? extends T> source)

Creates a replaying ConnectableObservable with an unbounded buffer.

Type Parameters:

T - the value type

Parameters:

source - the source observable

Returns:

the new ConnectableObservable instance

create

public static <T> ConnectableObservable<T> create(ObservableSource<T> source,
                                                  int bufferSize,
                                                  boolean eagerTruncate)

Creates a replaying ConnectableObservable with a size bound buffer.

Type Parameters:

T - the value type

Parameters:

source - the source ObservableSource to use

bufferSize - the maximum number of elements to hold

eagerTruncate - if true, the head reference is refreshed to avoid unwanted item retention

Returns:

the new ConnectableObservable instance

create

public static <T> ConnectableObservable<T> create(ObservableSource<T> source,
                                                  long maxAge,
                                                  java.util.concurrent.TimeUnit unit,
                                                  Scheduler scheduler,
                                                  boolean eagerTruncate)

Creates a replaying ConnectableObservable with a time bound buffer.

Type Parameters:

T - the value type

Parameters:

source - the source ObservableSource to use

maxAge - the maximum age of entries

unit - the unit of measure of the age amount

scheduler - the target scheduler providing the current time

eagerTruncate - if true, the head reference is refreshed to avoid unwanted item retention

Returns:

the new ConnectableObservable instance

create

public static <T> ConnectableObservable<T> create(ObservableSource<T> source,
                                                  long maxAge,
                                                  java.util.concurrent.TimeUnit unit,
                                                  Scheduler scheduler,
                                                  int bufferSize,
                                                  boolean eagerTruncate)

Creates a replaying ConnectableObservable with a size and time bound buffer.

Type Parameters:

T - the value type

Parameters:

source - the source ObservableSource to use

maxAge - the maximum age of entries

unit - the unit of measure of the age amount

scheduler - the target scheduler providing the current time

bufferSize - the maximum number of elements to hold

eagerTruncate - if true, the head reference is refreshed to avoid unwanted item retention

Returns:

the new ConnectableObservable instance

create

static <T> ConnectableObservable<T> create(ObservableSource<T> source,
                                           ObservableReplay.BufferSupplier<T> bufferFactory)

Creates a OperatorReplay instance to replay values of the given source observable.

Type Parameters:

T - the value type

Parameters:

source - the source observable

bufferFactory - the factory to instantiate the appropriate buffer when the observable becomes active

Returns:

the connectable observable

source

public ObservableSource<T> source()

Description copied from interface: HasUpstreamObservableSource

Returns the upstream source of this Observable.

Allows discovering the chain of observables.

Specified by:

source in interface HasUpstreamObservableSource<T>

Returns:

the source ObservableSource

reset

public void reset()

Description copied from class: ConnectableObservable

Resets this ConnectableObservable into its fresh state if it has terminated or has been disposed.

Calling this method on a fresh or active ConnectableObservable has no effect.

Scheduler:

The behavior is determined by the implementor of this abstract class.

Specified by:

reset in class ConnectableObservable<T>

subscribeActual

protected void subscribeActual(Observer<? super T> observer)

Description copied from class: Observable

Operator implementations (both source and intermediate) should implement this method that performs the necessary business logic and handles the incoming Observers.

There is no need to call any of the plugin hooks on the current Observable instance or the Observer; all hooks and basic safeguards have been applied by Observable.subscribe(Observer) before this method gets called.

Specified by:

subscribeActual in class Observable<T>

Parameters:

observer - the incoming Observer, never null

connect

public void connect(Consumer<? super Disposable> connection)

Description copied from class: ConnectableObservable

Instructs the ConnectableObservable to begin emitting the items from its underlying Observable to its Observers.

Scheduler:

The behavior is determined by the implementor of this abstract class.

Specified by:

connect in class ConnectableObservable<T>

Parameters:

connection - the action that receives the connection subscription before the subscription to source happens allowing the caller to synchronously disconnect a synchronous source

See Also:

ReactiveX documentation: Connect