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

@@ -82,8 +82,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,31 @@
 
 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 {
+public class RoundRobinSocketProviderImpl extends BaseSocketChannelProvider implements RefreshableSocketProvider {
 
-    /**
-     * Timeout to establish socket connection with an individual server.
-     */
-    private int timeout; // 0 is infinite.
+    private static final int NO_TIMEOUT = 0;
 
     /**
-     * Server addresses as configured.
+     * Timeout to establish socket connection with an individual server.
      */
-    private final List<String> addresses = new ArrayList<>();
+    private int timeout = NO_TIMEOUT;
 
     /**
      * Socket addresses.
@@ -39,42 +41,63 @@ 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 List<SocketAddress> getAddresses() {
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            return Collections.unmodifiableList(this.socketAddresses);
+        } finally {
+            readLock.unlock();
+        }
     }
 
-    public String getLastObtainedAddress() {
-        int index = currentPosition.get();
-        return index >= 0 ? addresses.get(index) : null;
+    private InetSocketAddress getLastObtainedAddress() {
+        Lock readLock = addressListLock.readLock();
+        readLock.lock();
+        try {
+            int index = currentPosition.get();
+            return index >= 0 ? socketAddresses.get(index) : null;
+        } finally {
+            readLock.unlock();
+        }
     }
 
     /**
@@ -91,9 +114,7 @@ public RoundRobinSocketProviderImpl setTimeout(int timeout) {
         if (timeout < 0) {
             throw new IllegalArgumentException("timeout is negative.");
         }
-
         this.timeout = timeout;
-
         return this;
     }
 
@@ -108,6 +129,7 @@ public int getTimeout() {
     @Override
     protected SocketChannel doRetry(int retryNumber, Throwable lastError) {
         int attempts = getAddressCount();
+        // todo: recalc deadline?
         long deadline = System.currentTimeMillis() + timeout * attempts;
         while (!Thread.currentThread().isInterrupted()) {
             try {
@@ -135,18 +157,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/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