add a rr-socket provider ability to refresh its instances list.

add a reconnection process if an obtained list doesn't contain the node we currently connected to

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,11 +10,8 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 public abstract class TarantoolBase<Result> extends AbstractTarantoolOps<Integer, List<?>, Object, Result> {
-    protected String serverVersion;
-    /**
-     * Connection state
-     */
-    protected String salt;
+
+    protected TarantoolInstanceConnectionMeta instanceConnectionMeta;
     protected MsgPackLite msgPackLite = MsgPackLite.INSTANCE;
     protected AtomicLong syncId = new AtomicLong();
     protected int initialRequestSize = 4096;
@@ -26,8 +23,7 @@ public TarantoolBase(String username, String password, Socket socket) {
         super();
         try {
             TarantoolInstanceConnectionMeta connectMeta = BinaryProtoUtils.connect(socket, username, password);
-            this.serverVersion = connectMeta.getServerVersion();
-            this.salt = connectMeta.getSalt();
+            this.instanceConnectionMeta = connectMeta;
         } catch (IOException e) {
             throw new CommunicationException("Couldn't connect to tarantool", e);
         }
@@ -81,7 +77,7 @@ public void setInitialRequestSize(int initialRequestSize) {
     }
 
     public String getServerVersion() {
-        return serverVersion;
+        return instanceConnectionMeta.getServerVersion();
     }
 
 }