3.x: Fix MulticastProcessor not requesting more after limit is reached (#6714)

* 3.x: Fix MulticastProcessor not requesting more after limit is reached

* Test for more prefetch values and patterns.

Author: akarnokd

src/main/java/io/reactivex/rxjava3/processors/MulticastProcessor.java

@@ -570,6 +570,7 @@ void drain() {
                 }
             }
 
+            consumed = c;
             missed = wip.addAndGet(-missed);
             if (missed == 0) {
                 break;

src/test/java/io/reactivex/rxjava3/processors/MulticastProcessorTest.java

@@ -784,4 +784,41 @@ public void noUpstream() {
         assertTrue(mp.hasSubscribers());
     }
 
+    @Test
+    public void requestUpstreamPrefetchNonFused() {
+        for (int j = 1; j < 12; j++) {
+            MulticastProcessor<Integer> mp = MulticastProcessor.create(j, true);
+
+            TestSubscriber<Integer> ts = mp.test(0).withTag("Prefetch: " + j);
+
+            Flowable.range(1, 10).hide().subscribe(mp);
+
+            ts.assertEmpty()
+            .requestMore(3)
+            .assertValuesOnly(1, 2, 3)
+            .requestMore(3)
+            .assertValuesOnly(1, 2, 3, 4, 5, 6)
+            .requestMore(4)
+            .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+        }
+    }
+
+    @Test
+    public void requestUpstreamPrefetchNonFused2() {
+        for (int j = 1; j < 12; j++) {
+            MulticastProcessor<Integer> mp = MulticastProcessor.create(j, true);
+
+            TestSubscriber<Integer> ts = mp.test(0).withTag("Prefetch: " + j);
+
+            Flowable.range(1, 10).hide().subscribe(mp);
+
+            ts.assertEmpty()
+            .requestMore(2)
+            .assertValuesOnly(1, 2)
+            .requestMore(2)
+            .assertValuesOnly(1, 2, 3, 4)
+            .requestMore(6)
+            .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+        }
+    }
 }