channel is already closed when applying recovery strategy on a failed stream

[TrustNoOne]

The following code never nacks a "fail' message, every time it's restarted it reads it again.

It fails here with an AlreadyClosedException.

The documentation says RecoveryStrategy does not work with streams: why do I have to specify one then? Maybe It should be hardcoded to "none" if not supported?

 implicit val recoveryStrategy = RecoveryStrategy.nack(requeue = false)
    val src = RabbitSource(
      rabbitControl,
      channel(qos = 3),
      consume(queue(
        "hello",
        durable    = false,
        exclusive  = false,
        autoDelete = false
      )),
      body(as[String])
    )

    val res = src
      .runForeach { msg ⇒
        if (msg == "fail") throw new Exception("errorrr")
        println(msg)
      }

    Await.result(res, Duration.Inf)

[timcharper]

I don't remember writing anywhere in the documentation that recovery strategy doesn't work with streams... where did you read this?

I just pushed up a test change, too, which does show that this behavior works: a51d408

[timcharper]

Ah... I see what is happening.

The stream is closing because you did not set a resuming decider. Then, there is a race condition. The channel closes when the stream closes, but this is happening before the recoveryStrategy has a chance to apply.

[timcharper]

Not sure how to best solve this; any ideas?