Merge pull request #3331 from akarnokd/SchedulersTests2x

2.x: schedulers test of classic schedulers

Author: akarnokd

src/main/java/io/reactivex/internal/schedulers/ExecutorScheduler.java

@@ -97,8 +97,8 @@ public Disposable schedulePeriodicallyDirect(Runnable run, long initialDelay, lo
         }
         return super.schedulePeriodicallyDirect(run, initialDelay, period, unit);
     }
-    
-    static final class ExecutorWorker extends Scheduler.Worker implements Runnable {
+    /* public: test support. */
+    public static final class ExecutorWorker extends Scheduler.Worker implements Runnable {
         final Executor executor;
 
         final MpscLinkedQueue<Runnable> queue;
@@ -109,6 +109,8 @@ static final class ExecutorWorker extends Scheduler.Worker implements Runnable {
         static final AtomicIntegerFieldUpdater<ExecutorWorker> WIP =
                 AtomicIntegerFieldUpdater.newUpdater(ExecutorWorker.class, "wip");
 
+        final SetCompositeResource<Disposable> tasks = new SetCompositeResource<>(Disposable::dispose);
+        
         public ExecutorWorker(Executor executor) {
             this.executor = executor;
             this.queue = new MpscLinkedQueue<>();
@@ -153,26 +155,56 @@ public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
 
             MultipleAssignmentResource<Disposable> mar = new MultipleAssignmentResource<>(Disposable::dispose, first);
 
-            Disposable delayed;
-
             Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
 
-            Runnable r = () -> mar.setResource(schedule(decoratedRun));
+            ScheduledRunnable sr = new ScheduledRunnable(() -> {
+                mar.setResource(schedule(decoratedRun));
+            }, tasks);
+            tasks.add(sr);
 
             if (executor instanceof ScheduledExecutorService) {
                 try {
-                    Future<?> f = ((ScheduledExecutorService)executor).schedule(r, delay, unit);
-                    delayed = () -> f.cancel(true);
+                    Future<?> f = ((ScheduledExecutorService)executor).schedule(sr, delay, unit);
+                    sr.setFuture(f);
                 } catch (RejectedExecutionException ex) {
                     disposed = true;
                     RxJavaPlugins.onError(ex);
                     return EmptyDisposable.INSTANCE;
                 }
             } else {
-                delayed = HELPER.scheduleDirect(r, delay, unit);
+                Disposable d = HELPER.scheduleDirect(sr, delay, unit);
+                sr.setFuture(new Future<Object>() {
+                    @Override
+                    public boolean cancel(boolean mayInterruptIfRunning) {
+                        d.dispose();
+                        return false;
+                    }
+
+                    @Override
+                    public boolean isCancelled() {
+                        return false;
+                    }
+
+                    @Override
+                    public boolean isDone() {
+                        return false;
+                    }
+
+                    @Override
+                    public Object get() throws InterruptedException, ExecutionException {
+                        return null;
+                    }
+
+                    @Override
+                    public Object get(long timeout, TimeUnit unit)
+                            throws InterruptedException, ExecutionException, TimeoutException {
+                        return null;
+                    }
+                    
+                });
             }
 
-            first.setResource(delayed);
+            first.setResource(sr);
 
             return mar;
         }
@@ -181,6 +213,7 @@ public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
         public void dispose() {
             if (!disposed) {
                 disposed = true;
+                tasks.dispose();
                 if (WIP.getAndIncrement(this) == 0) {
                     queue.clear();
                 }

src/main/java/io/reactivex/schedulers/Schedulers.java

@@ -93,4 +93,20 @@ public static Scheduler immediate() {
     public static Scheduler from(Executor executor) {
         return new ExecutorScheduler(executor);
     }
+    
+    public static void shutdown() {
+        computation().shutdown();
+        io().shutdown();
+        newThread().shutdown();
+        single().shutdown();
+        trampoline().shutdown();
+    }
+    
+    public static void start() {
+        computation().start();
+        io().start();
+        newThread().start();
+        single().start();
+        trampoline().start();
+    }
 }

src/test/java/io/reactivex/GroupByTests.java

@@ -27,7 +27,10 @@ public void testTakeUnsubscribesOnGroupBy() {
         .groupBy(event -> event.type)
         .take(1)
         .toBlocking()
-        .forEach(System.out::println);
+        .forEach(v -> {
+            System.out.println(v);
+            v.take(1).subscribe();  // FIXME groups need consumption to a certain degree to cancel upstream
+        });
 
         System.out.println("**** finished");
     }

src/test/java/io/reactivex/internal/operators/OperatorConcatTest.java

@@ -129,6 +129,14 @@ public void testSimpleAsyncConcat() {
         inOrder.verify(observer, times(1)).onNext("six");
     }
 
+    @Test
+    public void testNestedAsyncConcatLoop() throws Throwable {
+        for (int i = 0; i < 100; i++) {
+            System.out.println("testNestedAsyncConcat >> " + i);
+            testNestedAsyncConcat();
+        }
+    }
+    
     /**
      * Test an async Observable that emits more async Observables
      */
@@ -242,8 +250,8 @@ public void run() {
         inOrder.verify(observer, times(1)).onNext("eight");
         inOrder.verify(observer, times(1)).onNext("nine");
 
-        inOrder.verify(observer, times(1)).onComplete();
         verify(observer, never()).onError(any(Throwable.class));
+        inOrder.verify(observer, times(1)).onComplete();
     }
 
     @Test
@@ -469,7 +477,7 @@ public void cancel() {
         private final List<T> values;
         private Thread t = null;
         private int count = 0;
-        private boolean subscribed = true;
+        private volatile boolean subscribed = true;
         private final CountDownLatch once;
         private final CountDownLatch okToContinue;
         private final CountDownLatch threadHasStarted = new CountDownLatch(1);

src/test/java/io/reactivex/internal/operators/OperatorOnErrorResumeNextViaObservableTest.java

@@ -100,7 +100,7 @@ public String apply(String s) {
 
     @Test
     @Ignore("Publishers should not throw")
-    public void testResumeNextWithFailedOnSubscribe() {
+    public void testResumeNextWithFailureOnSubscribe() {
         Observable<String> testObservable = Observable.create(new Publisher<String>() {
 
             @Override
@@ -122,7 +122,7 @@ public void subscribe(Subscriber<? super String> t1) {
 
     @Test
     @Ignore("Publishers should not throw")
-    public void testResumeNextWithFailedOnSubscribeAsync() {
+    public void testResumeNextWithFailureOnSubscribeAsync() {
         Observable<String> testObservable = Observable.create(new Publisher<String>() {
 
             @Override