Replace AutoCloseable with reactor's Disposable. Retry and timeout fixes

[conniey]

• Replace AutoCloseable with Disposable.
  • In the next PR, if the parent is already disposed, we don't want to re-create its child. AutoCloseable doesn't have a way to query the current status of an AMQP channel.
  • Example: If a connection is closed, we don't want to re-create a session from that channel. And if a session is closed, we don't want to create links from it.
• Fixes bug where when a transient error occurs, and while waiting for the retry period to elapse, if a new subscriber comes in, it'll immediately try to get a new connection, rather than waiting for that retry period to elapse.
• Fixes bug where TimeoutException would occur in doOnEach.

<10484.19> 01/12/2020 23:51:54 - Error: JAVA-Sender#0<stdout>: : Client index 0: Error occurred. java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 60000ms in 'doOnEach' (and no fallback has been configured)
<10484.19> 01/12/2020 23:51:54 - Error: JAVA-Sender#0<stdout>:  Exception caught (Identifier: Client index 0): class java.util.concurrent.TimeoutException Did not observe any item or terminal signal within 60000ms in 'doOnEach' (and no fallback has been configured)

[JonathanGiles]

This looks like a breaking API change? Has this not been published as a GA release yet?

[conniey]

@JonathanGiles No. This is still a -beta.6. It has not been GA'd yet. We're hoping to make it GA on Friday with 1.0.1.