Class ObservableCache<T>

java.lang.Object
io.reactivex.rxjava3.core.Observable<T>
io.reactivex.rxjava3.internal.operators.observable.AbstractObservableWithUpstream<T,T>
io.reactivex.rxjava3.internal.operators.observable.ObservableCache<T>
Type Parameters:
T - the source element type
All Implemented Interfaces:
ObservableSource<T>, Observer<T>, HasUpstreamObservableSource<T>
public final class ObservableCache<T> extends AbstractObservableWithUpstream<T,T> implements Observer<T>
An observable which auto-connects to another observable, caches the elements from that observable but allows terminating the connection and completing the cache.

  • Field Details

    • once

      final AtomicBoolean once
      The subscription to the source should happen at most once.

    • capacityHint

      final int capacityHint
      The number of items per cached nodes.

    • observers

      final AtomicReference<ObservableCache.CacheDisposable<T>[]> observers
      The current known array of observer state to notify.

    • EMPTY

      static final ObservableCache.CacheDisposable[] EMPTY
      A shared instance of an empty array of observers to avoid creating a new empty array when all observers dispose.

    • TERMINATED

      static final ObservableCache.CacheDisposable[] TERMINATED
      A shared instance indicating the source has no more events and there is no need to remember observers anymore.

    • size

      volatile long size
      The total number of elements in the list available for reads.

    • tail

      ObservableCache.Node<T> tail
      The current tail of the linked structure holding the items.

    • tailOffset

      int tailOffset
      How many items have been put into the tail node so far.

    • error

      Throwable error
      If observers is TERMINATED, this holds the terminal error if not null.

    • done

      volatile boolean done
      True if the source has terminated.

  • Constructor Details

    • ObservableCache

      public ObservableCache(Observable<T> source, int capacityHint)
      Constructs an empty, non-connected cache.
      Parameters:
      source - the source to subscribe to for the first incoming observer
      capacityHint - the number of items expected (reduce allocation frequency)

  • Method Details

    • subscribeActual

      protected void subscribeActual(Observer<? super T> t)
      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:
      t - the incoming Observer, never null

    • isConnected

      boolean isConnected()
      Check if this cached observable is connected to its source.
      Returns:
      true if already connected

    • hasObservers

      boolean hasObservers()
      Returns true if there are observers subscribed to this observable.
      Returns:
      true if the cache has observers

    • cachedEventCount

      long cachedEventCount()
      Returns the number of events currently cached.
      Returns:
      the number of currently cached event count

    • add

      void add(ObservableCache.CacheDisposable<T> consumer)
      Atomically adds the consumer to the observers copy-on-write array if the source has not yet terminated.
      Parameters:
      consumer - the consumer to add

    • remove

      void remove(ObservableCache.CacheDisposable<T> consumer)
      Atomically removes the consumer from the observers copy-on-write array.
      Parameters:
      consumer - the consumer to remove

    • replay

      void replay(ObservableCache.CacheDisposable<T> consumer)
      Replays the contents of this cache to the given consumer based on its current state and number of items requested by it.
      Parameters:
      consumer - the consumer to continue replaying items to

    • onSubscribe

      public void onSubscribe(Disposable d)
      Description copied from interface: Observer
      Provides the Observer with the means of cancelling (disposing) the connection (channel) with the Observable in both synchronous (from within Observer.onNext(Object)) and asynchronous manner.
      Specified by:
      onSubscribe in interface Observer<T>
      Parameters:
      d - the Disposable instance whose Disposable.dispose() can be called anytime to cancel the connection

    • onNext

      public void onNext(T t)
      Description copied from interface: Observer
      Provides the Observer with a new item to observe.

      The Observable may call this method 0 or more times.

      The Observable will not call this method again after it calls either Observer.onComplete() or Observer.onError(Throwable).

      Specified by:
      onNext in interface Observer<T>
      Parameters:
      t - the item emitted by the Observable

    • onError

      public void onError(Throwable t)
      Description copied from interface: Observer
      Notifies the Observer that the Observable has experienced an error condition.

      If the Observable calls this method, it will not thereafter call Observer.onNext(T) or Observer.onComplete().

      Specified by:
      onError in interface Observer<T>
      Parameters:
      t - the exception encountered by the Observable

    • onComplete

      public void onComplete()
      Description copied from interface: Observer
      Notifies the Observer that the Observable has finished sending push-based notifications.

      The Observable will not call this method if it calls Observer.onError(Throwable).

      Specified by:
      onComplete in interface Observer<T>