AMQP Fix #3023

[bturos]

Offer new cached connection when previous one has been closed and released

References #3023

ℹ️ I'm not sure whether abort is the best way to simulate connection closed by the server. Hopefully, for the purpose of unit test, this is good enough

[ennru]

Scala 3 tries to apply the tail recursiveness even to provider.get. Moving provider.get into a separate method and calling that to retrieve the connection from the provider should convince the compiler this is OK to do.

[error] -- Error: /home/runner/work/alpakka/alpakka/amqp/src/main/scala/akka/stream/alpakka/amqp/AmqpConnectionProvider.scala:375:27 
[error] 375 |          val connection = provider.get
[error]     |                           ^^^^^^^^^^^^
[error]     |                 Cannot rewrite recursive call: it targets a supertype

[ennru]

Thank you!
LGTM.

It is still draft, do you expect to change something more?

[bturos]

Thank you! LGTM.

It is still draft, do you expect to change something more?

Nope, it was just a measure of precaution in case something else pops up. Moving to Open, thank you for reviewing :)