/**
 * Copyright 2013 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 rx.Notification;
import rx.Observer;
import rx.Scheduler;
import rx.concurrency.Schedulers;
import rx.util.functions.Action0;

/* package */class ScheduledObserver<T> implements Observer<T> {
    private final Observer<T> underlying;
    private final Scheduler scheduler;

    public ScheduledObserver(Observer<T> underlying, Scheduler scheduler) {
        this.underlying = underlying;
        this.scheduler = scheduler;
    }

    @Override
    public void onCompleted() {
        enqueue(new Notification<T>());
    }

    @Override
    public void onError(final Exception e) {
        enqueue(new Notification<T>(e));
    }

    @Override
    public void onNext(final T v) {
        enqueue(new Notification<T>(v));
    }

    private void enqueue(final Notification<T> notification) {

        Schedulers.currentThread().schedule(new Action0() {
            @Override
            public void call() {

                scheduler.schedule(new Action0() {
                    @Override
                    public void call() {
                        switch (notification.getKind()) {
                        case OnNext:
                            underlying.onNext(notification.getValue());
                            break;
                        case OnError:
                            underlying.onError(notification.getException());
                            break;
                        case OnCompleted:
                            underlying.onCompleted();
                            break;
                        default:
                            throw new IllegalStateException("Unknown kind of notification " + notification);

                        }
                    }
                });
            }

        });
    };

}

@@ -613,16 +613,118 @@
 ators;%0A%0A
+import java.util.concurrent.ConcurrentLinkedQueue;%0Aimport java.util.concurrent.atomic.AtomicInteger;%0A%0A
 import r
@@ -784,42 +784,8 @@
 er;%0A
-import rx.concurrency.Schedulers;%0A
 impo
@@ -962,16 +962,192 @@
 duler;%0A%0A
+    private final ConcurrentLinkedQueue%3CNotification%3CT%3E%3E queue = new ConcurrentLinkedQueue%3CNotification%3CT%3E%3E();%0A    private final AtomicInteger counter = new AtomicInteger(0);%0A%0A
     publ
@@ -1540,17 +1540,20 @@
 final T 
-v
+args
 ) %7B%0A    
@@ -1584,17 +1584,20 @@
 tion%3CT%3E(
-v
+args
 ));%0A    
@@ -1624,22 +1624,16 @@
 enqueue(
-final 
 Notifica
@@ -1656,25 +1656,24 @@
 tion) %7B%0A
-%0A
         
 Schedule
@@ -1668,124 +1668,497 @@
     
-Schedulers.currentThread().schedule(new Action0() %7B%0A            @Override%0A            public void call() %7B%0A%0A        
+// this must happen before 'counter' is used to provide synchronization between threads%0A        queue.offer(notification);%0A%0A        // we now use counter to atomically determine if we need to start processing or not%0A        // it will be 0 if it's the first notification or the scheduler has finished processing work%0A        // and we need to start doing it again%0A        if (counter.getAndIncrement() == 0) %7B%0A            processQueue();%0A        %7D%0A    %7D%0A%0A    private void processQueue() %7B%0A
     
@@ -2192,32 +2192,24 @@
 Action0() %7B%0A
-        
             
@@ -2230,24 +2230,16 @@
         
-        
 public v
@@ -2251,32 +2251,77 @@
 all() %7B%0A        
+        Notification%3CT%3E not = queue.poll();%0A%0A
                 
@@ -2327,33 +2327,24 @@
  switch (not
-ification
 .getKind()) 
@@ -2357,32 +2357,24 @@
             
-    
-    
 case OnNext:
@@ -2374,24 +2374,16 @@
 OnNext:%0A
-        
         
@@ -2411,33 +2411,24 @@
 g.onNext(not
-ification
 .getValue())
@@ -2449,39 +2449,23 @@
         
-      
-  
 break;%0A
-        
         
@@ -2502,32 +2502,24 @@
             
-   
-     
 underlying.o
@@ -2528,25 +2528,16 @@
 rror(not
-ification
 .getExce
@@ -2562,32 +2562,24 @@
             
-        
 break;%0A     
@@ -2585,32 +2585,24 @@
             
-    
-    
 case OnCompl
@@ -2607,24 +2607,16 @@
 pleted:%0A
-        
         
@@ -2669,32 +2669,24 @@
             
- 
-       
 break;%0A     
@@ -2696,24 +2696,16 @@
         
-     
-   
 default:
@@ -2705,24 +2705,16 @@
 efault:%0A
-        
         
@@ -2798,68 +2798,263 @@
  not
-ification);%0A%0A                        %7D%0A                    %7D
+);%0A%0A                %7D%0A%0A                // decrement count and if we still have work to do%0A                // recursively schedule ourselves to process again%0A                if (counter.decrementAndGet() %3E 0) %7B%0A                    scheduler.schedule(this);
 %0A   
@@ -3063,26 +3063,25 @@
            %7D
-);
+%0A
 %0A           
@@ -3075,33 +3075,32 @@
 %0A%0A            %7D%0A
-%0A
         %7D);%0A    
@@ -3100,13 +3100,11 @@
 );%0A    %7D
-;%0A
 %0A%7D%0A

Fix concurrency bug in ScheduledObserver