2.x: cleanup & coverage 10/24-2 (#4763)

ReactiveX · Oct 25, 2016 · 318bf43 · 318bf43
1 parent 3634c92
commit 318bf43
Show file tree

Hide file tree

Showing 33 changed files with 1,254 additions and 338 deletions.
diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableGroupJoin.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableGroupJoin.java
@@ -484,7 +484,7 @@ public void onNext(Object t) {
 
         @Override
         public void onError(Throwable t) {
-            parent.innerError(t);
+            parent.innerCloseError(t);
         }
 
         @Override

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableInterval.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableInterval.java
@@ -82,15 +82,10 @@ public void run() {
 
                 if (r != 0L) {
                     actual.onNext(count++);
-                    if (r != Long.MAX_VALUE) {
-                        decrementAndGet();
-                    }
+                    BackpressureHelper.produced(this, 1);
                 } else {
-                    try {
-                        actual.onError(new MissingBackpressureException("Can't deliver value " + count + " due to lack of requests"));
-                    } finally {
-                        DisposableHelper.dispose(resource);
-                    }
+                    actual.onError(new MissingBackpressureException("Can't deliver value " + count + " due to lack of requests"));
+                    DisposableHelper.dispose(resource);
                 }
             }
         }

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableJoin.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableJoin.java
@@ -55,8 +55,8 @@ public FlowableJoin(
     @Override
     protected void subscribeActual(Subscriber<? super R> s) {
 
-        GroupJoinSubscription<TLeft, TRight, TLeftEnd, TRightEnd, R> parent =
-                new GroupJoinSubscription<TLeft, TRight, TLeftEnd, TRightEnd, R>(s, leftEnd, rightEnd, resultSelector);
+        JoinSubscription<TLeft, TRight, TLeftEnd, TRightEnd, R> parent =
+                new JoinSubscription<TLeft, TRight, TLeftEnd, TRightEnd, R>(s, leftEnd, rightEnd, resultSelector);
 
         s.onSubscribe(parent);
 
@@ -69,7 +69,7 @@ protected void subscribeActual(Subscriber<? super R> s) {
         other.subscribe(right);
     }
 
-    static final class GroupJoinSubscription<TLeft, TRight, TLeftEnd, TRightEnd, R>
+    static final class JoinSubscription<TLeft, TRight, TLeftEnd, TRightEnd, R>
     extends AtomicInteger implements Subscription, JoinSupport {
 
 
@@ -111,7 +111,7 @@ static final class GroupJoinSubscription<TLeft, TRight, TLeftEnd, TRightEnd, R>
 
         static final Integer RIGHT_CLOSE = 4;
 
-        GroupJoinSubscription(Subscriber<? super R> actual, Function<? super TLeft, ? extends Publisher<TLeftEnd>> leftEnd,
+        JoinSubscription(Subscriber<? super R> actual, Function<? super TLeft, ? extends Publisher<TLeftEnd>> leftEnd,
                 Function<? super TRight, ? extends Publisher<TRightEnd>> rightEnd,
                         BiFunction<? super TLeft, ? super TRight, ? extends R> resultSelector) {
             this.actual = actual;

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureDrop.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureDrop.java
@@ -81,9 +81,7 @@ public void onNext(T t) {
             long r = get();
             if (r != 0L) {
                 actual.onNext(t);
-                if (r != Long.MAX_VALUE) {
-                    decrementAndGet();
-                }
+                BackpressureHelper.produced(this, 1);
             } else {
                 try {
                     onDrop.accept(t);

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureError.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureError.java
@@ -65,9 +65,7 @@ public void onNext(T t) {
             long r = get();
             if (r != 0L) {
                 actual.onNext(t);
-                if (r != Long.MAX_VALUE) {
-                    decrementAndGet();
-                }
+                BackpressureHelper.produced(this, 1);
             } else {
                 onError(new MissingBackpressureException("could not emit value due to lack of requests"));
             }

diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishMulticast.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowablePublishMulticast.java
@@ -102,20 +102,14 @@ public void onNext(R t) {
 
         @Override
         public void onError(Throwable t) {
-            try {
-                actual.onError(t);
-            } finally {
-                processor.dispose();
-            }
+            actual.onError(t);
+            processor.dispose();
         }
 
         @Override
         public void onComplete() {
-            try {
-                actual.onComplete();
-            } finally {
-                processor.dispose();
-            }
+            actual.onComplete();
+            processor.dispose();
         }
 
         @Override
@@ -214,12 +208,10 @@ public void onNext(T t) {
             if (done) {
                 return;
             }
-            if (sourceMode == QueueSubscription.NONE) {
-                if (!queue.offer(t)) {
-                    SubscriptionHelper.cancel(s);
-                    onError(new MissingBackpressureException());
-                    return;
-                }
+            if (sourceMode == QueueSubscription.NONE && !queue.offer(t)) {
+                s.get().cancel();
+                onError(new MissingBackpressureException());
+                return;
             }
             drain();
         }
@@ -473,20 +465,7 @@ static final class MulticastSubscription<T>
         @Override
         public void request(long n) {
             if (SubscriptionHelper.validate(n)) {
-                for (;;) {
-                    long r = get();
-                    if (r == Long.MIN_VALUE) {
-                        return;
-                    }
-                    if (r != Long.MAX_VALUE) {
-                        long u = BackpressureHelper.addCap(r, n);
-                        if (compareAndSet(r, u)) {
-                            break;
-                        }
-                    } else {
-                        break;
-                    }
-                }
+                BackpressureHelper.addCancel(this, n);
                 parent.drain();
             }
         }