In order to use the Akka Streams TestKit, we need to add another dependency to our project configuration to build.sbt:

libraryDependencies ++= "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion % Test

Let's see how the TestSink and TestSource provided by this module can simplify the way we formulate our testing logic. Now we'll test the whole flow from the Boy to the Baker:

"the whole flow" should {
  "produce cookies" in {
    val testSink = TestSink.probe[ReadyCookies]
    val source = TestSource.probe[ShoppingList]
    val (publisher: TestPublisher.Probe[ShoppingList],
         subscriber: TestSubscriber.Probe[ReadyCookies]) =
      source.via(Bakery.flow).toMat(testSink)(Keep.both).run()
    subscriber.request(10)
    publisher.sendNext(ShoppingList(30, 1000, 100, 100))
    subscriber.expectNext(40.seconds, ReadyCookies(12))
 subscriber.expectNext(40.seconds, ReadyCookies(12))
  }
}

In this scenario, we first create TestSink and TestSource probes using constructors provided by the testing toolkit. Then we materialize them to publisher and subscriber in order to be able to drive the flow. Here, we're using the toMat syntax again. Until now, we implicitly used the default value (Keep.left) but now we want to keep both materialized results of the flow and of the sink. Running the flow returns its materialized instance which is a pair: TestPublisher and TestSubscriber. 

We then use subscriber to request 10 messages from the flow. In Reactive Streams, the producer is not supposed to send anything downstream until there is demand, and we express the demand with this call. We expect the flow to output elements representing RawCookies(12). Thus, our subscriber.request translates to 120 cookies to be produced.

Having this demand, we then initiate the flow by sending the next shopping list from the source. 

Finally, we expect at least two batches of cookies to arrive at the sink. We provide sufficient time for the stream to push messages through all of the stages, accounting for delays in the mixing and baking stage.

We also cannot reliably predict how many cookies will be made because of the way we drop messages at the mixing stage in the case of MotorOverheatException and SlowRotationSpeedException.

In this example, we barely scratched the surface of all of the possibilities provided by the Akka Streams TestKit. As you develop Akka Streams-based systems it is worth revisiting both the documentation and the source code of the library and keeping in mind the different testing approaches they offer.