/**
 * Copyright 2014 Netflix, Inc.
 * 
 * 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 rx.operators;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

import rx.Observable.Operator;
import rx.Scheduler;
import rx.Subscriber;
import rx.functions.Action0;
import rx.schedulers.ImmediateScheduler;
import rx.schedulers.TrampolineScheduler;

/**
 * Delivers events on the specified Scheduler asynchronously via an unbounded buffer.
 * 
 * <img width="640" src="https://github.com/Netflix/RxJava/wiki/images/rx-operators/observeOn.png">
 */
public class OperatorObserveOn<T> implements Operator<T, T> {

    private final Scheduler scheduler;

    /**
     * @param scheduler
     */
    public OperatorObserveOn(Scheduler scheduler) {
        this.scheduler = scheduler;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super T> child) {
        if (scheduler instanceof ImmediateScheduler) {
            // avoid overhead, execute directly
            return child;
        } else if (scheduler instanceof TrampolineScheduler) {
            // avoid overhead, execute directly
            return child;
        } else {
            return new ObserveOnSubscriber<T>(scheduler, child);
        }
    }

    /** Observe through individual queue per observer. */
    private static class ObserveOnSubscriber<T> extends Subscriber<T> {
        private final NotificationLite<T> on = NotificationLite.instance();
        final Subscriber<? super T> observer;
        private final Scheduler.Worker recursiveScheduler;

        private final ConcurrentLinkedQueue<Object> queue = new ConcurrentLinkedQueue<Object>();
        final AtomicLong counter = new AtomicLong(0);

        public ObserveOnSubscriber(Scheduler scheduler, Subscriber<? super T> subscriber) {
            super(subscriber);
            this.observer = subscriber;
            this.recursiveScheduler = scheduler.createWorker();
            subscriber.add(recursiveScheduler);
        }

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

        @Override
        public void onCompleted() {
            queue.offer(on.completed());
            schedule();
        }

        @Override
        public void onError(final Throwable e) {
            queue.offer(on.error(e));
            schedule();
        }

        protected void schedule() {
            if (counter.getAndIncrement() == 0) {
                recursiveScheduler.schedule(new Action0() {

                    @Override
                    public void call() {
                        pollQueue();
                    }

                });
            }
        }

        private void pollQueue() {
            do {
                Object v = queue.poll();
                on.accept(observer, v);
            } while (counter.decrementAndGet() > 0);
        }

    }

}
@@ -634,40 +634,40 @@
 til.
-concurrent.ConcurrentLinkedQueue
+ArrayList;%0Aimport java.util.List
 ;%0Aim
@@ -2115,85 +2115,37 @@
 ate 
-final ConcurrentLinkedQueue%3CObject%3E queue = new ConcurrentLinkedQueue%3CObject%3E
+FastList queue = new FastList
 ();%0A
@@ -2539,24 +2539,62 @@
 inal T t) %7B%0A
+            synchronized (this) %7B%0A    
             
@@ -2595,29 +2595,27 @@
       queue.
-offer
+add
 (on.next(t))
@@ -2612,24 +2612,38 @@
 n.next(t));%0A
+            %7D%0A
             
@@ -2715,24 +2715,62 @@
 mpleted() %7B%0A
+            synchronized (this) %7B%0A    
             
@@ -2771,29 +2771,27 @@
       queue.
-offer
+add
 (on.complete
@@ -2792,24 +2792,38 @@
 mpleted());%0A
+            %7D%0A
             
@@ -2908,24 +2908,62 @@
 owable e) %7B%0A
+            synchronized (this) %7B%0A    
             
@@ -2972,13 +2972,11 @@
 eue.
-offer
+add
 (on.
@@ -2982,24 +2982,38 @@
 .error(e));%0A
+            %7D%0A
             
@@ -3430,24 +3430,89 @@
     
-Object
+FastList vs;%0A                synchronized (this) %7B%0A                   
  v
+s
  = queue
 .pol
@@ -3511,125 +3511,962 @@
 ueue
-.poll();%0A                on.accept(observer, v);%0A            %7D while (counter.decrementAndGet() %3E 0);%0A        %7D%0A%0A    %7D%0A
+;%0A                    queue = new FastList();%0A                %7D%0A                for (Object v : vs.array) %7B%0A                    if (v == null) %7B%0A                        break;%0A                    %7D%0A                    on.accept(observer, v);%0A                %7D%0A                if (counter.addAndGet(-vs.size) == 0) %7B%0A                    break;%0A                %7D%0A            %7D while (true);%0A        %7D%0A%0A    %7D%0A%0A    static final class FastList %7B%0A        Object%5B%5D array;%0A        int size;%0A%0A        public void add(Object o) %7B%0A            int s = size;%0A            Object%5B%5D a = array;%0A            if (a == null) %7B%0A                a = new Object%5B16%5D;%0A                array = a;%0A            %7D else if (s == a.length) %7B%0A                Object%5B%5D array2 = new Object%5Bs + (s %3E%3E 2)%5D;%0A                System.arraycopy(a, 0, array2, 0, s);%0A                a = array2;%0A                array = a;%0A            %7D%0A            a%5Bs%5D = o;%0A            size = s + 1;%0A        %7D%0A    %7D
 %0A%7D

Change to batch dequeue.