src/main/java/io/reactivex/internal/operators/flowable/FlowableCache.java
void replay(CacheSubscription<T> consumer) {
        // make sure there is only one replay going on at a time
        if (consumer.getAndIncrement() != 0) {
            return;
        }

        // see if there were more replay request in the meantime
        int missed = 1;
        // read out state into locals upfront to avoid being re-read due to volatile reads
        long index = consumer.index;
        int offset = consumer.offset;
        Node<T> node = consumer.node;
        AtomicLong requested = consumer.requested;
        Subscriber<? super T> downstream = consumer.downstream;
        int capacity = capacityHint;

        for (;;) {
            // first see if the source has terminated, read order matters!
            boolean sourceDone = done;
            // and if the number of items is the same as this consumer has received
            boolean empty = size == index;

            // if the source is done and we have all items so far, terminate the consumer
            if (sourceDone && empty) {
                // release the node object to avoid leaks through retained consumers
                consumer.node = null;
                // if error is not null then the source failed
                Throwable ex = error;
                if (ex != null) {
                    downstream.onError(ex);
                } else {
                    downstream.onComplete();
                }
                return;
            }

            // there are still items not sent to the consumer
            if (!empty) {
                // see how many items the consumer has requested in total so far
                long consumerRequested = requested.get();
                // MIN_VALUE indicates a cancelled consumer, we stop replaying
                if (consumerRequested == Long.MIN_VALUE) {
                    // release the node object to avoid leaks through retained consumers
                    consumer.node = null;
                    return;
                }
                // if the consumer has requested more and there is more, we will emit an item
                if (consumerRequested != index) {

                    // if the offset in the current node has reached the node capacity
                    if (offset == capacity) {
                        // switch to the subsequent node
                        node = node.next;
                        // reset the in-node offset
                        offset = 0;
                    }

                    // emit the cached item
                    downstream.onNext(node.values[offset]);

                    // move the node offset forward
                    offset++;
                    // move the total consumed item count forward
                    index++;

                    // retry for the next item/terminal event if any
                    continue;
                }
            }

            // commit the changed references back
            consumer.index = index;
            consumer.offset = offset;
            consumer.node = node;
            // release the changes and see if there were more replay request in the meantime
            missed = consumer.addAndGet(-missed);
            if (missed == 0) {
                break;
            }
        }
    }
void replay(CacheSubscription<T> consumer) {
        // make sure there is only one replay going on at a time
        if (consumer.getAndIncrement() != 0) {
            return;
        }

        // see if there were more replay request in the meantime
        int missed = 1;
        // read out state into locals upfront to avoid being re-read due to volatile reads
        long index = consumer.index;
        int offset = consumer.offset;
        Node<T> node = consumer.node;
        AtomicLong requested = consumer.requested;
        Subscriber<? super T> downstream = consumer.downstream;
        int capacity = capacityHint;

        for (;;) {
            // first see if the source has terminated, read order matters!
            boolean sourceDone = done;
            // and if the number of items is the same as this consumer has received
            boolean empty = size == index;

            // if the source is done and we have all items so far, terminate the consumer
            if (sourceDone && empty) {
                // release the node object to avoid leaks through retained consumers
                consumer.node = null;
                // if error is not null then the source failed
                Throwable ex = error;
                if (ex != null) {
                    downstream.onError(ex);
                } else {
                    downstream.onComplete();
                }
                return;
            }

            // there are still items not sent to the consumer
            if (!empty) {
                // see how many items the consumer has requested in total so far
                long consumerRequested = requested.get();
                // MIN_VALUE indicates a cancelled consumer, we stop replaying
                if (consumerRequested == Long.MIN_VALUE) {
                    // release the node object to avoid leaks through retained consumers
                    consumer.node = null;
                    return;
                }
                // if the consumer has requested more and there is more, we will emit an item
                if (consumerRequested != index) {

                    // if the offset in the current node has reached the node capacity
                    if (offset == capacity) {
                        // switch to the subsequent node
                        node = node.next;
                        // reset the in-node offset
                        offset = 0;
                    }

                    // emit the cached item
                    downstream.onNext(node.values[offset]);

                    // move the node offset forward
                    offset++;
                    // move the total consumed item count forward
                    index++;

                    // retry for the next item/terminal event if any
                    continue;
                }
            }

            // commit the changed references back
            consumer.index = index;
            consumer.offset = offset;
            consumer.node = node;
            // release the changes and see if there were more replay request in the meantime
            missed = consumer.addAndGet(-missed);
            if (missed == 0) {
                break;
            }
        }
    }
Replays the contents of this cache to the given consumer based on its
current state and number of items requested by it.
@param consumer the consumer to continue replaying items to