Fix MulticastProcessor lockstep request accounting

Author: akarnokd

src/main/java/hu/akarnokd/rxjava2/processors/MulticastProcessor.java

@@ -57,8 +57,6 @@ public final class MulticastProcessor<T> extends FlowableProcessor<T> {
 
     int consumed;
 
-    long emitted;
-
     int fusionMode;
 
     @SuppressWarnings("rawtypes")
@@ -360,27 +358,30 @@ void drain() {
         AtomicReference<MulticastSubscription<T>[]> subs = subscribers;
         int c = consumed;
         int lim = limit;
-        long e = emitted;
         SimpleQueue<T> q = queue;
         int fm = fusionMode;
 
         outer:
         for (;;) {
 
             MulticastSubscription<T>[] as = subs.get();
+            int n = as.length;
 
-            long r = Long.MAX_VALUE;
+            if (n != 0) {
+                long r = -1L;
 
-            for (MulticastSubscription<T> a : as) {
-                long ra = a.get();
-                if (ra >= 0L) {
-                    r = Math.min(r, ra);
+                for (MulticastSubscription<T> a : as) {
+                    long ra = a.get();
+                    if (ra >= 0L) {
+                        if (r == -1L) {
+                            r = ra - a.emitted;
+                        } else {
+                            r = Math.min(r, ra - a.emitted);
+                        }
+                    }
                 }
-            }
-
-            if (as.length != 0) {
 
-                while (e != r) {
+                while (r > 0L) {
                     MulticastSubscription<T>[] bs = subs.get();
 
                     if (bs == TERMINATED) {
@@ -430,7 +431,7 @@ void drain() {
                         inner.onNext(v);
                     }
 
-                    e++;
+                    r--;
 
                     if (fm != QueueSubscription.SYNC) {
                         if (++c == lim) {
@@ -440,7 +441,7 @@ void drain() {
                     }
                 }
 
-                if (e == r) {
+                if (r == 0) {
                     MulticastSubscription<T>[] bs = subs.get();
 
                     if (bs == TERMINATED) {
@@ -471,7 +472,6 @@ void drain() {
             int w = wip.get();
             if (w == missed) {
                 consumed = c;
-                emitted = e;
                 missed = wip.addAndGet(-missed);
                 if (missed == 0) {
                     break;
@@ -490,6 +490,8 @@ static final class MulticastSubscription<T> extends AtomicLong implements Subscr
 
         final MulticastProcessor<T> parent;
 
+        long emitted;
+
         MulticastSubscription(Subscriber<? super T> actual, MulticastProcessor<T> parent) {
             this.actual = actual;
             this.parent = parent;
@@ -524,6 +526,7 @@ public void cancel() {
 
         void onNext(T t) {
             if (get() != Long.MIN_VALUE) {
+                emitted++;
                 actual.onNext(t);
             }
         }

src/test/java/hu/akarnokd/rxjava2/processors/MulticastProcessorTest.java

@@ -495,4 +495,36 @@ public Integer apply(Integer v) throws Exception {
 
         mp.test().assertFailure(IOException.class);
     }
+
+    @Test
+    public void lockstep() {
+        MulticastProcessor<Integer> mp = MulticastProcessor.create();
+
+        TestSubscriber<Integer> ts1 = mp.test();
+        mp.start();
+
+        mp.onNext(1);
+        mp.onNext(2);
+
+        ts1.assertValues(1, 2);
+
+        TestSubscriber<Integer> ts2 = mp.test(0);
+
+        ts2.assertEmpty();
+
+        mp.onNext(3);
+
+        ts1.assertValues(1, 2);
+        ts2.assertEmpty();
+
+        mp.onComplete();
+
+        ts1.assertValues(1, 2);
+        ts2.assertEmpty();
+
+        ts2.request(1);
+
+        ts1.assertResult(1, 2, 3);
+        ts2.assertResult(3);
+    }
 }