fixes AssertSubscriber to terminate upstream if ASYNC fusion

Signed-off-by: Oleh Dokuka <[email protected]>
rsocket · Feb 27, 2021 · ce62903 · ce62903
1 parent 61652c3
commit ce62903
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/rsocket-core/src/test/java/io/rsocket/internal/subscriber/AssertSubscriber.java b/rsocket-core/src/test/java/io/rsocket/internal/subscriber/AssertSubscriber.java
@@ -943,6 +943,7 @@ void drain() {
       t = qs.poll();
       if (t == null) {
         if (done) {
+          qs.clear(); // clear upstream to terminated it due to the contract
           cdl.countDown();
           return;
         }