Merge pull request #499 from benjchristensen/observeOn-refactor

ObserveOn Refactor

Author: benjchristensen

rxjava-core/src/main/java/rx/Notification.java

@@ -116,6 +116,16 @@ public boolean isOnNext() {
         return getKind() == Kind.OnNext;
     }
 
+    public void accept(Observer<? super T> observer) {
+        if (isOnNext()) {
+            observer.onNext(getValue());
+        } else if (isOnCompleted()) {
+            observer.onCompleted();
+        } else if (isOnError()) {
+            observer.onError(getThrowable());
+        }
+    }
+
     public static enum Kind {
         OnNext, OnError, OnCompleted
     }

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

@@ -15,13 +15,20 @@
  */
 package rx.operators;
 
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import rx.Notification;
 import rx.Observable;
 import rx.Observable.OnSubscribeFunc;
 import rx.Observer;
 import rx.Scheduler;
 import rx.Subscription;
+import rx.concurrency.CurrentThreadScheduler;
 import rx.concurrency.ImmediateScheduler;
 import rx.subscriptions.CompositeSubscription;
+import rx.util.functions.Action0;
+import rx.util.functions.Action1;
 
 /**
  * Asynchronously notify Observers on the specified Scheduler.
@@ -38,6 +45,9 @@ private static class ObserveOn<T> implements OnSubscribeFunc<T> {
         private final Observable<? extends T> source;
         private final Scheduler scheduler;
 
+        final ConcurrentLinkedQueue<Notification<? extends T>> queue = new ConcurrentLinkedQueue<Notification<? extends T>>();
+        final AtomicInteger counter = new AtomicInteger(0);
+
         public ObserveOn(Observable<? extends T> source, Scheduler scheduler) {
             this.source = source;
             this.scheduler = scheduler;
@@ -48,11 +58,55 @@ public Subscription onSubscribe(final Observer<? super T> observer) {
             if (scheduler instanceof ImmediateScheduler) {
                 // do nothing if we request ImmediateScheduler so we don't invoke overhead
                 return source.subscribe(observer);
+            } else if (scheduler instanceof CurrentThreadScheduler) {
+                // do nothing if we request CurrentThreadScheduler so we don't invoke overhead
+                return source.subscribe(observer);
             } else {
-                CompositeSubscription s = new CompositeSubscription();
-                s.add(source.subscribe(new ScheduledObserver<T>(s, observer, scheduler)));
-                return s;
+                return observeOn(observer, scheduler);
             }
         }
+
+        public Subscription observeOn(final Observer<? super T> observer, Scheduler scheduler) {
+            final CompositeSubscription s = new CompositeSubscription();
+
+            s.add(source.materialize().subscribe(new Action1<Notification<? extends T>>() {
+
+                @Override
+                public void call(Notification<? extends T> e) {
+                    // this must happen before 'counter' is used to provide synchronization between threads
+                    queue.offer(e);
+
+                    // we now use counter to atomically determine if we need to start processing or not
+                    // it will be 0 if it's the first notification or the scheduler has finished processing work
+                    // and we need to start doing it again
+                    if (counter.getAndIncrement() == 0) {
+                        processQueue(s, observer);
+                    }
+
+                }
+            }));
+
+            return s;
+        }
+
+        private void processQueue(CompositeSubscription s, final Observer<? super T> observer) {
+            s.add(scheduler.schedule(new Action1<Action0>() {
+                @Override
+                public void call(Action0 self) {
+                    Notification<? extends T> not = queue.poll();
+                    if (not != null) {
+                        not.accept(observer);
+                    }
+
+                    // decrement count and if we still have work to do
+                    // recursively schedule ourselves to process again
+                    if (counter.decrementAndGet() > 0) {
+                        self.call();
+                    }
+
+                }
+            }));
+        }
     }
+
 }

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

@@ -28,12 +28,17 @@ public void testUnsubscribeOfNewThread() throws InterruptedException {
     public void testUnsubscribeOfThreadPoolForIO() throws InterruptedException {
         testUnSubscribeForScheduler(Schedulers.threadPoolForIO());
     }
-    
+
     @Test
     public void testUnsubscribeOfThreadPoolForComputation() throws InterruptedException {
         testUnSubscribeForScheduler(Schedulers.threadPoolForComputation());
     }
-    
+
+    @Test
+    public void testUnsubscribeOfImmediateThread() throws InterruptedException {
+        testUnSubscribeForScheduler(Schedulers.immediate());
+    }
+
     @Test
     public void testUnsubscribeOfCurrentThread() throws InterruptedException {
         testUnSubscribeForScheduler(Schedulers.currentThread());
@@ -56,7 +61,7 @@ public Long call(Long aLong) {
                     }
                 })
                 .subscribeOn(scheduler)
-                .observeOn(Schedulers.currentThread())
+                .observeOn(scheduler)
                 .subscribe(new Observer<Long>() {
                     @Override
                     public void onCompleted() {

rxjava-core/src/test/java/rx/concurrency/SchedulerUnsubscribeTest.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
- *
+ * 
+ * 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.
@@ -16,6 +16,7 @@
 package rx.operators;
 
 import static org.junit.Assert.*;
+import static org.mockito.Matchers.*;
 import static org.mockito.Mockito.*;
 import static rx.operators.OperationObserveOn.*;
 
@@ -30,6 +31,7 @@
 import rx.Observable;
 import rx.Observer;
 import rx.concurrency.Schedulers;
+import rx.util.functions.Action1;
 
 public class OperationObserveOnTest {
 
@@ -81,4 +83,53 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
         inOrder.verify(observer, times(1)).onCompleted();
         inOrder.verifyNoMoreInteractions();
     }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    public void testThreadName() throws InterruptedException {
+        Observable<String> obs = Observable.from("one", null, "two", "three", "four");
+
+        Observer<String> observer = mock(Observer.class);
+
+        InOrder inOrder = inOrder(observer);
+
+        final CountDownLatch completedLatch = new CountDownLatch(1);
+        doAnswer(new Answer<Void>() {
+
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable {
+                completedLatch.countDown();
+
+                return null;
+            }
+        }).when(observer).onCompleted();
+
+        doAnswer(new Answer<Void>() {
+
+            @Override
+            public Void answer(InvocationOnMock invocation) throws Throwable {
+                completedLatch.countDown();
+
+                return null;
+            }
+        }).when(observer).onError(any(Exception.class));
+
+        obs.observeOn(Schedulers.newThread()).doOnEach(new Action1<String>() {
+
+            @Override
+            public void call(String t1) {
+                String threadName = Thread.currentThread().getName();
+                boolean correctThreadName = threadName.startsWith("RxNewThreadScheduler");
+                System.out.println("ThreadName: " + threadName + "  Correct => " + correctThreadName);
+                assertTrue(correctThreadName);
+            }
+
+        }).subscribe(observer);
+
+        if (!completedLatch.await(1000, TimeUnit.MILLISECONDS)) {
+            fail("timed out waiting");
+        }
+
+        inOrder.verify(observer, times(1)).onCompleted();
+    }
 }