Class FlowableCache<T>

Type Parameters:
T - the source element type
All Implemented Interfaces:
FlowableSubscriber<T>, HasUpstreamPublisher<T>, org.reactivestreams.Publisher<T>, org.reactivestreams.Subscriber<T>

public final class FlowableCache<T> extends AbstractFlowableWithUpstream<T,T> implements FlowableSubscriber<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.
    • subscribers

      The current known array of subscriber state to notify.
    • EMPTY

      static final FlowableCache.CacheSubscription[] EMPTY
      A shared instance of an empty array of subscribers to avoid creating a new empty array when all subscribers cancel.
    • TERMINATED

      static final FlowableCache.CacheSubscription[] TERMINATED
      A shared instance indicating the source has no more events and there is no need to remember subscribers anymore.
    • size

      volatile long size
      The total number of elements in the list available for reads.
    • 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 subscribers is TERMINATED, this holds the terminal error if not null.
    • done

      volatile boolean done
      True if the source has terminated.
  • Constructor Details

    • FlowableCache

      public FlowableCache(Flowable<T> source, int capacityHint)
      Constructs an empty, non-connected cache.
      Parameters:
      source - the source to subscribe to for the first incoming subscriber
      capacityHint - the number of items expected (reduce allocation frequency)
  • Method Details

    • subscribeActual

      protected void subscribeActual(org.reactivestreams.Subscriber<? super T> t)
      Description copied from class: Flowable
      Operator implementations (both source and intermediate) should implement this method that performs the necessary business logic and handles the incoming Subscribers.

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

      Specified by:
      subscribeActual in class Flowable<T>
      Parameters:
      t - the incoming Subscriber, never null
    • isConnected

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

      boolean hasSubscribers()
      Returns true if there are observers subscribed to this observable.
      Returns:
      true if the cache has Subscribers
    • cachedEventCount

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

      void add(FlowableCache.CacheSubscription<T> consumer)
      Atomically adds the consumer to the subscribers copy-on-write array if the source has not yet terminated.
      Parameters:
      consumer - the consumer to add
    • remove

      void remove(FlowableCache.CacheSubscription<T> consumer)
      Atomically removes the consumer from the subscribers copy-on-write array.
      Parameters:
      consumer - the consumer to remove
    • replay

      void replay(FlowableCache.CacheSubscription<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(org.reactivestreams.Subscription s)
      Description copied from interface: FlowableSubscriber
      Implementors of this method should make sure everything that needs to be visible in Subscriber.onNext(Object) is established before calling Subscription.request(long). In practice this means no initialization should happen after the request() call and additional behavior is thread safe in respect to onNext.
      Specified by:
      onSubscribe in interface FlowableSubscriber<T>
      Specified by:
      onSubscribe in interface org.reactivestreams.Subscriber<T>
    • onNext

      public void onNext(T t)
      Specified by:
      onNext in interface org.reactivestreams.Subscriber<T>
    • onError

      public void onError(Throwable t)
      Specified by:
      onError in interface org.reactivestreams.Subscriber<T>
    • onComplete

      public void onComplete()
      Specified by:
      onComplete in interface org.reactivestreams.Subscriber<T>