title: Why does the Akka Streams TestSource not receive every pull as a request?
body: <p>I am trying to track down a bug in a custom Flow where there were extra requests going upstream. I wrote a test case which should have failed given the behaviour and logging I was seeing but it didn't.</p>
  
  <p>I then made a trivial Flow with an obvious extra pull and yet the test still did not fail.</p>
  
  <p>Here is an example:</p>
  
  <pre><code>class EagerFlow extends GraphStage[FlowShape[String, String]] {
  
    val in: Inlet[String] = Inlet("EagerFlow.in")
    val out: Outlet[String] = Outlet("EagerFlow.out")
  
    override def shape: FlowShape[String, String] = FlowShape(in, out)
  
    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) {
      setHandler(in, new InHandler {
        override def onPush() = {
          println("EagerFlow.in.onPush")
          println("EagerFlow.out.push")
          push(out, grab(in))
          println("EagerFlow.in.pull")
          pull(in)
        }
      })
      setHandler(out, new OutHandler {
        override def onPull() = {
          println("EagerFlow.out.onPull")
          println("EagerFlow.in.pull")
          pull(in)
        }
      })
    }
  }
  </code></pre>
  
  <p>Here is a test that should pass, i.e. confirm that there is an extra request.</p>
  
  <pre><code>"Eager Flow" should "pull on every push" in {
    val sourceProbe = TestSource.probe[String]
    val sinkProbe = TestSink.probe[String]
  
    val eagerFlow = Flow.fromGraph(new EagerFlow)
  
    val (source, sink) = sourceProbe.
      via(eagerFlow).
      toMat(sinkProbe)(Keep.both).
      run
  
    sink.request(1)
    source.expectRequest()
    source.expectNoMsg()
  
    source.sendNext("hello")
    sink.expectNext()
    source.expectRequest()
  }
  </code></pre>
  
  <p>In standard output you can see:</p>
  
  <pre><code>0C3E08A8 PULL DownstreamBoundary -&gt; TestSpec$EagerFlow@1d154180 (TestSpec$EagerFlow$$anon$4$$anon$6@7e62ecbd) [TestSpec$EagerFlow$$anon$4@36904dd4]
  EagerFlow.out.onPull
  EagerFlow.in.pull
  0C3E08A8 PULL TestSpec$EagerFlow@1d154180 -&gt; UpstreamBoundary (BatchingActorInputBoundary(id=0, fill=0/16, completed=false, canceled=false)) [UpstreamBoundary]
  EagerFlow.in.onPush
  EagerFlow.out.push
  EagerFlow.in.pull
  0C3E08A8 PULL TestSpec$EagerFlow@1d154180 -&gt; UpstreamBoundary (BatchingActorInputBoundary(id=0, fill=0/16, completed=false, canceled=false)) [UpstreamBoundary]
  
  assertion failed: timeout (3 seconds) during expectMsg:
    java.lang.AssertionError: assertion failed: timeout (3 seconds) during expectMsg:
    at scala.Predef$.assert(Predef.scala:170)
  at akka.testkit.TestKitBase$class.expectMsgPF(TestKit.scala:368)
  at akka.testkit.TestKit.expectMsgPF(TestKit.scala:737)
  at akka.stream.testkit.StreamTestKit$PublisherProbeSubscription.expectRequest(StreamTestKit.scala:665)
  at akka.stream.testkit.TestPublisher$Probe.expectRequest(StreamTestKit.scala:172)
  </code></pre>
  
  <p>(This is with the debug logging in <code>akka.stream.impl.fusing.GraphInterpreter#processEvent</code>, well actually using print-points but effectively the same thing)</p>
  
  <p>Why does this pull stop in the <code>UpstreamBoundary</code> and never make it back to the <code>TestSource.probe</code>?</p>
  
  <p>I have quite a few tests which use <code>expectNoMsg()</code> to ensure correct backpressure but it seems these will just give false positives. How should I be testing this if this doesn't work?</p>