Unit test to assert correct scheduler thread

Author: benjchristensen

rxjava-core/src/test/java/rx/operators/OperationObserveOnTest.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();
+    }
 }