Support for reconnectable, shared Mono<RSocket>

[OlegDokuka]

This PR enables a reconnectable RSocket Mono feature which allows to:

1. Have shared Mono instance without extra effort (extra effort == mono.cache())
2. Have allocation free blocking (ReconnectMono#block) as in the case of MonoProcessor
3. Have a Mono which reestablishes connection once the previous has been expired

Mono<RSocket> sharedRSocketMono =          
  RSocketFactory                           
               .connect()                  
               .singleSubscriberRequester()
               .reconnect()                
               .transport(transport)       
               .start();                   
                                           
 RSocket r1 = sharedRSocketMono.block();   // subscribes first so the connection establishement is initiated 
 RSocket r2 = sharedRSocketMono.block();   // attached to the previous process
                                           
 assert r1 == r2;    // same reference                       
                                           
 r1.dispose(); // disposes instance so it is not valid anymore                              
                                           
 assert r2.isDisposed()                    
                                           
 RSocket r3 = sharedRSocketMono.block();   // reistablish a new connection
 RSocket r4 = sharedRSocketMono.block();   // attached to the previous process
                                           
                                           
 assert r1 != r3; // new instance is give (ref check is not valid)                         
 assert r4 == r3; // r3 and r4 are ref equal 
                 
 assert r2.isDisposed(); // previous is still disposed
 assert !r4.isDisposed(); // a new one is healthy 	
                        

Signed-off-by: Oleh Dokuka [email protected]

[OlegDokuka]

cc\ @rstoyanchev @simonbasle @linux-china @smaldini @yschimke for your review folks

[simonbasle]

some minor points on the API, otherwise looks good to me

[rstoyanchev]

Looks great, exposed through ClientRSocketFactory as opposed to the other alternatives considered.

I'm wondering about using the new Retry from reactor-core which has some very nice built-in options.

For reconnect(), is it too simplistic maybe? Note that since the new Retry exposes a number of strategies it's much easier to configure. Perhaps we don't even need a no-arg reconnect() or do we think there is a good default reconnect strategy?

[OlegDokuka]

@rstoyanchev no-args reconnect() is basically fail fast version which does nothing if upstream fails.
I prepared that one if a user does not want to retry at all but wants to have a reconnectable or shared rsocket.

Do you think such an extension (with no retry) is unnecessary?

Note that since the new Retry exposes a

Retry is now a convenient way to do retry, so I'm going to replace function with that one shortly

[rstoyanchev]

I think it would also be worth mentioning or showing that downstream subscribers (i.e. individual requests) are expected to have retry logic, according to their own needs, in order to trigger a reconnect.

[OlegDokuka]

fixed. Please see -> https://github.com/rsocket/rsocket-java/pull/759/files#diff-cee939aabc492fd5530587020ce0a0beR293