2.x: Improve coverage & related cleanup 03/05 (#5891)

* 2.x: Improve coverage & related cleanup 03/05 * Fix camelCase local variable naming errors in tests.
ReactiveX · Mar 5, 2018 · cb05a26 · cb05a26
1 parent 51dd03b
commit cb05a26
Show file tree

Hide file tree

Showing 93 changed files with 2,523 additions and 433 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableIterable.java b/src/main/java/io/reactivex/internal/operators/flowable/BlockingFlowableIterable.java
@@ -125,9 +125,7 @@ public T next() {
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(batchSize);
-            }
+            SubscriptionHelper.setOnce(this, s, batchSize);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundary.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableBufferBoundary.java
@@ -327,9 +327,7 @@ static final class BufferOpenSubscriber<Open>
 
             @Override
             public void onSubscribe(Subscription s) {
-                if (SubscriptionHelper.setOnce(this, s)) {
-                    s.request(Long.MAX_VALUE);
-                }
+                SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
             }
 
             @Override
@@ -378,9 +376,7 @@ static final class BufferCloseSubscriber<T, C extends Collection<? super T>>
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableCache.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableCache.java
@@ -180,9 +180,7 @@ public void removeChild(ReplaySubscription<T> p) {
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(connection, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(connection, s, Long.MAX_VALUE);
         }
 
         /**

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableCombineLatest.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableCombineLatest.java
@@ -516,9 +516,7 @@ static final class CombineLatestInnerSubscriber<T>
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(prefetch);
-            }
+            SubscriptionHelper.setOnce(this, s, prefetch);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableGroupJoin.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableGroupJoin.java
@@ -417,9 +417,7 @@ public boolean isDisposed() {
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override
@@ -470,9 +468,7 @@ public boolean isDisposed() {
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithMaybe.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithMaybe.java
@@ -98,10 +98,8 @@ static final class MergeWithObserver<T> extends AtomicInteger
         }
 
         @Override
-        public void onSubscribe(Subscription d) {
-            if (SubscriptionHelper.setOnce(mainSubscription, d)) {
-                d.request(prefetch);
-            }
+        public void onSubscribe(Subscription s) {
+            SubscriptionHelper.setOnce(mainSubscription, s, prefetch);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithSingle.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableMergeWithSingle.java
@@ -98,10 +98,8 @@ static final class MergeWithObserver<T> extends AtomicInteger
         }
 
         @Override
-        public void onSubscribe(Subscription d) {
-            if (SubscriptionHelper.setOnce(mainSubscription, d)) {
-                d.request(prefetch);
-            }
+        public void onSubscribe(Subscription s) {
+            SubscriptionHelper.setOnce(mainSubscription, s, prefetch);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableReplay.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableReplay.java
@@ -526,36 +526,16 @@ static final class InnerSubscription<T> extends AtomicLong implements Subscripti
         public void request(long n) {
             // ignore negative requests
             if (SubscriptionHelper.validate(n)) {
-                // In general, RxJava doesn't prevent concurrent requests (with each other or with
-                // a cancel) so we need a CAS-loop, but we need to handle
-                // request overflow and cancelled/not requested state as well.
-                for (;;) {
-                    // get the current request amount
-                    long r = get();
-                    // if child called cancel() do nothing
-                    if (r == CANCELLED) {
-                        return;
-                    }
-                    // ignore zero requests except any first that sets in zero
-                    if (r >= 0L && n == 0) {
-                        return;
-                    }
-                    // otherwise, increase the request count
-                    long u = BackpressureHelper.addCap(r, n);
-
-                    // try setting the new request value
-                    if (compareAndSet(r, u)) {
-                        // increment the total request counter
-                        BackpressureHelper.add(totalRequested, n);
-                        // if successful, notify the parent dispatcher this child can receive more
-                        // elements
-                        parent.manageRequests();
-
-                        parent.buffer.replay(this);
-                        return;
-                    }
-                    // otherwise, someone else changed the state (perhaps a concurrent
-                    // request or cancellation) so retry
+                // add to the current requested and cap it at MAX_VALUE
+                // except when there was a concurrent cancellation
+                if (BackpressureHelper.addCancel(this, n) != CANCELLED) {
+                    // increment the total request counter
+                    BackpressureHelper.add(totalRequested, n);
+                    // if successful, notify the parent dispatcher this child can receive more
+                    // elements
+                    parent.manageRequests();
+                    // try replaying any cached content
+                    parent.buffer.replay(this);
                 }
             }
         }

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSamplePublisher.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSamplePublisher.java
@@ -93,8 +93,8 @@ public void onComplete() {
             completeMain();
         }
 
-        boolean setOther(Subscription o) {
-            return SubscriptionHelper.setOnce(other, o);
+        void setOther(Subscription o) {
+            SubscriptionHelper.setOnce(other, o, Long.MAX_VALUE);
         }
 
         @Override
@@ -150,9 +150,7 @@ static final class SamplerSubscriber<T> implements FlowableSubscriber<Object> {
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (parent.setOther(s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            parent.setOther(s);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableSkipUntil.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableSkipUntil.java
@@ -114,9 +114,7 @@ final class OtherSubscriber extends AtomicReference<Subscription>
 
             @Override
             public void onSubscribe(Subscription s) {
-                if (SubscriptionHelper.setOnce(this, s)) {
-                    s.request(Long.MAX_VALUE);
-                }
+                SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
             }
 
             @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTakeUntil.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTakeUntil.java
@@ -99,9 +99,7 @@ final class OtherSubscriber extends AtomicReference<Subscription> implements Flo
 
             @Override
             public void onSubscribe(Subscription s) {
-                if (SubscriptionHelper.setOnce(this, s)) {
-                    s.request(Long.MAX_VALUE);
-                }
+                SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
             }
 
             @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableTimeout.java
@@ -343,9 +343,7 @@ static final class TimeoutConsumer extends AtomicReference<Subscription>
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundary.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableWindowBoundary.java
@@ -95,9 +95,7 @@ static final class WindowBoundaryMainSubscriber<T, B>
 
         @Override
         public void onSubscribe(Subscription d) {
-            if (SubscriptionHelper.setOnce(upstream, d)) {
-                d.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(upstream, d, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableWithLatestFromMany.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableWithLatestFromMany.java
@@ -268,9 +268,7 @@ static final class WithLatestInnerSubscriber
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeDelayOtherPublisher.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeDelayOtherPublisher.java
@@ -120,9 +120,7 @@ static final class OtherSubscriber<T> extends
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeFromFuture.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeFromFuture.java
@@ -17,6 +17,7 @@
 
 import io.reactivex.*;
 import io.reactivex.disposables.*;
+import io.reactivex.exceptions.Exceptions;
 
 /**
  * Waits until the source Future completes or the wait times out; treats a {@code null}
@@ -50,17 +51,11 @@ protected void subscribeActual(MaybeObserver<? super T> observer) {
                 } else {
                     v = future.get(timeout, unit);
                 }
-            } catch (InterruptedException ex) {
-                if (!d.isDisposed()) {
-                    observer.onError(ex);
-                }
-                return;
-            } catch (ExecutionException ex) {
-                if (!d.isDisposed()) {
-                    observer.onError(ex.getCause());
+            } catch (Throwable ex) {
+                if (ex instanceof ExecutionException) {
+                    ex = ex.getCause();
                 }
-                return;
-            } catch (TimeoutException ex) {
+                Exceptions.throwIfFatal(ex);
                 if (!d.isDisposed()) {
                     observer.onError(ex);
                 }

diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeTakeUntilPublisher.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeTakeUntilPublisher.java
@@ -132,9 +132,7 @@ static final class TakeUntilOtherMaybeObserver<U>
 
             @Override
             public void onSubscribe(Subscription s) {
-                if (SubscriptionHelper.setOnce(this, s)) {
-                    s.request(Long.MAX_VALUE);
-                }
+                SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
             }
 
             @Override

diff --git a/src/main/java/io/reactivex/internal/operators/maybe/MaybeTimeoutPublisher.java b/src/main/java/io/reactivex/internal/operators/maybe/MaybeTimeoutPublisher.java
@@ -155,9 +155,7 @@ static final class TimeoutOtherMaybeObserver<T, U>
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableSkip.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableSkip.java
@@ -15,6 +15,7 @@
 
 import io.reactivex.*;
 import io.reactivex.disposables.Disposable;
+import io.reactivex.internal.disposables.DisposableHelper;
 
 public final class ObservableSkip<T> extends AbstractObservableWithUpstream<T, T> {
     final long n;
@@ -40,9 +41,11 @@ static final class SkipObserver<T> implements Observer<T>, Disposable {
         }
 
         @Override
-        public void onSubscribe(Disposable s) {
-            this.d = s;
-            actual.onSubscribe(this);
+        public void onSubscribe(Disposable d) {
+            if (DisposableHelper.validate(this.d, d)) {
+                this.d = d;
+                actual.onSubscribe(this);
+            }
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/parallel/ParallelJoin.java b/src/main/java/io/reactivex/internal/operators/parallel/ParallelJoin.java
@@ -516,9 +516,7 @@ static final class JoinInnerSubscriber<T>
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(prefetch);
-            }
+            SubscriptionHelper.setOnce(this, s, prefetch);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/parallel/ParallelReduceFull.java b/src/main/java/io/reactivex/internal/operators/parallel/ParallelReduceFull.java
@@ -180,9 +180,7 @@ static final class ParallelReduceFullInnerSubscriber<T>
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/parallel/ParallelSortedJoin.java b/src/main/java/io/reactivex/internal/operators/parallel/ParallelSortedJoin.java
@@ -280,9 +280,7 @@ static final class SortedJoinInnerSubscriber<T>
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/single/SingleDelayWithSingle.java b/src/main/java/io/reactivex/internal/operators/single/SingleDelayWithSingle.java
@@ -54,7 +54,7 @@ static final class OtherObserver<T, U>
 
         @Override
         public void onSubscribe(Disposable d) {
-            if (DisposableHelper.set(this, d)) {
+            if (DisposableHelper.setOnce(this, d)) {
 
                 actual.onSubscribe(this);
             }

diff --git a/src/main/java/io/reactivex/internal/operators/single/SingleTakeUntil.java b/src/main/java/io/reactivex/internal/operators/single/SingleTakeUntil.java
@@ -139,9 +139,7 @@ static final class TakeUntilOtherSubscriber
 
         @Override
         public void onSubscribe(Subscription s) {
-            if (SubscriptionHelper.setOnce(this, s)) {
-                s.request(Long.MAX_VALUE);
-            }
+            SubscriptionHelper.setOnce(this, s, Long.MAX_VALUE);
         }
 
         @Override