/*
 * Copyright (c) 2016-present, RxJava Contributors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
 * compliance with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
 * the License for the specific language governing permissions and limitations under the License.
 */

package io.reactivex.rxjava3.internal.operators.flowable;

import java.util.concurrent.atomic.*;

import org.reactivestreams.*;

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
import io.reactivex.rxjava3.internal.util.BackpressureHelper;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;

/**
 * An observable which auto-connects to another observable, caches the elements
 * from that observable but allows terminating the connection and completing the cache.
 *
 * @param <T> the source element type
 */
public final class FlowableCache<T> extends AbstractFlowableWithUpstream<T, T>
implements FlowableSubscriber<T> {

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

    /**
     * The number of items per cached nodes.
     */
    final int capacityHint;

    /**
     * The current known array of subscriber state to notify.
     */
    final AtomicReference<CacheSubscription<T>[]> subscribers;

    /**
     * A shared instance of an empty array of subscribers to avoid creating
     * a new empty array when all subscribers cancel.
     */
    @SuppressWarnings("rawtypes")
    static final CacheSubscription[] EMPTY = new CacheSubscription[0];
    /**
     * A shared instance indicating the source has no more events and there
     * is no need to remember subscribers anymore.
     */
    @SuppressWarnings("rawtypes")
    static final CacheSubscription[] TERMINATED = new CacheSubscription[0];

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

    /**
     * The starting point of the cached items.
     */
    final Node<T> head;

    /**
     * The current tail of the linked structure holding the items.
     */
    Node<T> tail;

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

    /**
     * If {@link #subscribers} is {@link #TERMINATED}, this holds the terminal error if not null.
     */
    Throwable error;

    /**
     * True if the source has terminated.
     */
    volatile boolean done;

    /**
     * Constructs an empty, non-connected cache.
     * @param source the source to subscribe to for the first incoming subscriber
     * @param capacityHint the number of items expected (reduce allocation frequency)
     */
    @SuppressWarnings("unchecked")
    public FlowableCache(Flowable<T> source, int capacityHint) {
        super(source);
        this.capacityHint = capacityHint;
        this.once = new AtomicBoolean();
        Node<T> n = new Node<>(capacityHint);
        this.head = n;
        this.tail = n;
        this.subscribers = new AtomicReference<>(EMPTY);
    }

    @Override
    protected void subscribeActual(Subscriber<? super T> t) {
        CacheSubscription<T> consumer = new CacheSubscription<>(t, this);
        t.onSubscribe(consumer);
        add(consumer);

        if (!once.get() && once.compareAndSet(false, true)) {
            source.subscribe(this);
        } else {
            replay(consumer);
        }
    }

    /**
     * Check if this cached observable is connected to its source.
     * @return true if already connected
     */
    /* public */boolean isConnected() {
        return once.get();
    }

    /**
     * Returns true if there are observers subscribed to this observable.
     * @return true if the cache has Subscribers
     */
    /* public */ boolean hasSubscribers() {
        return subscribers.get().length != 0;
    }

    /**
     * Returns the number of events currently cached.
     * @return the number of currently cached event count
     */
    /* public */ long cachedEventCount() {
        return size;
    }

    /**
     * Atomically adds the consumer to the {@link #subscribers} copy-on-write array
     * if the source has not yet terminated.
     * @param consumer the consumer to add
     */
    void add(CacheSubscription<T> consumer) {
        for (;;) {
            CacheSubscription<T>[] current = subscribers.get();
            if (current == TERMINATED) {
                return;
            }
            int n = current.length;

            @SuppressWarnings("unchecked")
            CacheSubscription<T>[] next = new CacheSubscription[n + 1];
            System.arraycopy(current, 0, next, 0, n);
            next[n] = consumer;

            if (subscribers.compareAndSet(current, next)) {
                return;
            }
        }
    }

    /**
     * Atomically removes the consumer from the {@link #subscribers} copy-on-write array.
     * @param consumer the consumer to remove
     */
    @SuppressWarnings("unchecked")
    void remove(CacheSubscription<T> consumer) {
        for (;;) {
            CacheSubscription<T>[] current = subscribers.get();
            int n = current.length;
            if (n == 0) {
                return;
            }

            int j = -1;
            for (int i = 0; i < n; i++) {
                if (current[i] == consumer) {
                    j = i;
                    break;
                }
            }

            if (j < 0) {
                return;
            }
            CacheSubscription<T>[] next;

            if (n == 1) {
                next = EMPTY;
            } else {
                next = new CacheSubscription[n - 1];
                System.arraycopy(current, 0, next, 0, j);
                System.arraycopy(current, j + 1, next, j, n - j - 1);
            }

            if (subscribers.compareAndSet(current, next)) {
                return;
            }
        }
    }

    /**
     * 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
     */
    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;
            }
        }
    }

    @Override
    public void onSubscribe(Subscription s) {
        s.request(Long.MAX_VALUE);
    }

    @Override
    public void onNext(T t) {
        int tailOffset = this.tailOffset;
        // if the current tail node is full, create a fresh node
        if (tailOffset == capacityHint) {
            Node<T> n = new Node<>(tailOffset);
            n.values[0] = t;
            this.tailOffset = 1;
            tail.next = n;
            tail = n;
        } else {
            tail.values[tailOffset] = t;
            this.tailOffset = tailOffset + 1;
        }
        size++;
        for (CacheSubscription<T> consumer : subscribers.get()) {
            replay(consumer);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public void onError(Throwable t) {
        if (done) {
            RxJavaPlugins.onError(t);
            return;
        }
        error = t;
        done = true;
        for (CacheSubscription<T> consumer : subscribers.getAndSet(TERMINATED)) {
            replay(consumer);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public void onComplete() {
        done = true;
        for (CacheSubscription<T> consumer : subscribers.getAndSet(TERMINATED)) {
            replay(consumer);
        }
    }

    /**
     * Hosts the downstream consumer and its current requested and replay states.
     * {@code this} holds the work-in-progress counter for the serialized replay.
     * @param <T> the value type
     */
    static final class CacheSubscription<T> extends AtomicInteger
    implements Subscription {

        private static final long serialVersionUID = 6770240836423125754L;

        final Subscriber<? super T> downstream;

        final FlowableCache<T> parent;

        final AtomicLong requested;

        Node<T> node;

        int offset;

        long index;

        /**
         * Constructs a new instance with the actual downstream consumer and
         * the parent cache object.
         * @param downstream the actual consumer
         * @param parent the parent that holds onto the cached items
         */
        CacheSubscription(Subscriber<? super T> downstream, FlowableCache<T> parent) {
            this.downstream = downstream;
            this.parent = parent;
            this.node = parent.head;
            this.requested = new AtomicLong();
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.addCancel(requested, n);
                parent.replay(this);
            }
        }

        @Override
        public void cancel() {
            if (requested.getAndSet(Long.MIN_VALUE) != Long.MIN_VALUE) {
                parent.remove(this);
            }
        }
    }

    /**
     * Represents a segment of the cached item list as
     * part of a linked-node-list structure.
     * @param <T> the element type
     */
    static final class Node<T> {

        /**
         * The array of values held by this node.
         */
        final T[] values;

        /**
         * The next node if not null.
         */
        volatile Node<T> next;

        @SuppressWarnings("unchecked")
        Node(int capacityHint) {
            this.values = (T[])new Object[capacityHint];
        }
    }
}
