Hi,
when using Akka Streams (2.5.9) I've recently bumped into a problem when testing a stream with a mapAsync stage. It unexpectedly ends with an assertion failing on *"timeout (3 seconds) during expectMsg while waiting for OnComplete"* To demonstrate it I've created a simple example. The code below works as expected. val sourceUnderTest = Source(1 to 2).mapAsync(2)(i => Future.successful(i * 2)) val c = TestSubscriber.manualProbe[Int]() val p = sourceUnderTest.to(Sink.fromSubscriber(c)).run() val sub = c.expectSubscription() sub.request(2) c.expectNextN(2) // List(2,4) c.expectComplete() // akka.stream.testkit.TestSubscriber$ManualProbe@fc258b1 However, when I add another mapping function (bold code), the test fails with a timeout. When I call additional c.request(1) (commented code) it ends correctly. So it looks like the mapping function adds an additional item into the stream, which seems strange. val sourceUnderTest = Source(1 to 2).mapAsync(2)(i => Future.successful(i * 2)*.**map(identity**)*) val c = TestSubscriber.manualProbe[Int]() val p = sourceUnderTest.to(Sink.fromSubscriber(c)).run() val sub = c.expectSubscription() sub.request(2) c.expectNextN(2) // List(2,4) // c.request(1) c.expectComplete() // ends with java.lang.AssertionError: assertion failed: timeout (3 seconds) during expectMsg while waiting for OnComplete Can anyone explain this strange behaviour ? Thanks Jakub -- >>>>>>>>>> Read the docs: http://akka.io/docs/ >>>>>>>>>> Check the FAQ: >>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user --- You received this message because you are subscribed to the Google Groups "Akka User List" group. To unsubscribe from this group and stop receiving emails from it, send an email to [email protected]. To post to this group, send email to [email protected]. Visit this group at https://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
