package com.github.davidmoten.rx2.internal.flowable;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import com.github.davidmoten.guavamini.Preconditions;

import io.reactivex.Flowable;
import io.reactivex.FlowableSubscriber;
import io.reactivex.internal.fuseable.SimplePlainQueue;
import io.reactivex.internal.queue.SpscLinkedArrayQueue;
import io.reactivex.internal.subscriptions.SubscriptionHelper;
import io.reactivex.internal.util.BackpressureHelper;

public final class FlowableStringSplitSimple extends Flowable<String> {

    private final Flowable<String> source;
    private final String delimiter;

    public FlowableStringSplitSimple(Flowable<String> source, String delimiter) {
        Preconditions.checkNotNull(source);
        Preconditions.checkNotNull(delimiter);
        Preconditions.checkArgument(delimiter.length() > 0);
        this.source = source;
        this.delimiter = delimiter;
    }

    @Override
    protected void subscribeActual(Subscriber<? super String> s) {
        source.subscribe(new StringSplitSubscriber(s, delimiter));
    }

    @SuppressWarnings("serial")
    private static final class StringSplitSubscriber extends AtomicLong
            implements FlowableSubscriber<String>, Subscription {

        private final Subscriber<? super String> actual;
        // queue of notifications
        private final transient SimplePlainQueue<String> queue = new SpscLinkedArrayQueue<String>(
                16);
        private final AtomicInteger wip = new AtomicInteger();
        private final AtomicBoolean once = new AtomicBoolean();

        private final DelimitedStringLinkedList ss;
        private volatile boolean cancelled;
        private Subscription parent;
        private boolean unbounded;

        private Throwable error;
        private volatile boolean done;

        StringSplitSubscriber(Subscriber<? super String> actual, String delimiter) {
            this.actual = actual;
            this.ss = new DelimitedStringLinkedList(delimiter);
        }

        @Override
        public void onSubscribe(Subscription subscription) {
            this.parent = subscription;
            actual.onSubscribe(this);
        }

        @Override
        public void cancel() {
            cancelled = true;
            parent.cancel();
        }

        @Override
        public void request(long n) {
            if (SubscriptionHelper.validate(n)) {
                BackpressureHelper.add(this, n);
                if (once.compareAndSet(false, true)) {
                    if (n == Long.MAX_VALUE) {
                        parent.request(Long.MAX_VALUE);
                        unbounded = true;
                    } else {
                        parent.request(1);
                    }
                }
                drain();
            }
        }

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

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

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

        private void drain() {
            if (wip.getAndIncrement() != 0) {
                return;
            }
            int missed = 1;
            while (true) {
                long r = get(); // requested
                long e = 0; // emitted
                while (e != r) {
                    if (cancelled) {
                        return;
                    }
                    if (find()) {
                        e++;
                    } else {
                        // must read `done` before poll occurs
                        boolean d = done;
                        String t = queue.poll();
                        if (t == null) {
                            if (d) {
                                Throwable err = error;
                                if (err != null) {
                                    ss.clear();
                                    actual.onError(err);
                                    return;
                                } else {
                                    String remaining = ss.remaining();
                                    final boolean checkCancelled;
                                    if (remaining != null) {
                                        ss.clear();
                                        queue.clear();
                                        actual.onNext(remaining);
                                        e++;
                                        checkCancelled = true;
                                    } else if (ss.addCalled()) {
                                        ss.clear();
                                        queue.clear();
                                        actual.onNext("");
                                        e++;
                                        checkCancelled = true;
                                    } else {
                                        checkCancelled = false;
                                    }
                                    if (!checkCancelled || !cancelled) {
                                        actual.onComplete();
                                    }
                                    return;
                                }
                            } else if (!unbounded) {
                                parent.request(1);
                            }
                            break;
                        } else {
                            ss.add(t);
                        }
                    }
                }
                if (e > 0 && r != Long.MAX_VALUE) {
                    this.addAndGet(-e);
                }
                missed = wip.addAndGet(-missed);
                if (missed == 0) {
                    return;
                }
            }
        }

        /**
         * Returns true if and only if a value emitted.
         * 
         * @return true if and only if a value emitted
         */
        private boolean find() {
            if (ss == null) {
                return false;
            }
            String s = ss.next();
            if (s != null) {
                actual.onNext(s);
                return true;
            } else {
                return false;
            }
        }
    }

}
