2.x: fix PublishProcessor cancel/emission overflow bug

ReactiveX · Oct 15, 2017 · 678c62f · 678c62f
1 parent cd6bc08
commit 678c62f
Show file tree

Hide file tree

Showing 3 changed files with 67 additions and 3 deletions.
diff --git a/src/main/java/io/reactivex/processors/PublishProcessor.java b/src/main/java/io/reactivex/processors/PublishProcessor.java
@@ -313,9 +313,7 @@ public void onNext(T t) {
             }
             if (r != 0L) {
                 actual.onNext(t);
-                if (r != Long.MAX_VALUE) {
-                    decrementAndGet();
-                }
+                BackpressureHelper.producedCancel(this, 1);
             } else {
                 cancel();
                 actual.onError(new MissingBackpressureException("Could not emit value due to lack of requests"));

diff --git a/src/test/java/io/reactivex/processors/BehaviorProcessorTest.java b/src/test/java/io/reactivex/processors/BehaviorProcessorTest.java
@@ -812,4 +812,37 @@ public void run() {
             ts.assertFailure(TestException.class);
         }
     }
+
+    @Test(timeout = 10000)
+    public void subscriberCancelOfferRace() {
+        for (int i = 0; i < 1000; i++) {
+            final BehaviorProcessor<Integer> pp = BehaviorProcessor.create();
+
+            final TestSubscriber<Integer> ts = pp.test(1);
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    for (int i = 0; i < 2; i++) {
+                        while (!pp.offer(i)) ;
+                    }
+                }
+            };
+
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    ts.cancel();
+                }
+            };
+
+            TestHelper.race(r1, r2);
+
+            if (ts.valueCount() > 0) {
+                ts.assertValuesOnly(0);
+            } else {
+                ts.assertEmpty();
+            }
+        }
+    }
 }
diff --git a/src/test/java/io/reactivex/processors/PublishProcessorTest.java b/src/test/java/io/reactivex/processors/PublishProcessorTest.java
@@ -677,4 +677,37 @@ public void run() {
         .awaitDone(5, TimeUnit.SECONDS)
         .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
     }
+
+    @Test(timeout = 10000)
+    public void subscriberCancelOfferRace() {
+        for (int i = 0; i < 1000; i++) {
+            final PublishProcessor<Integer> pp = PublishProcessor.create();
+
+            final TestSubscriber<Integer> ts = pp.test(1);
+
+            Runnable r1 = new Runnable() {
+                @Override
+                public void run() {
+                    for (int i = 0; i < 2; i++) {
+                        while (!pp.offer(i)) ;
+                    }
+                }
+            };
+
+            Runnable r2 = new Runnable() {
+                @Override
+                public void run() {
+                    ts.cancel();
+                }
+            };
+
+            TestHelper.race(r1, r2);
+
+            if (ts.valueCount() > 0) {
+                ts.assertValuesOnly(0);
+            } else {
+                ts.assertEmpty();
+            }
+        }
+    }
 }