package com.github.davidmoten.rx.util;

import java.util.Queue;

import rx.Subscriber;
import rx.exceptions.MissingBackpressureException;
import rx.internal.operators.NotificationLite;

/**
 * Optimized for when request method is called on same thread as the Observer
 * methods.
 * 
 * @param <T>
 *            type of the items being queued and emitted by this drainer
 */
public class DrainerSyncBiased<T> implements Drainer<T> {

    private final NotificationLite<Object> on = NotificationLite.instance();
    private final Queue<T> queue;
    private final Subscriber<? super T> child;

    public static <T> DrainerSyncBiased<T> create(Queue<T> queue, Subscriber<? super T> child) {
        return new DrainerSyncBiased<T>(queue, child);
    }

    private DrainerSyncBiased(Queue<T> queue, Subscriber<? super T> child) {
        this.queue = queue;
        this.child = child;
    }

    private long expected;
    private boolean busy;
    private boolean finished;
    private Throwable error;
    private long counter;
    // this is the value we take off any new request so that only what is
    // required is requested of upstream
    private long total;// totalRequested - currentExpected - numQueued -
                       // numEmitted

    @Override
    public void request(long n) {
        if (n <= 0)
            return;
        synchronized (this) {
            long expectedBefore = expected;
            expected += n;
            if (expected < 0) {
                expected = Long.MAX_VALUE;
            }
            total += expected - expectedBefore;
            if (busy) {
                counter++;
                return;
            } else
                busy = true;
        }
        try {
            drain();
        } finally {
            synchronized (this) {
                busy = false;
            }
        }

    }

    @Override
    public void onCompleted() {
        synchronized (this) {
            finished = true;
            if (busy) {
                counter++;
                return;
            } else
                busy = true;
        }
        try {
            drain();
        } finally {
            synchronized (this) {
                busy = false;
            }
        }
    }

    @Override
    public void onError(Throwable e) {
        synchronized (this) {
            error = e;
            finished = true;
            if (busy) {
                counter++;
                return;
            } else
                busy = true;
        }
        try {
            drain();
        } finally {
            synchronized (this) {
                busy = false;
            }
        }
    }

    @Override
    public void onNext(T t) {
        if (!queue.offer(t)) {
            // this should only happen when the queue is bounded
            onError(new MissingBackpressureException());
        } else {
            synchronized (this) {
                total--;
            }
            drain();
        }
    }

    @SuppressWarnings("unchecked")
    private void drain() {
        long emittedTotal = 0;
        long r;
        synchronized (this) {
            r = expected;
        }
        while (true) {
            long emitted = 0;
            synchronized (this) {
                counter = 1;
            }
            while (!child.isUnsubscribed()) {
                boolean isFinished;
                synchronized (this) {
                    isFinished = finished;
                }
                if (isFinished) {
                    Throwable error;
                    synchronized (this) {
                        error = this.error;
                    }
                    if (error != null) {
                        // errors shortcut the queue so
                        // release the elements in the queue for gc
                        queue.clear();
                        child.onError(error);
                        return;
                    } else if (queue.isEmpty()) {
                        child.onCompleted();
                        return;
                    }
                }
                if (r > 0) {
                    Object o = queue.poll();
                    if (o != null) {
                        child.onNext((T) on.getValue(o));
                        synchronized (this) {
                            total--;
                        }
                        r--;
                        emitted++;
                    } else {
                        break;
                    }
                } else {
                    break;
                }
            }
            if (emitted > 0) {
                emittedTotal += emitted;
                // interested in initial request being Long.MAX_VALUE rather
                // than accumulated requests reaching Long.MAX_VALUE so is fine
                // just to test the value of `r` instead of `requested`.
                if (r != Long.MAX_VALUE) {
                    synchronized (this) {
                        expected -= emitted;
                        total -= emitted;
                        r = expected;
                    }
                }
            } else {
                synchronized (this) {
                    if (--counter == 0) {
                        break;
                    } else {
                        // update r for the next time through the loop
                        r = expected;
                    }
                }
            }
        }
        if (emittedTotal > 0) {
            // producer.request(emittedTotal);
        }
    }

    @Override
    public long total() {
        synchronized (this) {
            return total;
        }
    }

}

@@ -1519,32 +1519,65 @@
 ;%0A            %7D%0A
+            if (total %3E 0) %7B%0A    
             tota
@@ -1608,16 +1608,322 @@
 Before;%0A
+                if (total %3C 0)%0A                    total = Long.MAX_VALUE;%0A            %7D else %7B%0A                // cannot overflow if total is negative because expected -%0A                // expectedBefore cannot be more than Long.MAX_VALUE%0A                total += expected - expectedBefore;%0A            %7D%0A
         
@@ -3395,39 +3395,8 @@
 ) %7B%0A
-        long emittedTotal = 0;%0A
     
@@ -4955,49 +4955,8 @@
 ) %7B%0A
-                emittedTotal += emitted;%0A
     
@@ -5315,50 +5315,8 @@
 ed;%0A
-                        total -= emitted;%0A
     
@@ -5691,32 +5691,32 @@
               %7D%0A
+
             %7D%0A  
@@ -5727,97 +5727,8 @@
   %7D%0A
-        if (emittedTotal %3E 0) %7B%0A            // producer.request(emittedTotal);%0A        %7D%0A
     

add comments and remove unused code