Handle http(s):// prefixed endpoints properly in all cases (#35)

* Handle http(s):// prefixed endpoints properly in all cases

Also
- Stop using deprecated method in grpc NameResolver API
- Add unit tests

* Check explicitly for timeout in WatchTest

Author: njhill

src/main/java/com/ibm/etcd/client/EtcdClient.java

@@ -37,6 +37,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import javax.net.ssl.SSLException;
 import javax.net.ssl.TrustManagerFactory;
@@ -105,6 +107,10 @@ public class EtcdClient implements KvStoreClient {
     public static final long DEFAULT_TIMEOUT_MS = 10_000L; // 10sec default
     public static final int DEFAULT_SESSION_TIMEOUT_SECS = 20; // 20sec default
 
+    // (not intended to be strict hostname validation here)
+    protected static final Pattern ADDR_PATT =
+            Pattern.compile("(?:https?://|dns:///)?([a-zA-Z0-9\\-.]+)(?::(\\d+))?");
+
     private final int sessionTimeoutSecs;
 
     private final ByteString name, password;
@@ -302,7 +308,7 @@ public Builder withSessionTimeoutSecs(int timeoutSecs) {
          * @param sizeInBytes
          */
         public Builder withMaxInboundMessageSize(int sizeInBytes) {
-            this.maxInboundMessageSize = sizeInBytes; 
+            this.maxInboundMessageSize = sizeInBytes;
             return this;
         }
 
@@ -312,7 +318,7 @@ public Builder withMaxInboundMessageSize(int sizeInBytes) {
         public EtcdClient build() {
             NettyChannelBuilder ncb;
             if (endpoints.size() == 1) {
-                ncb = NettyChannelBuilder.forTarget(endpoints.get(0));
+                ncb = NettyChannelBuilder.forTarget(endpointToUriString(endpoints.get(0)));
                 if (overrideAuthority != null) {
                     ncb.overrideAuthority(overrideAuthority);
                 }
@@ -336,6 +342,19 @@ public EtcdClient build() {
         }
     }
 
+    static String endpointToUriString(String endpoint) {
+        Preconditions.checkNotNull(endpoint, "null endpoint");
+        Matcher m = ADDR_PATT.matcher(endpoint.trim());
+        if (!m.matches()) {
+            throw new IllegalArgumentException("invalid endpoint: " + endpoint);
+        }
+        String portStr = m.group(2);
+        if (portStr == null) {
+            portStr = String.valueOf(DEFAULT_PORT);
+        }
+        return "dns:///" + m.group(1) + ":" + portStr;
+    }
+
     private static int defaultThreadCount() {
         return Math.min(6, Runtime.getRuntime().availableProcessors());
     }

src/main/java/com/ibm/etcd/client/StaticEtcdNameResolverFactory.java

@@ -22,8 +22,6 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
 import java.util.stream.Stream;
 
 import io.grpc.Attributes;
@@ -42,16 +40,12 @@ class StaticEtcdNameResolverFactory extends NameResolver.Factory {
 
     protected static final NameResolver.Factory DNS_PROVIDER = new DnsNameResolverProvider();
 
-    // (not intended to be strict hostname validation here)
-    protected static final Pattern ADDR_PATT =
-            Pattern.compile("(?:https?://)?([a-zA-Z0-9\\-.]+)(?::(\\d+))?");
-
     static class SubResolver {
         final NameResolver resolver;
         List<EquivalentAddressGroup> eagList = Collections.emptyList();
 
-        public SubResolver(URI uri, NameResolver.Helper helper) {
-            this.resolver = DNS_PROVIDER.newNameResolver(uri, helper);
+        public SubResolver(URI uri, NameResolver.Args args) {
+            this.resolver = DNS_PROVIDER.newNameResolver(uri, args);
         }
 
         void updateEagList(List<EquivalentAddressGroup> servers, boolean ownAuthority) {
@@ -79,34 +73,23 @@ public StaticEtcdNameResolverFactory(List<String> endpoints, String overrideAuth
             throw new IllegalArgumentException("endpoints");
         }
         this.overrideAuthority = overrideAuthority;
-        int count = endpoints.size();
-        uris = new URI[count];
-        for (int i = 0; i < count; i++) {
-            String endpoint = endpoints.get(i).trim();
-            Matcher m = ADDR_PATT.matcher(endpoint);
-            if (!m.matches()) {
-                throw new IllegalArgumentException("invalid endpoint: " + endpoint);
-            }
-            String portStr = m.group(2);
-            if (portStr == null) {
-                portStr = String.valueOf(EtcdClient.DEFAULT_PORT);
-            }
-            uris[i] = URI.create("dns:///" + m.group(1) + ":" + portStr);
-        }
-        if (count > 1) {
+        uris = endpoints.stream()
+                .map(ep -> URI.create(EtcdClient.endpointToUriString(ep)))
+                .toArray(URI[]::new);
+        if (uris.length > 1) {
             Collections.shuffle(Arrays.asList(uris));
         }
     }
 
     @Override
-    public NameResolver newNameResolver(URI targetUri, NameResolver.Helper helper) {
+    public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
         if (!ETCD.equals(targetUri.getScheme())) {
             return null;
         }
         if (uris.length == 1) {
-            return new SubResolver(uris[0], helper).resolver;
+            return new SubResolver(uris[0], args).resolver;
         }
-        SubResolver[] resolvers = createSubResolvers(helper);
+        SubResolver[] resolvers = createSubResolvers(args);
         return new NameResolver() {
             int currentCount = 0;
             @Override
@@ -157,11 +140,11 @@ public void shutdown() {
         };
     }
 
-    private SubResolver[] createSubResolvers(NameResolver.Helper helper) {
+    private SubResolver[] createSubResolvers(NameResolver.Args args) {
         int count = uris.length;
         SubResolver[] resolvers = new SubResolver[count];
         for (int i = 0; i < count; i++) {
-            resolvers[i] = new SubResolver(uris[i], helper);
+            resolvers[i] = new SubResolver(uris[i], args);
         }
         return resolvers;
     }

src/test/java/com/ibm/etcd/client/ClientBuilderTest.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2017, 2018 IBM Corp. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy
+ * of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package com.ibm.etcd.client;
+
+import static com.ibm.etcd.client.KeyUtils.bs;
+import static org.junit.Assert.assertEquals;
+
+import java.util.Collections;
+
+import org.junit.Test;
+
+import com.ibm.etcd.client.kv.KvClient;
+
+public class ClientBuilderTest {
+
+    @Test
+    public void testForEndpoint() throws Exception {
+        try (KvStoreClient client = EtcdClient.forEndpoint("localhost", 2379)
+                .withPlainText().build()) {
+            basicTest(client);
+        }
+    }
+
+    @Test
+    public void testForEndpointsSingle() throws Exception {
+        String[] endpoints = { "localhost:2379", "http://localhost:2379",
+                "https://localhost:2379", "dns:///localhost:2379" };
+
+        for (String endpoint : endpoints) {
+            try (KvStoreClient client = EtcdClient.forEndpoints(Collections.singletonList(endpoint))
+                    .withPlainText().build()) {
+                basicTest(client);
+            }
+        }
+    }
+
+    @Test
+    public void testForEndpointsMulti() throws Exception {
+        try (KvStoreClient client = EtcdClient.forEndpoints(
+                "localhost:2379,http://localhost:2379,https://localhost:2379,dns:///localhost:2379")
+                .withPlainText().build()) {
+            basicTest(client);
+        }
+    }
+
+    static void basicTest(KvStoreClient client) {
+        KvClient kvc = client.getKvClient();
+        kvc.put(bs("cbt"), bs("test")).sync();
+        assertEquals("test", kvc.delete(bs("cbt")).prevKv().sync()
+                .getPrevKvs(0).getValue().toStringUtf8());
+    }
+}

src/test/java/com/ibm/etcd/client/EtcdTestSuite.java

@@ -35,7 +35,7 @@
 import com.ibm.etcd.client.utils.RangeCacheTest;
 
 @RunWith(Suite.class)
-@SuiteClasses({KvTest.class, WatchTest.class, LeaseTest.class,
+@SuiteClasses({ClientBuilderTest.class, KvTest.class, WatchTest.class, LeaseTest.class,
     LockTest.class, PersistentLeaseKeyTest.class, RangeCacheTest.class})
 public class EtcdTestSuite {
 

src/test/java/com/ibm/etcd/client/KvTest.java

@@ -15,6 +15,7 @@
  */
 package com.ibm.etcd.client;
 
+import static com.ibm.etcd.client.KeyUtils.bs;
 import static org.junit.Assert.*;
 
 import java.util.UUID;
@@ -27,8 +28,6 @@
 
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.protobuf.ByteString;
-import com.ibm.etcd.client.EtcdClient;
-import com.ibm.etcd.client.KvStoreClient;
 import com.ibm.etcd.client.kv.KvClient;
 
 import io.grpc.Deadline;
@@ -188,10 +187,6 @@ public void testSyncDeadlock() throws Exception {
         }
     }
 
-    public static ByteString bs(String str) {
-        return ByteString.copyFromUtf8(str);
-    }
-
     static String t(long start) {
         return String.format("%.3f ", (System.currentTimeMillis() - start) / 1000.0);
     }

src/test/java/com/ibm/etcd/client/WatchTest.java

@@ -15,7 +15,7 @@
  */
 package com.ibm.etcd.client;
 
-import static com.ibm.etcd.client.KvTest.bs;
+import static com.ibm.etcd.client.KeyUtils.bs;
 import static com.ibm.etcd.client.KvTest.t;
 import static org.junit.Assert.*;
 
@@ -163,7 +163,8 @@ public void onCompleted() {
             proxy.start();
 
             // watch should be unaffected - next event seen should be the missed one
-            wu = (WatchUpdate)watchEvents.poll(5000L, TimeUnit.MILLISECONDS);
+            wu = (WatchUpdate) watchEvents.poll(6000L, TimeUnit.MILLISECONDS);
+            assertNotNull("Expected watch event not received after server reconnection", wu);
             assertEquals(bs("/watchtest/e"), wu.getEvents().get(0).getKv().getKey());
 
             watch.close();

src/test/java/com/ibm/etcd/client/utils/LeaderElectionTest.java

@@ -15,7 +15,7 @@
  */
 package com.ibm.etcd.client.utils;
 
-import static com.ibm.etcd.client.KvTest.bs;
+import static com.ibm.etcd.client.KeyUtils.bs;
 import static org.junit.Assert.*;
 
 import org.junit.Test;

src/test/java/com/ibm/etcd/client/utils/PersistentLeaseKeyTest.java

@@ -20,7 +20,7 @@
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
-import static com.ibm.etcd.client.KvTest.bs;
+import static com.ibm.etcd.client.KeyUtils.bs;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
 import org.junit.AfterClass;

src/test/java/com/ibm/etcd/client/utils/RangeCacheTest.java

@@ -18,7 +18,7 @@
 import static com.ibm.etcd.client.KeyUtils.fromHexString;
 import static com.ibm.etcd.client.KeyUtils.plusOne;
 import static com.ibm.etcd.client.KeyUtils.toHexString;
-import static com.ibm.etcd.client.KvTest.bs;
+import static com.ibm.etcd.client.KeyUtils.bs;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;