acked-graph.scala

      val input = (Stream.continually(Promise[Unit]) zip Range.inclusive(1, 5)).toList
      implicit val materializer = ActorMaterializer()

      val source = AckedSource(input)

      val Seq(s1, s2) = (1 to 2) map { n => AckedSink.fold[Int, Int](0)(_ + _) }

      val g = AckedFlowGraph.closed(s1, s2)((m1, m2) => (m1, m2)) { implicit b =>
        (s1, s2) =>

        import AckedFlowGraph.Implicits._

        val broadcast = b.add(AckedBroadcast[Int](2))

        source ~> broadcast ~> s1
        broadcast ~> s2
      }

      // If AckedBroadcast failed to create a unique promise for each stream, then "java.lang.IllegalStateException: Promise already completed" would be thrown.
      val (f1, f2) = g.run()
      val (r1, r2) = (await(f1), await(f2))

      r1 should be (15)
      r2 should be (15)
      input.foreach { case (p, _) => await(p.future) }