add cluster discovery service

Author: ivanovna

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

@@ -0,0 +1,89 @@
+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;
+
+    @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, int timeout) 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;
+        }
+    }
+
+    /**
+     * 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/RoundRobinSocketProviderImpl.java

@@ -3,62 +3,94 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 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;
 
 /**
  * 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. */
+public class RoundRobinSocketProviderImpl extends BaseSocketChannelProvider {
+
+    /**
+     * 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;
+
+    /**
+     * Server addresses as configured.
+     */
+    private final List<String> addresses = new ArrayList<>();
+
+    /**
+     * Socket addresses.
+     */
+    private final List<InetSocketAddress> socketAddresses = new ArrayList<>();
+
+    /**
+     * Current position within {@link #socketAddresses} array.
+     */
+    private AtomicInteger currentPosition = new AtomicInteger(-1);
+
+    /**
+     * Lock
+     */
+//    private ReadWriteLock addressListLock = new ReentrantReadWriteLock();
 
     /**
      * Constructs an instance.
      *
-     * @param addrs Array of addresses in a form of [host]:[port].
+     * @param addresses 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);
+    public RoundRobinSocketProviderImpl(String... addresses) {
+        if (addresses == null || addresses.length == 0) {
+            throw new IllegalArgumentException("Addresses are null or empty.");
+        }
 
-        sockAddrs = new InetSocketAddress[this.addrs.length];
+        updateAddressList(Arrays.asList(addresses));
+    }
 
-        for (int i = 0; i < this.addrs.length; i++) {
-            sockAddrs[i] = parseAddress(this.addrs[i]);
+    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);
         }
     }
 
     /**
      * @return Configured addresses in a form of [host]:[port].
      */
-    public String[] getAddresses() {
-        return this.addrs;
+    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.
      */
     public RoundRobinSocketProviderImpl setTimeout(int timeout) {
-        if (timeout < 0)
+        if (timeout < 0) {
             throw new IllegalArgumentException("timeout is negative.");
+        }
 
         this.timeout = timeout;
 
@@ -67,58 +99,20 @@ public RoundRobinSocketProviderImpl setTimeout(int timeout) {
 
     /**
      * @return Maximum amount of time to wait for a socket connection establishment
-     *         with an individual server.
+     * 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);
-        }
+    protected SocketChannel doRetry(int retryNumber, Throwable 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;
+                return openChannel(getNextSocketAddress(), timeout);
             } 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 +135,21 @@ public SocketChannel get(int retryNumber, Throwable lastError) {
      * @return Number of configured addresses.
      */
     protected int getAddressCount() {
-        return sockAddrs.length;
+        return socketAddresses.size();
     }
 
     /**
      * @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);
+        int position = currentPosition.updateAndGet(i -> (i + 1) % socketAddresses.size());
+        return socketAddresses.get(position);
     }
 
-    /**
-     * 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 setAddresses(Collection<String> addresses) {
+        if (addresses == null || addresses.isEmpty()) {
+            throw new IllegalArgumentException("Addresses are null or empty.");
+        }
+        updateAddressList(addresses);
     }
 }

src/main/java/org/tarantool/SingleSocketChannelProviderImpl.java

@@ -0,0 +1,40 @@
+package org.tarantool;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.channels.SocketChannel;
+
+/**
+ * Simple provider that produces a single connection
+ * To be used with {@link TarantoolClientImpl}.
+ */
+public class SingleSocketChannelProviderImpl extends BaseSocketChannelProvider {
+
+    private InetSocketAddress address;
+    private int timeout;
+
+    /**
+     *
+     *
+     * @param address instance address
+     * @param timeout time in millis
+     */
+    public SingleSocketChannelProviderImpl(String address, int timeout) {
+        this.address = parseAddress(address);
+        this.timeout = timeout;
+    }
+
+    @Override
+    protected SocketChannel doRetry(int retryNumber, Throwable lastError) {
+        long deadline = System.currentTimeMillis();
+        try {
+            return openChannel(address, timeout);
+        } catch (IOException e) {
+            if (deadline <= System.currentTimeMillis()) {
+                throw new CommunicationException("Connection time out.", e);
+            }
+            throw new CommunicationException("Connection error.", e);
+        }
+    }
+
+}

src/main/java/org/tarantool/SocketChannelProvider.java

@@ -4,6 +4,9 @@
 import java.nio.channels.SocketChannel;
 
 public interface SocketChannelProvider {
+
+    int RETRY_NO_LIMIT = -1;
+
     /**
      * Provides socket channel to init restore connection.
      * You could change hosts on fail and sleep between retries in this method