Add support for tarantool clusters

Refactor SocketChannelProvider implementations. Now we have two
SingleSocketChannelProviderImpl and RoundRobinSocketProviderImpl used by
simple and cluster clients respectively. To achieve this a
BaseSocketChannelProvider was extracted.

Add a service discovery implementation based on a tarantool stored
procedure which is called to obtain a new list of cluster nodes.

Integrate service discovery and current cluster client. The client now
is aware of potential nodes changing using a configurable background job
which periodically checks whether node addresses have changed. If so
the client refreshes the RoundRobinSocketProviderImpl and replaces old
nodes by new ones. It also requires some additional effort in case of
missing the current node in the list. We need to reconnect to another
node as soon as possible with a minimal delay between client requests.
To achieve this we currently try to catch a moment when the requests in
progress have been accomplished and we can finish reconnection process.

Fix a regression in TarantoolClientImpl. It is a wrong comparison
between response result code and original request operation code. To
perform a right thing TarantoolOp class was created to wrap an original
future (see TarantoolClientImpl.complete(packet, feature)).

Closes: 34

Author: nicktorwald

src/main/java/org/tarantool/BaseSocketChannelProvider.java

@@ -0,0 +1,119 @@
+package org.tarantool;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.SocketChannel;
+
+public abstract class BaseSocketChannelProvider implements SocketChannelProvider {
+
+    /**
+     * Limit of retries.
+     */
+    private int retriesLimit = RETRY_NO_LIMIT;
+
+    /**
+     * Timeout to establish socket connection with an individual server.
+     */
+    private int timeout = NO_TIMEOUT;
+
+    @Override
+    public final SocketChannel get(int retryNumber, Throwable lastError) {
+        if (areRetriesExhausted(retryNumber)) {
+            throw new CommunicationException("Connection retries exceeded.", lastError);
+        }
+
+        return doRetry(retryNumber, lastError);
+    }
+
+    protected abstract SocketChannel doRetry(int retryNumber, Throwable lastError);
+
+    /**
+     * 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 attempts.
+     *
+     * @param retriesLimit Limit of retries to use.
+     */
+    public void setRetriesLimit(int retriesLimit) {
+        this.retriesLimit = retriesLimit;
+    }
+
+    /**
+     * @return Maximum reconnect attempts to make before raising exception.
+     */
+    public int getRetriesLimit() {
+        return retriesLimit;
+    }
+
+    /**
+     * Parse a string address in the form of [host]:[port]
+     * and builds a socket address.
+     *
+     * @param address Server address.
+     * @return Socket address.
+     */
+    protected InetSocketAddress parseAddress(String address) {
+        int idx = address.indexOf(':');
+        String host = (idx < 0) ? address : address.substring(0, idx);
+        int port = (idx < 0) ? 3301 : Integer.parseInt(address.substring(idx + 1));
+        return new InetSocketAddress(host, port);
+    }
+
+    protected SocketChannel openChannel(InetSocketAddress socketAddress) throws IOException {
+        SocketChannel channel = null;
+        try {
+            channel = SocketChannel.open();
+            channel.socket().connect(socketAddress, timeout);
+            return channel;
+        } catch (IOException e) {
+            if (channel != null) {
+                try {
+                    channel.close();
+                } catch (IOException ignored) {
+                    // No-op.
+                }
+            }
+            throw e;
+        }
+    }
+
+    /**
+     * Sets maximum amount of time to wait for a socket connection establishment
+     * with an individual server.
+     * <p>
+     * Zero means infinite timeout.
+     *
+     * @param timeout timeout value, ms.
+     * @throws IllegalArgumentException if timeout is negative.
+     */
+    public void setTimeout(int timeout) {
+        if (timeout < 0) {
+            throw new IllegalArgumentException("timeout is negative.");
+        }
+        this.timeout = timeout;
+    }
+
+    /**
+     * @return Maximum amount of time to wait for a socket connection establishment
+     * with an individual server.
+     */
+    public int getTimeout() {
+        return timeout;
+    }
+
+    /**
+     * 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/RefreshableSocketProvider.java

@@ -0,0 +1,12 @@
+package org.tarantool;
+
+import java.net.SocketAddress;
+import java.util.Collection;
+
+public interface RefreshableSocketProvider {
+
+    Collection<SocketAddress> getAddresses();
+
+    void refreshAddresses(Collection<String> addresses);
+
+}

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

@@ -2,123 +2,106 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
 
 /**
  * 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;
+public class RoundRobinSocketProviderImpl extends BaseSocketChannelProvider implements RefreshableSocketProvider {
 
     /**
-     * Constructs an instance.
-     *
-     * @param addrs Array of addresses in a form of [host]:[port].
+     * Socket addresses.
      */
-    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);
+    private final List<InetSocketAddress> socketAddresses = new ArrayList<>();
 
-        sockAddrs = new InetSocketAddress[this.addrs.length];
-
-        for (int i = 0; i < this.addrs.length; i++) {
-            sockAddrs[i] = parseAddress(this.addrs[i]);
-        }
-    }
+    /**
+     * Current position within {@link #socketAddresses} array.
+     */
+    private AtomicInteger currentPosition = new AtomicInteger(-1);
 
     /**
-     * @return Configured addresses in a form of [host]:[port].
+     * Lock
      */
-    public String[] getAddresses() {
-        return this.addrs;
-    }
+    private ReadWriteLock addressListLock = new ReentrantReadWriteLock();
 
     /**
-     * Sets maximum amount of time to wait for a socket connection establishment
-     * with an individual server.
-     *
-     * Zero means infinite timeout.
+     * Constructs an instance.
      *
-     * @param timeout Timeout value, ms.
-     * @return {@code this}.
-     * @throws IllegalArgumentException If timeout is negative.
+     * @param addresses Array of addresses in a form of [host]:[port].
      */
-    public RoundRobinSocketProviderImpl setTimeout(int timeout) {
-        if (timeout < 0)
-            throw new IllegalArgumentException("timeout is negative.");
-
-        this.timeout = timeout;
+    public RoundRobinSocketProviderImpl(String... addresses) {
+        if (addresses.length == 0) {
+            throw new IllegalArgumentException("Addresses list must contain at least one address.");
+        }
 
-        return this;
+        updateAddressList(Arrays.asList(addresses));
     }
 
-    /**
-     * @return Maximum amount of time to wait for a socket connection establishment
-     *         with an individual server.
-     */
-    public int getTimeout() {
-        return timeout;
+    private void updateAddressList(Collection<String> addresses) {
+        Lock writeLock = addressListLock.writeLock();
+        writeLock.lock();
+        try {
+            InetSocketAddress lastAddress = getLastObtainedAddress();
+            socketAddresses.clear();
+            addresses.stream()
+                    .map(this::parseAddress)
+                    .collect(Collectors.toCollection(() -> socketAddresses));
+            if (lastAddress != null) {
+                int recoveredPosition = socketAddresses.indexOf(lastAddress);
+                currentPosition.set(recoveredPosition);
+            } else {
+                currentPosition.set(-1);
+            }
+        } finally {
+            writeLock.unlock();
+        }
     }
 
     /**
-     * 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}.
+     * @return resolved socket addresses
      */
-    public RoundRobinSocketProviderImpl setRetriesLimit(int retriesLimit) {
-        this.retriesLimit = retriesLimit;
-
-        return this;
+    public List<SocketAddress> getAddresses() {
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            return Collections.unmodifiableList(this.socketAddresses);
+        } finally {
+            readLock.unlock();
+        }
     }
 
-    /**
-     * @return Maximum reconnect attempts to make before raising exception.
-     */
-    public int getRetriesLimit() {
-        return retriesLimit;
+    protected InetSocketAddress getLastObtainedAddress() {
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            int index = currentPosition.get();
+            return index >= 0 ? socketAddresses.get(index) : null;
+        } finally {
+            readLock.unlock();
+        }
     }
 
-    /** {@inheritDoc} */
     @Override
-    public SocketChannel get(int retryNumber, Throwable lastError) {
-        if (areRetriesExhausted(retryNumber)) {
-            throw new CommunicationException("Connection retries exceeded.", lastError);
-        }
+    protected SocketChannel doRetry(int retryNumber, Throwable lastError) {
         int attempts = getAddressCount();
-        long deadline = System.currentTimeMillis() + timeout * attempts;
+        // todo: recalc deadline?
+        long deadline = System.currentTimeMillis() + getTimeout() * attempts;
         while (!Thread.currentThread().isInterrupted()) {
-            SocketChannel channel = null;
             try {
-                channel = SocketChannel.open();
-                InetSocketAddress addr = getNextSocketAddress();
-                channel.socket().connect(addr, timeout);
-                return channel;
+                return openChannel(getNextSocketAddress());
             } 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);
@@ -141,42 +124,33 @@ public SocketChannel get(int retryNumber, Throwable lastError) {
      * @return Number of configured addresses.
      */
     protected int getAddressCount() {
-        return sockAddrs.length;
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            return socketAddresses.size();
+        } finally {
+            readLock.unlock();
+        }
     }
 
     /**
      * @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);
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            int position = currentPosition.updateAndGet(i -> (i + 1) % socketAddresses.size());
+            return socketAddresses.get(position);
+        } finally {
+            readLock.unlock();
+        }
     }
 
-    /**
-     * 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;
+    public void refreshAddresses(Collection<String> addresses) {
+        if (addresses == null || addresses.isEmpty()) {
+            throw new IllegalArgumentException("Addresses are null or empty.");
+        }
+        updateAddressList(addresses);
     }
 }