/*
 * Copyright 2016-present David Karnok
 *
 * 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 hu.akarnokd.rxjava3.operators;

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

import org.reactivestreams.*;

import io.reactivex.rxjava3.exceptions.Exceptions;
import io.reactivex.rxjava3.operators.SimpleQueue;
import io.reactivex.rxjava3.internal.subscribers.*;
import io.reactivex.rxjava3.internal.subscriptions.*;
import io.reactivex.rxjava3.internal.util.*;
import io.reactivex.rxjava3.parallel.ParallelFlowable;

/**
 * Subscription coordinator that merges items from a fixed set of source Publishers by
 * picking the smallest available next item from them based on a Comparator.
 *
 * @param <T> the value type
 * @since 0.17.9
 */
public final class BasicMergeSubscription<T>
extends AtomicInteger
implements Subscription, InnerQueuedSubscriberSupport<T> {
    private static final long serialVersionUID = -8467324377226330554L;

    final Subscriber<? super T> downstream;

    final Comparator<? super T> comparator;

    final InnerQueuedSubscriber<T>[] subscribers;

    final boolean delayErrors;

    final AtomicThrowable errors;

    final AtomicLong requested;

    final Object[] latest;

    volatile boolean cancelled;

    @SuppressWarnings("unchecked")
    public BasicMergeSubscription(Subscriber<? super T> downstream, Comparator<? super T> comparator, int n, int prefetch, boolean delayErrors) {
        this.downstream = downstream;
        this.comparator = comparator;
        this.delayErrors = delayErrors;
        InnerQueuedSubscriber<T>[] subs = new InnerQueuedSubscriber[n];
        for (int i = 0; i < n; i++) {
            subs[i] = new InnerQueuedSubscriber<>(this, prefetch);
        }
        this.subscribers = subs;
        this.requested = new AtomicLong();
        this.errors = new AtomicThrowable();
        this.latest = new Object[n];
    }

    public void subscribe(Publisher<T>[] sources, int n) {
      InnerQueuedSubscriber<T>[] subs = subscribers;
      for (int i = 0; i < n && !cancelled; i++) {
          Publisher<T> p = sources[i];
          if (p != null) {
              p.subscribe(subs[i]);
          } else {
              EmptySubscription.error(new NullPointerException("The " + i + "th source is null"), subs[i]);
              if (!delayErrors) {
                  break;
              }
          }
      }
    }

    public void subscribe(ParallelFlowable<T> source) {
      source.subscribe(subscribers);
    }

    @Override
    public void request(long n) {
        if (SubscriptionHelper.validate(n)) {
            BackpressureHelper.add(requested, n);
            drain();
        }
    }

    void cancelSources() {
        for (InnerQueuedSubscriber<T> d : subscribers) {
            d.cancel();
        }
    }

    void clearSources() {
        Arrays.fill(latest, this);
        for (InnerQueuedSubscriber<T> d : subscribers) {
            SimpleQueue<T> q = d.queue();
            if (q != null) {
                q.clear();
            }
        }
    }

    void cancelAndClearSources() {
        Arrays.fill(latest, this);
        for (InnerQueuedSubscriber<T> d : subscribers) {
            d.cancel();
            SimpleQueue<T> q = d.queue();
            if (q != null) {
                q.clear();
            }
        }
    }

    @Override
    public void cancel() {
        if (!cancelled) {
            cancelled = true;
            cancelSources();
            errors.tryTerminateAndReport();
            if (getAndIncrement() == 0) {
                clearSources();
            }
        }
    }

    @Override
    public void innerNext(InnerQueuedSubscriber<T> inner, T value) {
        inner.queue().offer(value);
        drain();
    }

    @Override
    public void innerError(InnerQueuedSubscriber<T> inner, Throwable e) {
        if (errors.tryAddThrowableOrReport(e)) {
            if (!delayErrors) {
                cancelSources();
            } else {
                inner.setDone();
            }
            drain();
        }
    }

    @Override
    public void innerComplete(InnerQueuedSubscriber<T> inner) {
        inner.setDone();
        drain();
    }

    @SuppressWarnings("unchecked")
    @Override
    public void drain() {
        if (getAndIncrement() != 0) {
            return;
        }

        int missed = 1;

        Subscriber<? super T> a = downstream;
        AtomicThrowable err = errors;
        InnerQueuedSubscriber<T>[] subs = subscribers;
        int n = subs.length;
        Object[] latest = this.latest;
        Comparator<? super T> comp = comparator;

        for (;;) {

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

            while (e != r) {
                if (cancelled) {
                    clearSources();
                    return;
                }

                if (!delayErrors && err.get() != null) {
                    cancelAndClearSources();
                    a.onError(err.terminate());
                    return;
                }

                boolean d = true;
                int hasValue = 0;
                boolean empty = true;

                T smallest = null;
                int pick = -1;

                for (int i = 0; i < n; i++) {
                    InnerQueuedSubscriber<T> inner = subs[i];
                    boolean innerDone = inner.isDone();
                    if (!innerDone) {
                        d = false;
                    }
                    Object v = latest[i];
                    if (v == null) {
                        SimpleQueue<T> q = inner.queue();
                        try {
                            v = q != null ? q.poll() : null;
                        } catch (Throwable ex) {
                            Exceptions.throwIfFatal(ex);
                            err.tryAddThrowableOrReport(ex);
                            inner.setDone();
                            if (!delayErrors) {
                                cancelAndClearSources();
                                err.tryTerminateConsumer(a);
                                return;
                            }
                            v = this;
                        }

                        if (v != null) {
                            latest[i] = v;
                            hasValue++;
                            empty = false;
                        } else
                        if (innerDone) {
                            latest[i] = this;
                            hasValue++;
                        }
                    } else {
                        hasValue++;
                        if (v != this) {
                            empty = false;
                        }
                    }

                    if (v != null && v != this) {
                        boolean smaller;
                        try {
                            if (smallest != null) {
                                smaller = comp.compare(smallest, (T)v) > 0;
                            } else {
                                smaller = true;
                            }
                        } catch (Throwable ex) {
                            Exceptions.throwIfFatal(ex);
                            err.tryAddThrowableOrReport(ex);
                            cancelAndClearSources();
                            a.onError(err.terminate());
                            return;
                        }
                        if (smaller) {
                            smallest = (T)v;
                            pick = i;
                        }
                    }
                }

                if (hasValue == n && pick >= 0) {
                    a.onNext(smallest);
                    latest[pick] = null;
                    subs[pick].request(1);

                    e++;
                } else {
                    if (d && empty) {
                        if (err.get() != null) {
                            a.onError(err.terminate());
                        } else {
                            a.onComplete();
                        }
                        return;
                    }
                    break;
                }
            }

            if (e == r) {
                if (cancelled) {
                    clearSources();
                    return;
                }

                if (!delayErrors && err.get() != null) {
                    cancelAndClearSources();
                    a.onError(err.terminate());
                    return;
                }

                boolean d = true;
                boolean empty = true;

                for (int i = 0; i < subs.length; i++) {
                    InnerQueuedSubscriber<T> inner = subs[i];
                    if (!inner.isDone()) {
                        d = false;
                        break;
                    }
                    Object o = latest[i];
                    SimpleQueue<T> q = inner.queue();
                    if (o == null && q != null) {
                        try {
                            o = q.poll();
                        } catch (Throwable ex) {
                            Exceptions.throwIfFatal(ex);
                            err.tryAddThrowableOrReport(ex);
                            if (!delayErrors) {
                                cancelAndClearSources();
                                err.tryTerminateConsumer(a);
                                return;
                            }
                            o = this;
                        }

                        latest[i] = o;
                    }
                    if (o != null && o != this) {
                        empty = false;
                        break;
                    }
                }

                if (d && empty) {
                    if (err.get() != null) {
                        a.onError(err.terminate());
                    } else {
                        a.onComplete();
                    }
                    return;
                }
            }

            if (e != 0L) {
                BackpressureHelper.produced(requested, e);
            }

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