Add method for reconnection (#160)

* Add method for reconnection

* Added comments for tests

Author: wey1and

src/main/java/io/tarantool/driver/api/TarantoolClient.java

@@ -26,19 +26,22 @@ public interface TarantoolClient<T extends Packable, R extends Collection<T>>
 
     /**
      * Get the Tarantool client config passed to this client
+     *
      * @return {@link TarantoolClientConfig} instance
      */
     TarantoolClientConfig getConfig();
 
     /**
      * Get the Tarantool server version
+     *
      * @return {@link TarantoolVersion}
      * @throws TarantoolClientException if the client is not connected
      */
     TarantoolVersion getVersion() throws TarantoolClientException;
 
     /**
      * Provides CRUD and other operations for a Tarantool space
+     *
      * @param spaceName name of the space, must not be null or empty
      * @return Tarantool space operations interface
      * @throws TarantoolClientException if the client is not connected
@@ -47,6 +50,7 @@ public interface TarantoolClient<T extends Packable, R extends Collection<T>>
 
     /**
      * Provides CRUD and other operations for a Tarantool space
+     *
      * @param spaceId ID of the space, must be greater than 0
      * @return Tarantool space operations implementation
      * @throws TarantoolClientException if the client is not connected
@@ -55,6 +59,7 @@ public interface TarantoolClient<T extends Packable, R extends Collection<T>>
 
     /**
      * Provides operations for Tarantool spaces and indexes metadata
+     *
      * @return Tarantool metadata operations implementation
      * @throws TarantoolClientException if the client is not connected
      */
@@ -67,4 +72,11 @@ public interface TarantoolClient<T extends Packable, R extends Collection<T>>
      * @return connection listeners
      */
     TarantoolConnectionListeners getConnectionListeners();
+
+    /**
+     * Starts the process of establishing lacking connections to each host
+     *
+     * @return returns true if the establishing process has been started, else false
+     */
+    boolean establishLackingConnections();
 }

src/main/java/io/tarantool/driver/core/AbstractTarantoolClient.java

@@ -148,6 +148,11 @@ private TarantoolConnectionManager connectionManager() {
         return this.connectionManager;
     }
 
+    @Override
+    public boolean establishLackingConnections() {
+        return connectionManager.establishLackingConnections();
+    }
+
     @Override
     public TarantoolVersion getVersion() throws TarantoolClientException {
         try {

src/main/java/io/tarantool/driver/core/ProxyTarantoolClient.java

@@ -416,6 +416,11 @@ public CompletableFuture<List<?>> eval(String expression,
         return client.eval(expression, arguments, argumentsMapper, resultMapper);
     }
 
+    @Override
+    public boolean establishLackingConnections() {
+        return this.client.establishLackingConnections();
+    }
+
     @Override
     public void close() throws Exception {
         this.client.close();

src/main/java/io/tarantool/driver/core/RetryingTarantoolClient.java

@@ -378,6 +378,11 @@ public CompletableFuture<List<?>> eval(String expression, List<?> arguments,
         return wrapOperation(() -> client.eval(expression, arguments, argumentsMapper, resultMapper));
     }
 
+    @Override
+    public boolean establishLackingConnections() {
+        return this.client.establishLackingConnections();
+    }
+
     @Override
     public void close() throws Exception {
         client.close();

src/main/java/io/tarantool/driver/core/connection/AbstractTarantoolConnectionManager.java

@@ -104,6 +104,11 @@ public CompletableFuture<TarantoolConnection> getConnection() {
         });
     }
 
+    @Override
+    public boolean establishLackingConnections() {
+        return connectionMode.compareAndSet(ConnectionMode.OFF, ConnectionMode.PARTIAL);
+    }
+
     private CompletableFuture<TarantoolConnection> getConnectionInternal() {
         CompletableFuture<TarantoolConnection> result;
         ConnectionMode currentMode = connectionMode.get();

src/main/java/io/tarantool/driver/core/connection/TarantoolConnectionManager.java

@@ -18,4 +18,11 @@ public interface TarantoolConnectionManager extends AutoCloseable {
      * @return a future with next connection in order
      */
     CompletableFuture<TarantoolConnection> getConnection();
+
+    /**
+     * Starts the process of establishing lacking connections to each host
+     *
+     * @return returns true if the establishing process has been started, else false
+     */
+    boolean establishLackingConnections();
 }

src/test/java/io/tarantool/driver/integration/ClusterDiscoveryIT.java

@@ -42,7 +42,7 @@ public void httpClusterDiscovererTest() throws TarantoolClientException {
         HTTPDiscoveryClusterAddressProvider addressProvider = getHttpProvider();
         Collection<TarantoolServerAddress> nodes = addressProvider.getAddresses();
 
-        assertEquals(nodes.size(), 2);
+        assertEquals(nodes.size(), 3);
         Set<TarantoolServerAddress> nodeSet = new HashSet<>(nodes);
         assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER1_URI)));
         assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER2_URI)));
@@ -68,7 +68,7 @@ public void binaryClusterDiscovererTest() {
         TarantoolClusterAddressProvider addressProvider = getBinaryProvider();
 
         Collection<TarantoolServerAddress> nodes = addressProvider.getAddresses();
-        assertEquals(nodes.size(), 2);
+        assertEquals(nodes.size(), 3);
         Set<TarantoolServerAddress> nodeSet = new HashSet<>(nodes);
         assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER1_URI)));
         assertTrue(nodeSet.contains(new TarantoolServerAddress(TEST_ROUTER2_URI)));

src/test/java/io/tarantool/driver/integration/ReconnectIT.java

@@ -0,0 +1,98 @@
+package io.tarantool.driver.integration;
+
+import io.tarantool.driver.api.TarantoolClient;
+import io.tarantool.driver.api.TarantoolClientFactory;
+import io.tarantool.driver.api.TarantoolResult;
+import io.tarantool.driver.api.TarantoolServerAddress;
+import io.tarantool.driver.api.tuple.TarantoolTuple;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.output.WaitingConsumer;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeoutException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@Testcontainers
+public class ReconnectIT extends SharedCartridgeContainer {
+
+    private static String USER_NAME;
+    private static String PASSWORD;
+
+    @BeforeAll
+    public static void setUp() throws TimeoutException {
+        startCluster();
+
+        WaitingConsumer waitingConsumer = new WaitingConsumer();
+        container.followOutput(waitingConsumer);
+        waitingConsumer.waitUntil(f -> f.getUtf8String().contains("The cluster is balanced ok"));
+
+        USER_NAME = container.getUsername();
+        PASSWORD = container.getPassword();
+    }
+
+    @Test
+    public void test_should_reconnect_ifReconnectIsInvoked() throws Exception {
+        //when
+        TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> client =
+                TarantoolClientFactory.createClient()
+                        .withAddresses(
+                                new TarantoolServerAddress(container.getRouterHost(), container.getMappedPort(3301)),
+                                new TarantoolServerAddress(container.getRouterHost(), container.getMappedPort(3311)),
+                                new TarantoolServerAddress(container.getRouterHost(), container.getMappedPort(3312))
+                        )
+                        .withCredentials(USER_NAME, PASSWORD)
+                        .withConnections(10)
+                        .build();
+
+        // getting all routers uuids
+        final Set<String> routerUuids = getInstancesUuids(client);
+
+        // stop routers
+        container.execInContainer("cartridge", "stop", "--run-dir=/tmp/run", "router");
+        container.execInContainer("cartridge", "stop", "--run-dir=/tmp/run", "second-router");
+
+        // check that there is only one instance left
+        assertEquals(getInstanceUuid(client), getInstanceUuid(client));
+
+        // start routers
+        container.execInContainer("cartridge", "start", "--run-dir=/tmp/run", "--data-dir=/tmp/data", "-d");
+
+        client.establishLackingConnections();
+        Thread.sleep(3000);
+
+        // getting all routers uuids after restarting
+        final Set<String> uuidsAfterReconnect = getInstancesUuids(client);
+
+        // check that amount of routers is equal initial amount
+        assertEquals(routerUuids.size(), uuidsAfterReconnect.size());
+    }
+
+    /**
+     * Return all instances uuids from cluster, using round robin connection selection strategy
+     *
+     * @param client Tarantool client
+     * @return set of instances uuids from cluster
+     */
+    private Set<String> getInstancesUuids(TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> client) {
+        String firstUuid = getInstanceUuid(client);
+
+        final Set<String> routerUuids = new HashSet<>();
+        routerUuids.add(firstUuid);
+
+        String currentUuid = "";
+        while (!firstUuid.equals(currentUuid)) {
+            currentUuid = getInstanceUuid(client);
+            routerUuids.add(currentUuid);
+        }
+
+        return routerUuids;
+    }
+
+    private String getInstanceUuid(TarantoolClient<TarantoolTuple, TarantoolResult<TarantoolTuple>> client) {
+        return (String) client.eval("return box.info().uuid").join().get(0);
+    }
+}

src/test/resources/cartridge/instances.yml

@@ -8,6 +8,11 @@ testapp.second-router:
   advertise_uri: localhost:3311
   http_port: 8091
 
+testapp.third-router:
+  workdir: ./tmp/db_dev/3312
+  advertise_uri: localhost:3312
+  http_port: 8092
+
 testapp.s1-master:
   workdir: ./tmp/db_dev/3302
   advertise_uri: localhost:3302

src/test/resources/cartridge/topology.lua

@@ -1,19 +1,23 @@
 cartridge = require('cartridge')
-replicasets = {{
-    alias = 'app-router',
-    roles = {'vshard-router', 'app.roles.custom', 'app.roles.api_router'},
-    join_servers = {{uri = 'localhost:3301'}}
-}, {
-    alias = 'app-router-second',
-    roles = {'vshard-router', 'app.roles.custom', 'app.roles.api_router'},
-    join_servers = {{uri = 'localhost:3311'}}
-}, {
-    alias = 's1-storage',
-    roles = {'vshard-storage', 'app.roles.api_storage'},
-    join_servers = {{uri = 'localhost:3302'}, {uri = 'localhost:3303'}}
-}, {
-    alias = 's2-storage',
-    roles = {'vshard-storage', 'app.roles.api_storage'},
-    join_servers = {{uri = 'localhost:3304'}, {uri = 'localhost:3305'}}
-}}
-return cartridge.admin_edit_topology({replicasets = replicasets})
+replicasets = { {
+                    alias = 'app-router',
+                    roles = { 'vshard-router', 'app.roles.custom', 'app.roles.api_router' },
+                    join_servers = { { uri = 'localhost:3301' } }
+                }, {
+                    alias = 'app-router-second',
+                    roles = { 'vshard-router', 'app.roles.custom', 'app.roles.api_router' },
+                    join_servers = { { uri = 'localhost:3311' } }
+                }, {
+                    alias = 'app-router-third',
+                    roles = { 'vshard-router', 'app.roles.custom', 'app.roles.api_router' },
+                    join_servers = { { uri = 'localhost:3312' } }
+                }, {
+                    alias = 's1-storage',
+                    roles = { 'vshard-storage', 'app.roles.api_storage' },
+                    join_servers = { { uri = 'localhost:3302' }, { uri = 'localhost:3303' } }
+                }, {
+                    alias = 's2-storage',
+                    roles = { 'vshard-storage', 'app.roles.api_storage' },
+                    join_servers = { { uri = 'localhost:3304' }, { uri = 'localhost:3305' } }
+                } }
+return cartridge.admin_edit_topology({ replicasets = replicasets })