add round robin fail-over for a client

Added cluster-ready client implementation that
would fail-over to the next configured server in a
round-robin fashion.

Closes tarantool#37

Author: ztarvos

.travis.yml

@@ -18,7 +18,7 @@ script:
     else
         mvn verify
     fi
-  - cat testroot/jdk-testing.log
+  - head -n -0 testroot/*.log
 
 after_success:
   - |

pom.xml

@@ -62,6 +62,9 @@
                         </goals>
                     </execution>
                 </executions>
+                <configuration>
+                    <trimStackTrace>false</trimStackTrace>
+                </configuration>
             </plugin>
             <plugin>
                 <groupId>org.jacoco</groupId>

src/main/java/org/tarantool/RoundRobinSocketProviderImpl.java

@@ -0,0 +1,182 @@
+package org.tarantool;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.SocketChannel;
+import java.util.Arrays;
+
+/**
+ * Basic reconnection strategy that changes addresses in a round-robin fashion.
+ * To be used with {@link TarantoolClientImpl}.
+ */
+public class RoundRobinSocketProviderImpl implements SocketChannelProvider {
+    /** Timeout to establish socket connection with an individual server. */
+    private int timeout; // 0 is infinite.
+    /** Limit of retries. */
+    private int retriesLimit = -1; // No-limit.
+    /** Server addresses as configured. */
+    private final String[] addrs;
+    /** Socket addresses. */
+    private final InetSocketAddress[] sockAddrs;
+    /** Current position within {@link #sockAddrs} array. */
+    private int pos;
+
+    /**
+     * Constructs an instance.
+     *
+     * @param addrs Array of addresses in a form of [host]:[port].
+     */
+    public RoundRobinSocketProviderImpl(String... addrs) {
+        if (addrs == null || addrs.length == 0)
+            throw new IllegalArgumentException("addrs is null or empty.");
+
+        this.addrs = Arrays.copyOf(addrs, addrs.length);
+
+        sockAddrs = new InetSocketAddress[this.addrs.length];
+
+        for (int i = 0; i < this.addrs.length; i++) {
+            sockAddrs[i] = parseAddress(this.addrs[i]);
+        }
+    }
+
+    /**
+     * @return Configured addresses in a form of [host]:[port].
+     */
+    public String[] getAddresses() {
+        return this.addrs;
+    }
+
+    /**
+     * Sets maximum amount of time to wait for a socket connection establishment
+     * with an individual server.
+     *
+     * Zero means infinite timeout.
+     *
+     * @param timeout Timeout value, ms.
+     * @return {@code this}.
+     * @throws IllegalArgumentException If timeout is negative.
+     */
+    public RoundRobinSocketProviderImpl setTimeout(int timeout) {
+        if (timeout < 0)
+            throw new IllegalArgumentException("timeout is negative.");
+
+        this.timeout = timeout;
+
+        return this;
+    }
+
+    /**
+     * @return Maximum amount of time to wait for a socket connection establishment
+     *         with an individual server.
+     */
+    public int getTimeout() {
+        return timeout;
+    }
+
+    /**
+     * Sets maximum amount of reconnect attempts to be made before an exception is raised.
+     * The retry count is maintained by a {@link #get(int, Throwable)} caller
+     * when a socket level connection was established.
+     *
+     * Negative value means unlimited.
+     *
+     * @param retriesLimit Limit of retries to use.
+     * @return {@code this}.
+     */
+    public RoundRobinSocketProviderImpl setRetriesLimit(int retriesLimit) {
+        this.retriesLimit = retriesLimit;
+
+        return this;
+    }
+
+    /**
+     * @return Maximum reconnect attempts to make before raising exception.
+     */
+    public int getRetriesLimit() {
+        return retriesLimit;
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public SocketChannel get(int retryNumber, Throwable lastError) {
+        if (areRetriesExhausted(retryNumber)) {
+            throw new CommunicationException("Connection retries exceeded.", lastError);
+        }
+        int attempts = getAddressCount();
+        long deadline = System.currentTimeMillis() + timeout * attempts;
+        while (!Thread.currentThread().isInterrupted()) {
+            SocketChannel channel = null;
+            try {
+                channel = SocketChannel.open();
+                InetSocketAddress addr = getNextSocketAddress();
+                channel.socket().connect(addr, timeout);
+                return channel;
+            } catch (IOException e) {
+                if (channel != null) {
+                    try {
+                        channel.close();
+                    } catch (IOException ignored) {
+                        // No-op.
+                    }
+                }
+                long now = System.currentTimeMillis();
+                if (deadline <= now) {
+                    throw new CommunicationException("Connection time out.", e);
+                }
+                if (--attempts == 0) {
+                    // Tried all addresses without any lack, but still have time.
+                    attempts = getAddressCount();
+                    try {
+                        Thread.sleep((deadline - now) / attempts);
+                    } catch (InterruptedException ignored) {
+                        Thread.currentThread().interrupt();
+                    }
+                }
+            }
+        }
+        throw new CommunicationException("Thread interrupted.", new InterruptedException());
+    }
+
+    /**
+     * @return Number of configured addresses.
+     */
+    protected int getAddressCount() {
+        return sockAddrs.length;
+    }
+
+    /**
+     * @return Socket address to use for the next reconnection attempt.
+     */
+    protected InetSocketAddress getNextSocketAddress() {
+        InetSocketAddress res = sockAddrs[pos];
+        pos = (pos + 1) % sockAddrs.length;
+        return res;
+    }
+
+    /**
+     * Parse a string address in the form of [host]:[port]
+     * and builds a socket address.
+     *
+     * @param addr Server address.
+     * @return Socket address.
+     */
+    protected InetSocketAddress parseAddress(String addr) {
+        int idx = addr.indexOf(':');
+        String host = (idx < 0) ? addr : addr.substring(0, idx);
+        int port = (idx < 0) ? 3301 : Integer.parseInt(addr.substring(idx + 1));
+        return new InetSocketAddress(host, port);
+    }
+
+    /**
+     * Provides a decision on whether retries limit is hit.
+     *
+     * @param retries Current count of retries.
+     * @return {@code true} if retries are exhausted.
+     */
+    private boolean areRetriesExhausted(int retries) {
+        int limit = getRetriesLimit();
+        if (limit < 0)
+            return false;
+        return retries >= limit;
+    }
+}

src/main/java/org/tarantool/TarantoolClientImpl.java

@@ -246,14 +246,15 @@ protected synchronized void die(String message, Exception cause) {
         if (thumbstone != null) {
             return;
         }
-        this.thumbstone = new CommunicationException(message, cause);
+        final CommunicationException err = new CommunicationException(message, cause);
+        this.thumbstone = err;
         while (!futures.isEmpty()) {
             Iterator<Map.Entry<Long, FutureImpl<?>>> iterator = futures.entrySet().iterator();
             while (iterator.hasNext()) {
                 Map.Entry<Long, FutureImpl<?>> elem = iterator.next();
                 if (elem != null) {
                     FutureImpl<?> future = elem.getValue();
-                    fail(future, cause);
+                    fail(future, err);
                 }
                 iterator.remove();
             }
@@ -606,6 +607,14 @@ protected boolean isDead(FutureImpl<?> q) {
         return false;
     }
 
+    /**
+     * A subclass may use this as a trigger to start retries.
+     * This method is called when state becomes ALIVE.
+     */
+    protected void onReconnect() {
+        // No-op, override.
+    }
+
     public Exception getThumbstone() {
         return thumbstone;
     }
@@ -679,6 +688,7 @@ protected boolean compareAndSet(int expect, int update) {
             if (update == ALIVE) {
                 CountDownLatch latch = nextAliveLatch.getAndSet(new CountDownLatch(1));
                 latch.countDown();
+                onReconnect();
             } else if (update == CLOSED) {
                 closedLatch.countDown();
             }
@@ -706,7 +716,9 @@ private CountDownLatch getStateLatch(int state) {
                     throw new IllegalStateException("State is CLOSED.");
                 }
                 CountDownLatch latch = nextAliveLatch.get();
-                return  (getState() == ALIVE) ? null : latch;
+                /* It may happen so that an error is detected but the state is still alive.
+                 Wait for the 'next' alive state in such cases. */
+                return  (getState() == ALIVE && thumbstone == null) ? null : latch;
             }
             return null;
         }