/**
 * Copyright 2014 Netflix, Inc.
 *
 * 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 rx.internal.operators;

import java.util.*;
import java.util.concurrent.atomic.*;

import rx.*;
import rx.Observable;
import rx.Observable.Operator;
import rx.functions.Action0;
import rx.internal.util.atomic.SpscLinkedArrayQueue;
import rx.subjects.*;
import rx.subscriptions.Subscriptions;

/**
 * Creates windows of values into the source sequence with skip frequency and size bounds.
 *
 * If skip == size then the windows are non-overlapping, otherwise, windows may overlap
 * or can be discontinuous. The returned Observable sequence is cold and need to be
 * consumed while the window operation is in progress.
 *
 * <p>Note that this conforms the Rx.NET behavior, but does not match former RxJava
 * behavior, which operated as a regular buffer and mapped its lists to Observables.</p>
 *
 * @param <T> the value type
 */
public final class OperatorWindowWithSize<T> implements Operator<Observable<T>, T> {
    final int size;
    final int skip;

    public OperatorWindowWithSize(int size, int skip) {
        this.size = size;
        this.skip = skip;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super Observable<T>> child) {
        if (skip == size) {
            WindowExact<T> parent = new WindowExact<T>(child, size);

            child.add(parent.cancel);
            child.setProducer(parent.createProducer());

            return parent;
        } else
        if (skip > size) {
            WindowSkip<T> parent = new WindowSkip<T>(child, size, skip);

            child.add(parent.cancel);
            child.setProducer(parent.createProducer());

            return parent;
        }

        WindowOverlap<T> parent = new WindowOverlap<T>(child, size, skip);

        child.add(parent.cancel);
        child.setProducer(parent.createProducer());

        return parent;

    }

    static final class WindowExact<T> extends Subscriber<T> implements Action0 {
        final Subscriber<? super Observable<T>> actual;

        final int size;

        final AtomicInteger wip;

        final Subscription cancel;

        int index;

        Subject<T, T> window;

        public WindowExact(Subscriber<? super Observable<T>> actual, int size) {
            this.actual = actual;
            this.size = size;
            this.wip = new AtomicInteger(1);
            this.cancel = Subscriptions.create(this);
            this.add(cancel);
            this.request(0);
        }

        @Override
        public void onNext(T t) {
            int i = index;

            Subject<T, T> w = window;
            if (i == 0) {
                wip.getAndIncrement();

                w = UnicastSubject.create(size, this);
                window = w;

                actual.onNext(w);
            }
            i++;

            w.onNext(t);

            if (i == size) {
                index = 0;
                window = null;
                w.onCompleted();
            } else {
                index = i;
            }
        }

        @Override
        public void onError(Throwable e) {
            Subject<T, T> w = window;

            if (w != null) {
                window = null;
                w.onError(e);
            }
            actual.onError(e);
        }

        @Override
        public void onCompleted() {
            Subject<T, T> w = window;

            if (w != null) {
                window = null;
                w.onCompleted();
            }
            actual.onCompleted();
        }

        Producer createProducer() {
            return new Producer() {
                @Override
                public void request(long n) {
                    if (n < 0L) {
                        throw new IllegalArgumentException("n >= 0 required but it was " + n);
                    }
                    if (n != 0L) {
                        long u = BackpressureUtils.multiplyCap(size, n);
                        WindowExact.this.request(u);
                    }
                }
            };
        }

        @Override
        public void call() {
            if (wip.decrementAndGet() == 0) {
                unsubscribe();
            }
        }
    }

    static final class WindowSkip<T> extends Subscriber<T> implements Action0 {
        final Subscriber<? super Observable<T>> actual;

        final int size;

        final int skip;

        final AtomicInteger wip;

        final Subscription cancel;

        int index;

        Subject<T, T> window;

        public WindowSkip(Subscriber<? super Observable<T>> actual, int size, int skip) {
            this.actual = actual;
            this.size = size;
            this.skip = skip;
            this.wip = new AtomicInteger(1);
            this.cancel = Subscriptions.create(this);
            this.add(cancel);
            this.request(0);
        }

        @Override
        public void onNext(T t) {
            int i = index;

            Subject<T, T> w = window;
            if (i == 0) {
                wip.getAndIncrement();

                w = UnicastSubject.create(size, this);
                window = w;

                actual.onNext(w);
            }
            i++;

            if (w != null) {
                w.onNext(t);
            }

            if (i == size) {
                index = i;
                window = null;
                w.onCompleted();
            } else
            if (i == skip) {
                index = 0;
            } else {
                index = i;
            }

        }

        @Override
        public void onError(Throwable e) {
            Subject<T, T> w = window;

            if (w != null) {
                window = null;
                w.onError(e);
            }
            actual.onError(e);
        }

        @Override
        public void onCompleted() {
            Subject<T, T> w = window;

            if (w != null) {
                window = null;
                w.onCompleted();
            }
            actual.onCompleted();
        }

        Producer createProducer() {
            return new WindowSkipProducer();
        }

        @Override
        public void call() {
            if (wip.decrementAndGet() == 0) {
                unsubscribe();
            }
        }

        final class WindowSkipProducer extends AtomicBoolean implements Producer {
            /** */
            private static final long serialVersionUID = 4625807964358024108L;

            @Override
            public void request(long n) {
                if (n < 0L) {
                    throw new IllegalArgumentException("n >= 0 required but it was " + n);
                }
                if (n != 0L) {
                    WindowSkip<T> parent = WindowSkip.this;
                    if (!get() && compareAndSet(false, true)) {
                        long u = BackpressureUtils.multiplyCap(n, parent.size);
                        long v = BackpressureUtils.multiplyCap(parent.skip - parent.size, n - 1);
                        long w = BackpressureUtils.addCap(u, v);
                        parent.request(w);
                    } else {
                        long u = BackpressureUtils.multiplyCap(n, parent.skip);
                        parent.request(u);
                    }
                }
            }
        }
    }

    static final class WindowOverlap<T> extends Subscriber<T> implements Action0 {
        final Subscriber<? super Observable<T>> actual;

        final int size;

        final int skip;

        final AtomicInteger wip;

        final Subscription cancel;

        final ArrayDeque<Subject<T, T>> windows;

        final AtomicLong requested;

        final AtomicInteger drainWip;

        final Queue<Subject<T, T>> queue;

        Throwable error;

        volatile boolean done;

        int index;

        int produced;

        public WindowOverlap(Subscriber<? super Observable<T>> actual, int size, int skip) {
            this.actual = actual;
            this.size = size;
            this.skip = skip;
            this.wip = new AtomicInteger(1);
            this.windows = new ArrayDeque<Subject<T, T>>();
            this.drainWip = new AtomicInteger();
            this.requested = new AtomicLong();
            this.cancel = Subscriptions.create(this);
            this.add(cancel);
            this.request(0);
            int maxWindows = (size + (skip - 1)) / skip;
            this.queue = new SpscLinkedArrayQueue<Subject<T, T>>(maxWindows);
        }

        @Override
        public void onNext(T t) {
            int i = index;

            ArrayDeque<Subject<T, T>> q = windows;

            if (i == 0 && !actual.isUnsubscribed()) {
                wip.getAndIncrement();

                Subject<T, T> w = UnicastSubject.create(16, this);
                q.offer(w);

                queue.offer(w);
                drain();
            }

            for (Subject<T, T> w : windows) {
                w.onNext(t);
            }

            int p = produced + 1;

            if (p == size) {
                produced = p - skip;

                Subject<T, T> w = q.poll();
                if (w != null) {
                    w.onCompleted();
                }
            } else {
                produced = p;
            }

            i++;
            if (i == skip) {
                index = 0;
            } else {
                index = i;
            }
        }

        @Override
        public void onError(Throwable e) {
            for (Subject<T, T> w : windows) {
                w.onError(e);
            }
            windows.clear();

            error = e;
            done = true;
            drain();
        }

        @Override
        public void onCompleted() {
            for (Subject<T, T> w : windows) {
                w.onCompleted();
            }
            windows.clear();

            done = true;
            drain();
        }

        Producer createProducer() {
            return new WindowOverlapProducer();
        }

        @Override
        public void call() {
            if (wip.decrementAndGet() == 0) {
                unsubscribe();
            }
        }

        void drain() {
            AtomicInteger dw = drainWip;
            if (dw.getAndIncrement() != 0) {
                return;
            }

            final Subscriber<? super Subject<T, T>> a = actual;
            final Queue<Subject<T, T>> q = queue;

            int missed = 1;

            for (;;) {

                long r = requested.get();
                long e = 0L;

                while (e != r) {
                    boolean d = done;
                    Subject<T, T> v = q.poll();
                    boolean empty = v == null;

                    if (checkTerminated(d, empty, a, q)) {
                        return;
                    }

                    if (empty) {
                        break;
                    }

                    a.onNext(v);

                    e++;
                }

                if (e == r) {
                    if (checkTerminated(done, q.isEmpty(), a, q)) {
                        return;
                    }
                }

                if (e != 0 && r != Long.MAX_VALUE) {
                    requested.addAndGet(-e);
                }

                missed = dw.addAndGet(-missed);
                if (missed == 0) {
                    break;
                }
            }
        }

        boolean checkTerminated(boolean d, boolean empty, Subscriber<? super Subject<T, T>> a, Queue<Subject<T, T>> q) {
            if (a.isUnsubscribed()) {
                q.clear();
                return true;
            }
            if (d) {
                Throwable e = error;
                if (e != null) {
                    q.clear();
                    a.onError(e);
                    return true;
                } else
                if (empty) {
                    a.onCompleted();
                    return true;
                }
            }
            return false;
        }

        final class WindowOverlapProducer extends AtomicBoolean implements Producer {
            /** */
            private static final long serialVersionUID = 4625807964358024108L;

            @Override
            public void request(long n) {
                if (n < 0L) {
                    throw new IllegalArgumentException("n >= 0 required but it was " + n);
                }
                if (n != 0L) {

                    WindowOverlap<T> parent = WindowOverlap.this;

                    if (!get() && compareAndSet(false, true)) {
                        long u = BackpressureUtils.multiplyCap(parent.skip, n - 1);
                        long v = BackpressureUtils.addCap(u, parent.size);

                        parent.request(v);
                    } else {
                        long u = BackpressureUtils.multiplyCap(parent.skip, n);
                        WindowOverlap.this.request(u);
                    }

                    BackpressureUtils.getAndAddRequest(parent.requested, n);
                    parent.drain();
                }
            }
        }
    }

}
