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.internal.fuseable.SimpleQueue;
import io.reactivex.internal.queue.MpscLinkedQueue;
import io.reactivex.internal.subscriptions.SubscriptionHelper;
import io.reactivex.internal.util.BackpressureHelper;
import io.reactivex.internal.util.NotificationLite;

public final class FlowableStringSplit extends Flowable<String> {

    private final Flowable<String> source;
    private final String searchFor;
    private final int bufferSize;

    public FlowableStringSplit(Flowable<String> source, String searchFor, int bufferSize) {
        Preconditions.checkNotNull(source);
        Preconditions.checkNotNull(searchFor);
        Preconditions.checkArgument(bufferSize > 0);
        this.source = source;
        this.searchFor = searchFor;
        this.bufferSize = bufferSize;
    }

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

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

        private final Subscriber<? super String> actual;
        private final String searchFor;
        private final int bufferSize;
        // queue of notifications
        private final SimpleQueue<Object> queue = new MpscLinkedQueue<Object>();
        private final AtomicInteger wip = new AtomicInteger();
        private final AtomicBoolean once = new AtomicBoolean();

        private volatile boolean cancelled;

        private StringBuilder leftOver;
        private int index;
        private int searchIndex;
        private Subscription parent;
        private boolean unbounded;

        StringSplitSubscriber(Subscriber<? super String> actual, String searchFor, int bufferSize) {
            this.actual = actual;
            this.searchFor = searchFor;
            this.bufferSize = bufferSize;
        }

        @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(128);
                    }
                }
                drain();
            }
        }

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

        @Override
        public void onComplete() {
            queue.offer(NotificationLite.complete());
            drain();
        }

        @Override
        public void onError(Throwable e) {
            queue.offer(NotificationLite.error(e));
            drain();
        }

        private void drain() {
            if (wip.getAndIncrement() != 0) {
                return;
            }
            int missed = wip.get();
            long r = get(); // requested
            while (true) {
                long e = 0; // emitted
                while (e != r) {
                    if (cancelled) {
                        return;
                    }
                    if (find()) {
                        e++;
                    } else {
                        Object o = null;
                        try {
                            o = queue.poll();
                        } catch (Exception e1) {
                            o = NotificationLite.error(e1);
                        }
                        if (o == null) {
                            if (!unbounded) {
                                parent.request(1);
                            }
                            break;
                        } else if (NotificationLite.isComplete(o)) {
                            if (leftOver != null) {
                                String s = leftOver.substring(index, leftOver.length());
                                leftOver = null;
                                queue.clear();
                                actual.onNext(s.toString());
                                e++;
                            }
                            actual.onComplete();
                            return;
                        } else if (NotificationLite.isError(o)) {
                            leftOver = null;
                            queue.clear();
                            actual.onError(NotificationLite.getError(o));
                            return;
                        } else {
                            if (leftOver == null) {
                                leftOver = new StringBuilder(bufferSize);
                            }
                            leftOver.append((String) o);
                        }
                    }
                }
                if (e > 0) {
                    r = BackpressureHelper.produced(this, e);
                    if (r == 0) {
                        if (wip.addAndGet(-missed) == 0) {
                            return;
                        }
                        // otherwise loop again and make sure to refresh r
                        // r = get();
                    }
                } else if (wip.addAndGet(-missed) == 0) {
                    return;
                } else {
                    r = get();
                }
            }
        }

        /**
         * Returns true if and only if a value emitted.
         * 
         * @return true if and only if a value emitted
         */
        private boolean find() {
            if (leftOver == null) {
                return false;
            } else {
                int i = leftOver.indexOf(searchFor, searchIndex);
                if (i != -1) {
                    // emit and adjust indexes
                    String s = leftOver.substring(searchIndex, i);
                    updateIndexes(i);
                    actual.onNext(s);
                    return true;
                } else {
                    // emit nothing but adjust searchIndex to the right
                    searchIndex = Math.max(searchIndex, leftOver.length() - searchFor.length() - 1);
                    return false;
                }
            }
        }

        private void updateIndexes(int i) {
            searchIndex = i + searchFor.length();
            if (searchIndex > bufferSize - bufferSize/4) {
                // shrink leftOver
                leftOver.delete(0, searchIndex);
                index = 0;
                searchIndex = 0;
            } else {
                index = searchIndex;
            }
        }
    }

}

@@ -3804,17 +3804,9 @@
 d = 
-wip.get()
+1
 ;%0D%0A 
@@ -5834,25 +5834,9 @@
     
-    if (r == 0) %7B
+%7D
 %0D%0A  
@@ -5845,36 +5845,33 @@
             
-        if (
+missed = 
 wip.addAndGe
@@ -5884,269 +5884,38 @@
 sed)
- == 0) %7B%0D%0A                            return;%0D%0A                        %7D%0D%0A                        // otherwise loop again and make sure to refresh r%0D%0A                        // r = get();%0D%0A                    %7D%0D%0A                %7D else if (wip.addAndGet(-
+;%0D%0A                if ( 
 missed
-)
  == 
@@ -5962,37 +5962,26 @@
            %7D
- else %7B%0D%0A    
+%0D%0A
             
@@ -5996,35 +5996,16 @@
 get();%0D%0A
-                %7D%0D%0A
         
@@ -7059,17 +7059,19 @@
 fferSize
-/
+ / 
 4) %7B%0D%0A  

simplify missed