argh, you are right, I have created a failing test for that. Out of curiousity I also tried similar when throwing from `onNext` and we have the same problem there. What is the expected semantics? Do we have to deliver all (same) elements to all subscribers before emitting the error?

This test fails:

    "deliver all elements two consumers before emitting error" in {
      val consumer = StreamTestKit.consumerProbe[Int]
      val consumer2 = StreamTestKit.consumerProbe[Int]
      val p = Flow(List(1, 2, 3, 4, 5)).transform(new Transformer[Int, Int] {
        override def onNext(elem: Int) =
          if (elem == 4) throw new RuntimeException(s"Boom") with NoStackTrace
          else List(elem)
      }).toProducer(materializer)

      p.produceTo(consumer)
      p.produceTo(consumer2)

      val subscription = consumer.expectSubscription()
      val subscription2 = consumer2.expectSubscription()

      subscription.requestMore(10)
      subscription2.requestMore(2)

      consumer.expectNext(1)
      consumer.expectNext(2)
      consumer.expectNext(3)
      consumer.expectError.getMessage should be("Boom")

      consumer2.expectNext(1)
      consumer2.expectNext(2)
      consumer2.expectNoMsg(1.second)

      subscription2.requestMore(2)

      consumer2.expectNext(3)
      consumer2.expectError.getMessage should be("Boom")
    }

consumer2 receives the error after element 2, i.e. it never gets 3