How is ResilientStream supposed to work in the context of RMQ? about fs2-rabbit HOT 2 OPEN

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

from fs2-rabbit.

It depends on how you construct your program I believe.
Restarting just the consumer stream from createAckerConsumer means re-using the same connection and the same channel. The broker is still awaiting an ack from that channel.

If you what to use ResilientStream your stream should also spawn and release the channel I think.
I believe restarting the channel would be enough, if not include also the creation of the connection under the resilient stream.

Try something like this:

    val config: Fs2RabbitConfig = ???
    val run: IO[Nothing] = RabbitClient
      .default[IO](config)
      .resource
      .evalMap { connection =>
        ResilientStream.run {
          for {
            given AMQPChannel <- fs2.Stream.resource(connection.createConnectionChannel)
            (acker, stream) <- fs2.Stream.eval(connection.createAckerConsumer(QueueName("queue_name")))
            program <- stream.evalMap(msg => IO(println(msg)) >> acker(Ack(msg.deliveryTag)))
          } yield program
        }
      }
      .useForever

Let me know if this works and how ( just channel ? )

from fs2-rabbit.