2.x: fix window-timed test leaking an MBE that affects unrelated tests (#4755)

Author: akarnokd

src/main/java/io/reactivex/internal/operators/flowable/FlowableDistinct.java

@@ -18,11 +18,10 @@
 
 import org.reactivestreams.*;
 
-import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.functions.ObjectHelper;
-import io.reactivex.internal.fuseable.*;
+import io.reactivex.internal.fuseable.QueueFuseable;
 import io.reactivex.internal.subscribers.BasicFuseableSubscriber;
 import io.reactivex.internal.subscriptions.EmptySubscription;
 import io.reactivex.plugins.RxJavaPlugins;

src/main/java/io/reactivex/internal/operators/observable/ObservableDistinct.java

@@ -17,12 +17,10 @@
 import java.util.concurrent.Callable;
 
 import io.reactivex.*;
-import io.reactivex.disposables.Disposable;
 import io.reactivex.exceptions.Exceptions;
 import io.reactivex.functions.Function;
 import io.reactivex.internal.disposables.EmptyDisposable;
 import io.reactivex.internal.functions.ObjectHelper;
-import io.reactivex.internal.fuseable.SimpleQueue;
 import io.reactivex.internal.observers.BasicFuseableObserver;
 import io.reactivex.plugins.RxJavaPlugins;
 

src/main/java/io/reactivex/internal/operators/observable/ObservableFilter.java

@@ -14,7 +14,6 @@
 package io.reactivex.internal.operators.observable;
 
 import io.reactivex.*;
-import io.reactivex.disposables.Disposable;
 import io.reactivex.functions.Predicate;
 import io.reactivex.internal.observers.BasicFuseableObserver;
 

src/test/java/io/reactivex/internal/operators/flowable/FlowableWindowWithTimeTest.java

@@ -17,7 +17,7 @@
 
 import java.util.*;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.*;
 
 import org.junit.*;
 import org.reactivestreams.*;
@@ -528,8 +528,15 @@ public void exactBoundaryError() {
     }
 
     @Test
-    public void restartTimerMany() {
+    public void restartTimerMany() throws Exception {
+        final AtomicBoolean cancel1 = new AtomicBoolean();
         Flowable.intervalRange(1, 1000, 1, 1, TimeUnit.MILLISECONDS)
+        .doOnCancel(new Action() {
+            @Override
+            public void run() throws Exception {
+                cancel1.set(true);
+            }
+        })
         .window(1, TimeUnit.MILLISECONDS, Schedulers.single(), 2, true)
         .flatMap(Functions.<Flowable<Long>>identity())
         .take(500)
@@ -539,6 +546,13 @@ public void restartTimerMany() {
         .assertValueCount(500)
         .assertNoErrors()
         .assertComplete();
+
+        int timeout = 20;
+        while (timeout-- > 0 && !cancel1.get()) {
+            Thread.sleep(100);
+        }
+
+        assertTrue("intervalRange was not cancelled!", cancel1.get());
     }
 
     @Test