provides implementation for reconnect mono

[OlegDokuka]

This PR provides derived implementation of Reaconnectable MonoProcessor which allows reobtaining the value from upstream + doing reconnection in case of error.

Behaviour

1. Subscribe to Mono and notify every Downstream subscriber about the value (the basic MonoProcessor behavior)
2. Subscribe to Mono;
   • In case of complete - reconnect;
   • In case of error - reconnect;
   • Notify Subscribers only if value or Reconnect Exhausted
3. Once the value is obtained there is a possibility to reset (or basically expire) cached value using

.build(/*applyResetOnValueHandler*/(value, callMeWhenResetRunnable) -> ...)

APi via given callMeWhenResetRunnable runnable.
4) Up on reset is called, the first subscription will be delayed depends on the time of the first subscription:
* Subscription happens immeditelly -> delay is defined by the min delay of retry strategy
* Subscription happens after some periode of time -> calculated delay is equal to subscriptionTime - resetTime (if negative -> subscribe immediatelly)

API

final ReconnectMono<RSocket> reconnectableSource =
					Mono.defer(monos::poll)
					    .as(Reconnect::fromSource)
					    .reconnectMax(Long.MAX_VALUE)
					    .exponentialBackoff(Duration.ofSeconds(minBackoff), Duration.ofSeconds(maxBackoff))
					    .timeout(Duration.ofSeconds(timeout))
					    .doOnReconnect(__ -> ...)
					    .build((rsocket, resetRunner) -> {
					        ...
					        rsocket.onClose().subscribe(null, null, resetRunner);
					    });

...
reconnectableSource.dispose()

API is derived from Retry + renaming and extra builder methods such as build
and Reconnect<T, C> doOnDispose(Consumer<? super T> onDispose); which allows disposing value in case of reconnectableSource disposure.

Why not Retry API

I tried to use Retry interface but then I figured out that the functionality that I need (and this functionality is basically calculateBackoff method is not coupled with Retry itself so I can not be sure that the given interface has a proper method and not just custom implementation. In addition, Reconnect does not require that Function<Flux, Flux> logic so it can be a standalone implementation which is builder only and extends a proper method from AbstractRetry

[OlegDokuka]

@rstoyanchev

[OlegDokuka]

We just discussed with @simonbasle that it makes sense to get rid of Retry at all and let the user configure the retry logic for the upstream.

WDYT on that @rstoyanchev?

[rstoyanchev]

It's looking more and more like a candidate for an overloaded cache in core.

[OlegDokuka]

Closing here and moves back to rsocket again :D

[simonbasle]

these two lines don't account for the possibility that the actual Subscriber hasn't necessarily immediately requested upon onSubscribe (even though this is usually the case for Monos).

this should be toAdd.complete(this.value);, as you correctly do in the complete() method.

[OlegDokuka]

Right. Missed that