Blocking implementation of ScheduledObserver

Author: mairbek

rxjava-core/src/main/java/rx/operators/OperationObserveOn.java

@@ -20,12 +20,12 @@
 
 import org.junit.Test;
 
+import org.mockito.InOrder;
 import rx.Observable;
 import rx.Observer;
 import rx.Scheduler;
 import rx.Subscription;
 import rx.concurrency.Schedulers;
-import rx.util.functions.Action0;
 import rx.util.functions.Func1;
 
 public class OperationObserveOn {
@@ -60,15 +60,32 @@ public void testObserveOn() {
             Observer<Integer> observer = mock(Observer.class);
             Observable.create(observeOn(Observable.toObservable(1, 2, 3), scheduler)).subscribe(observer);
 
-            verify(scheduler, times(4)).schedule(any(Action0.class));
-            verifyNoMoreInteractions(scheduler);
-
             verify(observer, times(1)).onNext(1);
             verify(observer, times(1)).onNext(2);
             verify(observer, times(1)).onNext(3);
             verify(observer, times(1)).onCompleted();
         }
 
+
+        @Test
+        @SuppressWarnings("unchecked")
+        public void testOrdering() {
+            Observable<String> obs = Observable.from("one", null, "two", "three", "four");
+
+            Observer<String> observer = mock(Observer.class);
+
+            InOrder inOrder = inOrder(observer);
+
+            obs.observeOn(Schedulers.threadPoolForComputation()).subscribe(observer);
+            inOrder.verify(observer, times(1)).onNext("one");
+            inOrder.verify(observer, times(1)).onNext(null);
+            inOrder.verify(observer, times(1)).onNext("two");
+            inOrder.verify(observer, times(1)).onNext("three");
+            inOrder.verify(observer, times(1)).onNext("four");
+            inOrder.verify(observer, times(1)).onCompleted();
+            inOrder.verifyNoMoreInteractions();
+        }
+
     }
 
 }

rxjava-core/src/main/java/rx/operators/ScheduledObserver.java

@@ -1,12 +1,12 @@
 /**
  * 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.
@@ -15,46 +15,83 @@
  */
 package rx.operators;
 
+import rx.Notification;
 import rx.Observer;
 import rx.Scheduler;
 import rx.util.functions.Action0;
 
+import java.util.concurrent.ConcurrentLinkedQueue;
+
 /* package */class ScheduledObserver<T> implements Observer<T> {
     private final Observer<T> underlying;
     private final Scheduler scheduler;
 
+    private final ConcurrentLinkedQueue<Notification<T>> queue = new ConcurrentLinkedQueue<Notification<T>>();
+    private final Object lock = new Object();
+
     public ScheduledObserver(Observer<T> underlying, Scheduler scheduler) {
         this.underlying = underlying;
         this.scheduler = scheduler;
     }
 
     @Override
     public void onCompleted() {
-        scheduler.schedule(new Action0() {
-            @Override
-            public void call() {
-                underlying.onCompleted();
-            }
-        });
+        enqueue(new Notification<T>());
     }
 
     @Override
     public void onError(final Exception e) {
-        scheduler.schedule(new Action0() {
-            @Override
-            public void call() {
-                underlying.onError(e);
-            }
-        });
+        enqueue(new Notification<T>(e));
     }
 
     @Override
     public void onNext(final T args) {
-        scheduler.schedule(new Action0() {
-            @Override
-            public void call() {
-                underlying.onNext(args);
-            }
-        });
+        enqueue(new Notification<T>(args));
+    }
+
+    private void enqueue(Notification<T> notification) {
+        boolean schedule;
+        synchronized (lock) {
+            schedule = queue.isEmpty();
+
+            queue.offer(notification);
+        }
+
+        if (schedule) {
+            scheduler.schedule(new Action0() {
+                @Override
+                public void call() {
+                    Notification<T> not = dequeue();
+
+                    if (not == null) {
+                        return;
+                    }
+
+                    switch (not.getKind()) {
+                        case OnNext:
+                            underlying.onNext(not.getValue());
+                            break;
+                        case OnError:
+                            underlying.onError(not.getException());
+                            break;
+                        case OnCompleted:
+                            underlying.onCompleted();
+                            break;
+                        default:
+                            throw new IllegalStateException("Unknown kind of notification " + not);
+
+                    }
+
+                    scheduler.schedule(this);
+
+                }
+            });
+        }
+    }
+
+    private Notification<T> dequeue() {
+        synchronized (lock) {
+            return queue.poll();
+        }
     }
 }