Merge pull request #20838 from drewhk/wip-19931-expectRequest-update-pending-drewhk

#19931 Track pending requests properly on expectRequest

akka-stream-testkit/src/main/scala/akka/stream/testkit/StreamTestKit.scala

@@ -168,7 +168,11 @@ object TestPublisher {
       this
     }
 
-    def expectRequest(): Long = subscription.expectRequest()
+    def expectRequest(): Long = {
+      val requests = subscription.expectRequest()
+      pendingRequests += requests
+      requests
+    }
 
     def expectCancellation(): Self = {
       subscription.expectCancellation()

akka-stream-testkit/src/test/scala/akka/stream/testkit/TestPublisherSubscriberSpec.scala

@@ -59,5 +59,20 @@ class TestPublisherSubscriberSpec extends AkkaSpec {
       upstreamSubscription.sendComplete()
     }
 
+    "properly update pendingRequest in expectRequest" in {
+      val upstream = TestPublisher.probe[Int]()
+      val downstream = TestSubscriber.manualProbe[Int]()
+
+      Source.fromPublisher(upstream).runWith(Sink.fromSubscriber(downstream))
+
+      downstream
+        .expectSubscription()
+        .request(10)
+
+      upstream.expectRequest() should ===(10)
+      upstream.sendNext(1)
+      downstream.expectNext(1)
+    }
+
   }
 }