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 Detail

      • once

        final java.util.concurrent.atomic.AtomicBoolean once
        The subscription to the source should happen at most once.
      • capacityHint

        final int capacityHint
        The number of items per cached nodes.
      • subscribers

        final java.util.concurrent.atomic.AtomicReference<FlowableCache.CacheSubscription<T>[]> 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.
      • tailOffset

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

        java.lang.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 Detail

      • 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 Detail

      • 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
      • 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​(java.lang.Throwable t)
        Specified by:
        onError in interface org.reactivestreams.Subscriber<T>
      • onComplete

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