/*
 * 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.observable;

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

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.core.Scheduler;
import io.reactivex.rxjava3.core.Scheduler.Worker;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.internal.disposables.*;
import io.reactivex.rxjava3.internal.queue.MpscLinkedQueue;
import io.reactivex.rxjava3.operators.SimplePlainQueue;
import io.reactivex.rxjava3.subjects.UnicastSubject;

public final class ObservableWindowTimed<T> extends AbstractObservableWithUpstream<T, Observable<T>> {
    final long timespan;
    final long timeskip;
    final TimeUnit unit;
    final Scheduler scheduler;
    final long maxSize;
    final int bufferSize;
    final boolean restartTimerOnMaxSize;

    public ObservableWindowTimed(Observable<T> source,
            long timespan, long timeskip, TimeUnit unit, Scheduler scheduler, long maxSize,
            int bufferSize, boolean restartTimerOnMaxSize) {
        super(source);
        this.timespan = timespan;
        this.timeskip = timeskip;
        this.unit = unit;
        this.scheduler = scheduler;
        this.maxSize = maxSize;
        this.bufferSize = bufferSize;
        this.restartTimerOnMaxSize = restartTimerOnMaxSize;
    }

    @Override
    protected void subscribeActual(Observer<? super Observable<T>> downstream) {
        if (timespan == timeskip) {
            if (maxSize == Long.MAX_VALUE) {
                source.subscribe(new WindowExactUnboundedObserver<>(
                        downstream,
                        timespan, unit, scheduler, bufferSize));
                return;
            }
            source.subscribe(new WindowExactBoundedObserver<>(
                        downstream,
                        timespan, unit, scheduler,
                        bufferSize, maxSize, restartTimerOnMaxSize));
            return;
        }
        source.subscribe(new WindowSkipObserver<>(downstream,
                timespan, timeskip, unit, scheduler.createWorker(), bufferSize));
    }

    abstract static class AbstractWindowObserver<T>
    extends AtomicInteger
    implements Observer<T>, Disposable {
        private static final long serialVersionUID = 5724293814035355511L;

        final Observer<? super Observable<T>> downstream;

        final SimplePlainQueue<Object> queue;

        final long timespan;
        final TimeUnit unit;
        final int bufferSize;

        long emitted;

        volatile boolean done;
        Throwable error;

        Disposable upstream;

        final AtomicBoolean downstreamCancelled;

        volatile boolean upstreamCancelled;

        final AtomicInteger windowCount;

        AbstractWindowObserver(Observer<? super Observable<T>> downstream, long timespan, TimeUnit unit, int bufferSize) {
            this.downstream = downstream;
            this.queue = new MpscLinkedQueue<>();
            this.timespan = timespan;
            this.unit = unit;
            this.bufferSize = bufferSize;
            this.downstreamCancelled = new AtomicBoolean();
            this.windowCount = new AtomicInteger(1);
        }

        @Override
        public final void onSubscribe(Disposable d) {
            if (DisposableHelper.validate(this.upstream, d)) {
                this.upstream = d;

                downstream.onSubscribe(this);

                createFirstWindow();
            }
        }

        abstract void createFirstWindow();

        @Override
        public final void onNext(T t) {
            queue.offer(t);
            drain();
        }

        @Override
        public final void onError(Throwable t) {
            error = t;
            done = true;
            drain();
        }

        @Override
        public final void onComplete() {
            done = true;
            drain();
        }

        @Override
        public final void dispose() {
            if (downstreamCancelled.compareAndSet(false, true)) {
                windowDone();
            }
        }

        @Override
        public final boolean isDisposed() {
            return downstreamCancelled.get();
        }

        final void windowDone() {
            if (windowCount.decrementAndGet() == 0) {
                cleanupResources();
                upstream.dispose();
                upstreamCancelled = true;
                drain();
            }
        }

        abstract void cleanupResources();

        abstract void drain();
    }

    static final class WindowExactUnboundedObserver<T>
            extends AbstractWindowObserver<T>
            implements Runnable {

        private static final long serialVersionUID = 1155822639622580836L;

        final Scheduler scheduler;

        UnicastSubject<T> window;

        final SequentialDisposable timer;

        static final Object NEXT_WINDOW = new Object();

        final Runnable windowRunnable;

        WindowExactUnboundedObserver(Observer<? super Observable<T>> actual, long timespan, TimeUnit unit,
                Scheduler scheduler, int bufferSize) {
            super(actual, timespan, unit, bufferSize);
            this.scheduler = scheduler;
            this.timer = new SequentialDisposable();
            this.windowRunnable = new WindowRunnable();
        }

        @Override
        void createFirstWindow() {
            if (!downstreamCancelled.get()) {
                windowCount.getAndIncrement();
                window = UnicastSubject.create(bufferSize, windowRunnable);

                emitted = 1;

                ObservableWindowSubscribeIntercept<T> intercept = new ObservableWindowSubscribeIntercept<>(window);
                downstream.onNext(intercept);

                timer.replace(scheduler.schedulePeriodicallyDirect(this, timespan, timespan, unit));

                if (intercept.tryAbandon()) {
                    window.onComplete();
                }
            }
        }

        @Override
        public void run() {
            queue.offer(NEXT_WINDOW);
            drain();
        }

        @Override
        void drain() {
            if (getAndIncrement() != 0) {
                return;
            }

            final SimplePlainQueue<Object> queue = this.queue;
            final Observer<? super Observable<T>> downstream = this.downstream;
            UnicastSubject<T> window = this.window;

            int missed = 1;
            for (;;) {

                if (upstreamCancelled) {
                    queue.clear();
                    window = null;
                    this.window = null;
                } else {
                    boolean isDone = done;
                    Object o = queue.poll();
                    boolean isEmpty = o == null;

                    if (isDone && isEmpty) {
                        Throwable ex = error;
                        if (ex != null) {
                            if (window != null) {
                                window.onError(ex);
                            }
                            downstream.onError(ex);
                        } else {
                            if (window != null) {
                                window.onComplete();
                            }
                            downstream.onComplete();
                        }
                        cleanupResources();
                        upstreamCancelled = true;
                        continue;
                    }
                    else if (!isEmpty) {

                        if (o == NEXT_WINDOW) {
                            if (window != null) {
                                window.onComplete();
                                window = null;
                                this.window = null;
                            }
                            if (downstreamCancelled.get()) {
                                timer.dispose();
                            } else {
                                emitted++;

                                windowCount.getAndIncrement();
                                window = UnicastSubject.create(bufferSize, windowRunnable);
                                this.window = window;

                                ObservableWindowSubscribeIntercept<T> intercept = new ObservableWindowSubscribeIntercept<>(window);
                                downstream.onNext(intercept);

                                if (intercept.tryAbandon()) {
                                    window.onComplete();
                                }
                            }
                        } else if (window != null) {
                            @SuppressWarnings("unchecked")
                            T item = (T)o;
                            window.onNext(item);
                        }

                        continue;
                    }
                }

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

        @Override
        void cleanupResources() {
            timer.dispose();
        }

        final class WindowRunnable implements Runnable {
            @Override
            public void run() {
                windowDone();
            }
        }
    }

    static final class WindowExactBoundedObserver<T>
    extends AbstractWindowObserver<T>
    implements Runnable {
        private static final long serialVersionUID = -6130475889925953722L;

        final Scheduler scheduler;
        final boolean restartTimerOnMaxSize;
        final long maxSize;
        final Scheduler.Worker worker;

        long count;

        UnicastSubject<T> window;

        final SequentialDisposable timer;

        WindowExactBoundedObserver(
                Observer<? super Observable<T>> actual,
                long timespan, TimeUnit unit, Scheduler scheduler,
                int bufferSize, long maxSize, boolean restartTimerOnMaxSize) {
            super(actual, timespan, unit, bufferSize);
            this.scheduler = scheduler;
            this.maxSize = maxSize;
            this.restartTimerOnMaxSize = restartTimerOnMaxSize;
            if (restartTimerOnMaxSize) {
                worker = scheduler.createWorker();
            } else {
                worker = null;
            }
            this.timer = new SequentialDisposable();
        }

        @Override
        void createFirstWindow() {
            if (!downstreamCancelled.get()) {
                emitted = 1;

                windowCount.getAndIncrement();
                window = UnicastSubject.create(bufferSize, this);

                ObservableWindowSubscribeIntercept<T> intercept = new ObservableWindowSubscribeIntercept<>(window);
                downstream.onNext(intercept);

                Runnable boundaryTask = new WindowBoundaryRunnable(this, 1L);
                if (restartTimerOnMaxSize) {
                    timer.replace(worker.schedulePeriodically(boundaryTask, timespan, timespan, unit));
                } else {
                    timer.replace(scheduler.schedulePeriodicallyDirect(boundaryTask, timespan, timespan, unit));
                }

                if (intercept.tryAbandon()) {
                    window.onComplete();
                }
            }
        }

        @Override
        public void run() {
            windowDone();
        }

        @Override
        void cleanupResources() {
            timer.dispose();
            Worker w = worker;
            if (w != null) {
                w.dispose();
            }
        }

        void boundary(WindowBoundaryRunnable sender) {
            queue.offer(sender);
            drain();
        }

        @Override
        void drain() {
            if (getAndIncrement() != 0) {
                return;
            }

            int missed = 1;
            final SimplePlainQueue<Object> queue = this.queue;
            final Observer<? super Observable<T>> downstream = this.downstream;
            UnicastSubject<T> window = this.window;

            for (;;) {

                if (upstreamCancelled) {
                    queue.clear();
                    window = null;
                    this.window = null;
                } else {

                    boolean isDone = done;
                    Object o = queue.poll();
                    boolean isEmpty = o == null;

                    if (isDone && isEmpty) {
                        Throwable ex = error;
                        if (ex != null) {
                            if (window != null) {
                                window.onError(ex);
                            }
                            downstream.onError(ex);
                        } else {
                            if (window != null) {
                                window.onComplete();
                            }
                            downstream.onComplete();
                        }
                        cleanupResources();
                        upstreamCancelled = true;
                        continue;
                    } else if (!isEmpty) {
                        if (o instanceof WindowBoundaryRunnable) {
                            WindowBoundaryRunnable boundary = (WindowBoundaryRunnable) o;
                            if (boundary.index == emitted || !restartTimerOnMaxSize) {
                                this.count = 0;
                                window = createNewWindow(window);
                            }
                        } else if (window != null) {
                            @SuppressWarnings("unchecked")
                            T item = (T)o;
                            window.onNext(item);

                            long count = this.count + 1;
                            if (count == maxSize) {
                                this.count = 0;
                                window = createNewWindow(window);
                            } else {
                                this.count = count;
                            }
                        }

                        continue;
                    }
                }

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

        UnicastSubject<T> createNewWindow(UnicastSubject<T> window) {
            if (window != null) {
                window.onComplete();
                window = null;
            }

            if (downstreamCancelled.get()) {
                cleanupResources();
            } else {
                long emitted = this.emitted;
                this.emitted = ++emitted;

                windowCount.getAndIncrement();
                window = UnicastSubject.create(bufferSize, this);
                this.window = window;

                ObservableWindowSubscribeIntercept<T> intercept = new ObservableWindowSubscribeIntercept<>(window);
                downstream.onNext(intercept);

                if (restartTimerOnMaxSize) {
                    timer.update(worker.schedulePeriodically(new WindowBoundaryRunnable(this, emitted), timespan, timespan, unit));
                }

                if (intercept.tryAbandon()) {
                    window.onComplete();
                }
            }

            return window;
        }

        static final class WindowBoundaryRunnable implements Runnable {

            final WindowExactBoundedObserver<?> parent;

            final long index;

            WindowBoundaryRunnable(WindowExactBoundedObserver<?> parent, long index) {
                this.parent = parent;
                this.index = index;
            }

            @Override
            public void run() {
                parent.boundary(this);
            }
        }
    }

    static final class WindowSkipObserver<T>
    extends AbstractWindowObserver<T>
    implements Runnable {
        private static final long serialVersionUID = -7852870764194095894L;

        final long timeskip;
        final Scheduler.Worker worker;

        final List<UnicastSubject<T>> windows;

        WindowSkipObserver(Observer<? super Observable<T>> actual,
                long timespan, long timeskip, TimeUnit unit,
                Worker worker, int bufferSize) {
            super(actual, timespan, unit, bufferSize);
            this.timeskip = timeskip;
            this.worker = worker;
            this.windows = new LinkedList<>();
        }

        @Override
        void createFirstWindow() {
            if (!downstreamCancelled.get()) {
                emitted = 1;

                windowCount.getAndIncrement();
                UnicastSubject<T> window = UnicastSubject.create(bufferSize, this);
                windows.add(window);

                ObservableWindowSubscribeIntercept<T> intercept = new ObservableWindowSubscribeIntercept<>(window);
                downstream.onNext(intercept);

                worker.schedule(new WindowBoundaryRunnable(this, false), timespan, unit);
                worker.schedulePeriodically(new WindowBoundaryRunnable(this, true), timeskip, timeskip, unit);

                if (intercept.tryAbandon()) {
                    window.onComplete();
                    windows.remove(window);
                }
            }
        }

        @Override
        void cleanupResources() {
            worker.dispose();
        }

        @Override
        void drain() {
            if (getAndIncrement() != 0) {
                return;
            }

            int missed = 1;
            final SimplePlainQueue<Object> queue = this.queue;
            final Observer<? super Observable<T>> downstream = this.downstream;
            final List<UnicastSubject<T>> windows = this.windows;

            for (;;) {
                if (upstreamCancelled) {
                    queue.clear();
                    windows.clear();
                } else {
                    boolean isDone = done;
                    Object o = queue.poll();
                    boolean isEmpty = o == null;

                    if (isDone && isEmpty) {
                        Throwable ex = error;
                        if (ex != null) {
                            for (UnicastSubject<T> window : windows) {
                                window.onError(ex);
                            }
                            downstream.onError(ex);
                        } else {
                            for (UnicastSubject<T> window : windows) {
                                window.onComplete();
                            }
                            downstream.onComplete();
                        }
                        cleanupResources();
                        upstreamCancelled = true;
                        continue;
                    } else if (!isEmpty) {
                        if (o == WINDOW_OPEN) {
                            if (!downstreamCancelled.get()) {
                                long emitted = this.emitted;
                                this.emitted = ++emitted;

                                windowCount.getAndIncrement();
                                UnicastSubject<T> window = UnicastSubject.create(bufferSize, this);
                                windows.add(window);

                                ObservableWindowSubscribeIntercept<T> intercept = new ObservableWindowSubscribeIntercept<>(window);
                                downstream.onNext(intercept);

                                worker.schedule(new WindowBoundaryRunnable(this, false), timespan, unit);

                                if (intercept.tryAbandon()) {
                                    window.onComplete();
                                }
                            }
                        } else if (o == WINDOW_CLOSE) {
                            if (!windows.isEmpty()) {
                                windows.remove(0).onComplete();
                            }
                        } else {
                            @SuppressWarnings("unchecked")
                            T item = (T)o;
                            for (UnicastSubject<T> window : windows) {
                                window.onNext(item);
                            }
                        }
                        continue;
                    }
                }
                missed = addAndGet(-missed);
                if (missed == 0) {
                    break;
                }
            }
        }

        @Override
        public void run() {
            windowDone();
        }

        void boundary(boolean isOpen) {
            queue.offer(isOpen ? WINDOW_OPEN : WINDOW_CLOSE);
            drain();
        }

        static final Object WINDOW_OPEN = new Object();
        static final Object WINDOW_CLOSE = new Object();

        static final class WindowBoundaryRunnable implements Runnable {

            final WindowSkipObserver<?> parent;

            final boolean isOpen;

            WindowBoundaryRunnable(WindowSkipObserver<?> parent, boolean isOpen) {
                this.parent = parent;
                this.isOpen = isOpen;
            }

            @Override
            public void run() {
                parent.boundary(isOpen);
            }
        }
    }

}
