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

Diff for: src/main/java/org/tarantool/BaseSocketChannelProvider.java

@@ -11,6 +11,11 @@ public abstract class BaseSocketChannelProvider implements SocketChannelProvider
      */
     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)) {
@@ -56,7 +61,7 @@ protected InetSocketAddress parseAddress(String address) {
         return new InetSocketAddress(host, port);
     }
 
-    protected SocketChannel openChannel(InetSocketAddress socketAddress, int timeout) throws IOException {
+    protected SocketChannel openChannel(InetSocketAddress socketAddress) throws IOException {
         SocketChannel channel = null;
         try {
             channel = SocketChannel.open();
@@ -74,6 +79,30 @@ protected SocketChannel openChannel(InetSocketAddress socketAddress, int timeout
         }
     }
 
+    /**
+     * 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.
      *
@@ -82,8 +111,9 @@ protected SocketChannel openChannel(InetSocketAddress socketAddress, int timeout
      */
     private boolean areRetriesExhausted(int retries) {
         int limit = getRetriesLimit();
-        if (limit < 0)
+        if (limit < 0) {
             return false;
+        }
         return retries >= limit;
     }
 }

Diff for: 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);
+
+}

Diff for: src/main/java/org/tarantool/RoundRobinSocketProviderImpl.java

@@ -2,29 +2,24 @@
 
 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 extends BaseSocketChannelProvider {
-
-    /**
-     * Timeout to establish socket connection with an individual server.
-     */
-    private int timeout; // 0 is infinite.
-
-    /**
-     * Server addresses as configured.
-     */
-    private final List<String> addresses = new ArrayList<>();
+public class RoundRobinSocketProviderImpl extends BaseSocketChannelProvider implements RefreshableSocketProvider {
 
     /**
      * Socket addresses.
@@ -39,79 +34,73 @@ public class RoundRobinSocketProviderImpl extends BaseSocketChannelProvider {
     /**
      * Lock
      */
-//    private ReadWriteLock addressListLock = new ReentrantReadWriteLock();
+    private ReadWriteLock addressListLock = new ReentrantReadWriteLock();
 
     /**
      * Constructs an instance.
      *
      * @param addresses Array of addresses in a form of [host]:[port].
      */
     public RoundRobinSocketProviderImpl(String... addresses) {
-        if (addresses == null || addresses.length == 0) {
-            throw new IllegalArgumentException("Addresses are null or empty.");
+        if (addresses.length == 0) {
+            throw new IllegalArgumentException("Addresses list must contain at least one address.");
         }
 
         updateAddressList(Arrays.asList(addresses));
     }
 
     private void updateAddressList(Collection<String> addresses) {
-        String lastAddress = getLastObtainedAddress();
-        this.addresses.clear();
-        this.addresses.addAll(addresses);
-        this.addresses.forEach(address -> socketAddresses.add(parseAddress(address)));
-        if (lastAddress != null) {
-            int recoveredPosition = this.addresses.indexOf(lastAddress);
-            currentPosition.set(recoveredPosition);
+        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();
         }
     }
 
     /**
-     * @return Configured addresses in a form of [host]:[port].
-     */
-    public List<String> getAddresses() {
-        return Collections.unmodifiableList(this.addresses);
-    }
-
-    public String getLastObtainedAddress() {
-        int index = currentPosition.get();
-        return index >= 0 ? addresses.get(index) : null;
-    }
-
-    /**
-     * 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.
-     * @return {@code this}.
-     * @throws IllegalArgumentException If timeout is negative.
+     * @return resolved socket addresses
      */
-    public RoundRobinSocketProviderImpl setTimeout(int timeout) {
-        if (timeout < 0) {
-            throw new IllegalArgumentException("timeout is negative.");
+    public List<SocketAddress> getAddresses() {
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            return Collections.unmodifiableList(this.socketAddresses);
+        } finally {
+            readLock.unlock();
         }
-
-        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;
+    protected InetSocketAddress getLastObtainedAddress() {
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            int index = currentPosition.get();
+            return index >= 0 ? socketAddresses.get(index) : null;
+        } finally {
+            readLock.unlock();
+        }
     }
 
     @Override
     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()) {
             try {
-                return openChannel(getNextSocketAddress(), timeout);
+                return openChannel(getNextSocketAddress());
             } catch (IOException e) {
                 long now = System.currentTimeMillis();
                 if (deadline <= now) {
@@ -135,18 +124,30 @@ protected SocketChannel doRetry(int retryNumber, Throwable lastError) {
      * @return Number of configured addresses.
      */
     protected int getAddressCount() {
-        return socketAddresses.size();
+        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() {
-        int position = currentPosition.updateAndGet(i -> (i + 1) % socketAddresses.size());
-        return socketAddresses.get(position);
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            int position = currentPosition.updateAndGet(i -> (i + 1) % socketAddresses.size());
+            return socketAddresses.get(position);
+        } finally {
+            readLock.unlock();
+        }
     }
 
-    public void setAddresses(Collection<String> addresses) {
+    public void refreshAddresses(Collection<String> addresses) {
         if (addresses == null || addresses.isEmpty()) {
             throw new IllegalArgumentException("Addresses are null or empty.");
         }

Diff for: src/main/java/org/tarantool/SingleSocketChannelProviderImpl.java

@@ -2,6 +2,7 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.nio.channels.SocketChannel;
 
 /**
@@ -11,24 +12,33 @@
 public class SingleSocketChannelProviderImpl extends BaseSocketChannelProvider {
 
     private InetSocketAddress address;
-    private int timeout;
 
     /**
-     *
+     *  Creates a simple provider
      *
      * @param address instance address
-     * @param timeout time in millis
      */
-    public SingleSocketChannelProviderImpl(String address, int timeout) {
+    public SingleSocketChannelProviderImpl(String address) {
+        setAddress(address);
+    }
+
+    public SocketAddress getAddress() {
+        return address;
+    }
+
+    public void setAddress(String address) {
+        if (StringUtils.isBlank(address)) {
+            throw new IllegalArgumentException("address must not be empty");
+        }
+
         this.address = parseAddress(address);
-        this.timeout = timeout;
     }
 
     @Override
     protected SocketChannel doRetry(int retryNumber, Throwable lastError) {
         long deadline = System.currentTimeMillis();
         try {
-            return openChannel(address, timeout);
+            return openChannel(address);
         } catch (IOException e) {
             if (deadline <= System.currentTimeMillis()) {
                 throw new CommunicationException("Connection time out.", e);

Diff for: src/main/java/org/tarantool/SocketChannelProvider.java

@@ -6,6 +6,7 @@
 public interface SocketChannelProvider {
 
     int RETRY_NO_LIMIT = -1;
+    int NO_TIMEOUT = 0;
 
     /**
      * Provides socket channel to init restore connection.

Diff for: src/main/java/org/tarantool/StringUtils.java

@@ -0,0 +1,21 @@
+package org.tarantool;
+
+public class StringUtils {
+
+    public static boolean isEmpty(String string) {
+        return (string == null) || (string.isEmpty());
+    }
+
+    public static boolean isNotEmpty(String string) {
+        return !isEmpty(string);
+    }
+
+    public static boolean isBlank(String string) {
+        return (string == null) || (string.trim().isEmpty());
+    }
+
+    public static boolean isNotBlank(String string) {
+        return !isBlank(string);
+    }
+
+}

Diff for: src/main/java/org/tarantool/TarantoolBase.java

@@ -10,6 +10,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 public abstract class TarantoolBase<Result> extends AbstractTarantoolOps<Integer, List<?>, Object, Result> {
+
     protected String serverVersion;
     /**
      * Connection state