provides a reconnectable Mono<RSocket> feature

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

Author: OlegDokuka

rsocket-core/build.gradle

@@ -43,6 +43,7 @@ dependencies {
     // TODO: Remove after JUnit5 migration
     testCompileOnly 'junit:junit'
     testImplementation 'org.hamcrest:hamcrest-library'
+    testImplementation 'io.projectreactor.addons:reactor-extra'
     testRuntimeOnly 'org.junit.vintage:junit-vintage-engine'
 }
 

rsocket-core/src/main/java/io/rsocket/Invalidate.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2015-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.rsocket;
+
+interface Invalidate {
+
+  void expire();
+}

rsocket-core/src/main/java/io/rsocket/RSocketFactory.java

@@ -47,6 +47,9 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import org.reactivestreams.Publisher;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 /** Factory for creating RSocket clients and servers. */
@@ -95,6 +98,9 @@ default <T extends Closeable> Start<T> transport(ServerTransport<T> transport) {
   public static class ClientRSocketFactory implements ClientTransportAcceptor {
     private static final String CLIENT_TAG = "client";
 
+    private static final Function<Flux<Throwable>, ? extends Publisher<?>> FAIL_WHEN_FACTORY =
+        f -> f.concatMap(Mono::error);
+
     private SocketAcceptor acceptor = (setup, sendingSocket) -> Mono.just(new AbstractRSocket() {});
 
     private Consumer<Throwable> errorConsumer = Throwable::printStackTrace;
@@ -125,6 +131,8 @@ public static class ClientRSocketFactory implements ClientTransportAcceptor {
     private boolean multiSubscriberRequester = true;
     private boolean leaseEnabled;
     private Supplier<Leases<?>> leasesSupplier = Leases::new;
+    private boolean reconnectEnabled;
+    private Function<Flux<Throwable>, ? extends Publisher<?>> whenFactory = FAIL_WHEN_FACTORY;
 
     private ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
 
@@ -230,6 +238,129 @@ public ClientRSocketFactory singleSubscriberRequester() {
       return this;
     }
 
+    /**
+     * Enables a reconnectable, shared instance of {@code Mono<RSocket>} so every subscriber will
+     * observe the same RSocket instance up on connection establishment. <br>
+     * For example:
+     *
+     * <pre>{@code
+     * Mono<RSocket> sharedRSocketMono =
+     *   RSocketFactory
+     *                .connect()
+     *                .singleSubscriberRequester()
+     *                .reconnect()
+     *                .transport(transport)
+     *                .start();
+     *
+     *  RSocket r1 = sharedRSocketMono.block();
+     *  RSocket r2 = sharedRSocketMono.block();
+     *
+     *  assert r1 == r2;
+     *
+     * }</pre>
+     *
+     * Apart of the shared behavior, once the previous connection has been expired (e.g. connection
+     * has been disposed), the same instance of {@code Mono<RSocket>} reestablish connection without
+     * any extra effort. <br>
+     * For example:
+     *
+     * <pre>{@code
+     * Mono<RSocket> sharedRSocketMono =
+     *   RSocketFactory
+     *                .connect()
+     *                .singleSubscriberRequester()
+     *                .reconnect()
+     *                .transport(transport)
+     *                .start();
+     *
+     *  RSocket r1 = sharedRSocketMono.block();
+     *  RSocket r2 = sharedRSocketMono.block();
+     *
+     *  assert r1 == r2;
+     *
+     *  r1.dispose()
+     *
+     *  assert r2.isDisposed()
+     *
+     *  RSocket r3 = sharedRSocketMono.block();
+     *  RSocket r4 = sharedRSocketMono.block();
+     *
+     *
+     *  assert r1 != r3;
+     *  assert r4 == r3;
+     *
+     * }</pre>
+     *
+     * @return a shared instance of {@code Mono<RSocket>}.
+     */
+    public ClientRSocketFactory reconnect() {
+      this.reconnectEnabled = true;
+      return this;
+    }
+
+    /**
+     * Enables a reconnectable, shared instance of {@code Mono<RSocket>} so every subscriber will
+     * observe the same RSocket instance up on connection establishment. <br>
+     * For example:
+     *
+     * <pre>{@code
+     * Mono<RSocket> sharedRSocketMono =
+     *   RSocketFactory
+     *                .connect()
+     *                .singleSubscriberRequester()
+     *                .reconnect()
+     *                .transport(transport)
+     *                .start();
+     *
+     *  RSocket r1 = sharedRSocketMono.block();
+     *  RSocket r2 = sharedRSocketMono.block();
+     *
+     *  assert r1 == r2;
+     *
+     * }</pre>
+     *
+     * Apart of the shared behavior, once the previous connection has been expired (e.g. connection
+     * has been disposed), the same instance of {@code Mono<RSocket>} reestablish connection without
+     * any extra effort. <br>
+     * For example:
+     *
+     * <pre>{@code
+     * Mono<RSocket> sharedRSocketMono =
+     *   RSocketFactory
+     *                .connect()
+     *                .singleSubscriberRequester()
+     *                .reconnect()
+     *                .transport(transport)
+     *                .start();
+     *
+     *  RSocket r1 = sharedRSocketMono.block();
+     *  RSocket r2 = sharedRSocketMono.block();
+     *
+     *  assert r1 == r2;
+     *
+     *  r1.dispose()
+     *
+     *  assert r2.isDisposed()
+     *
+     *  RSocket r3 = sharedRSocketMono.block();
+     *  RSocket r4 = sharedRSocketMono.block();
+     *
+     *
+     *  assert r1 != r3;
+     *  assert r4 == r3;
+     *
+     * }</pre>
+     *
+     * @param whenFactory a retry factory applied for {@link Mono.retryWhen(whenFactory)}
+     * @return a shared instance of {@code Mono<RSocket>}.
+     */
+    public ClientRSocketFactory reconnect(
+        Function<Flux<Throwable>, ? extends Publisher<?>> whenFactory) {
+      this.whenFactory = Objects.requireNonNull(whenFactory);
+      this.reconnectEnabled = true;
+      return this;
+    }
+
     public ClientRSocketFactory resume() {
       this.resumeEnabled = true;
       return this;
@@ -392,6 +523,17 @@ public Mono<RSocket> start() {
                                 .sendOne(setupFrame)
                                 .thenReturn(wrappedRSocketRequester);
                           });
+                })
+            .as(
+                source -> {
+                  if (reconnectEnabled) {
+                    return new ReconnectMono<>(
+                        source.retryWhen(whenFactory),
+                        Disposable::dispose,
+                        (r, i) -> r.onClose().subscribe(null, null, i::expire));
+                  } else {
+                    return source;
+                  }
                 });
       }
 
@@ -422,7 +564,7 @@ private ClientSetup clientSetup(DuplexConnection startConnection) {
       }
 
       private Mono<DuplexConnection> newConnection() {
-        return transportClient.get().connect(mtu);
+        return Mono.fromSupplier(transportClient).flatMap(t -> t.connect(mtu));
       }
     }
   }