Add CentralDogma backed EndpointGroup

client/java-armeria/src/main/java/com/linecorp/centraldogma/client/armeria/CentralDogmaEndpointGroup.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2018 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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.linecorp.centraldogma.client.armeria;
+
+import static java.util.Objects.requireNonNull;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.linecorp.armeria.client.Endpoint;
+import com.linecorp.armeria.client.endpoint.DynamicEndpointGroup;
+import com.linecorp.armeria.client.endpoint.EndpointGroup;
+import com.linecorp.centraldogma.client.CentralDogma;
+import com.linecorp.centraldogma.client.Watcher;
+import com.linecorp.centraldogma.common.Query;
+
+/**
+ * A {@link DynamicEndpointGroup} implementation that retrieves the {@link Endpoint} list from an entry in
+ * Central Dogma. The entry can be a JSON file or a plain text file.
+ *
+ * <p>For example, the following JSON array will be served as a list of {@link Endpoint}s:
+ * <pre>{@code
+ *  [
+ *      "host1:port1",
+ *      "host2:port2",
+ *      "host3:port3"
+ *  ]
+ * }</pre>
+ *
+ * <p>The JSON array file could be retrieved as an {@link EndpointGroup} using the following code:
+ * <pre>{@code
+ * CentralDogmaEndpointGroup<JsonNode> endpointGroup = CentralDogmaEndpointGroup.of(
+ *      centralDogma, "myProject", "myRepo",
+ *      Query.ofJson("/endpoints.json"),
+ *      EndpointListDecoder.JSON);
+ * endpointGroup.awaitInitialEndpoints();
+ * endpointGroup.endpoints();
+ * }</pre>
+ *
+ * @param <T> the type of the file in Central Dogma
+ */
+public final class CentralDogmaEndpointGroup<T> extends DynamicEndpointGroup {
+    private static final Logger logger = LoggerFactory.getLogger(CentralDogmaEndpointGroup.class);
+    private static final long WATCH_INITIALIZATION_TIMEOUT_MILLIS = TimeUnit.SECONDS.toMillis(10);
+
+    private final Watcher<T> instanceListWatcher;
+    private final EndpointListDecoder<T> endpointListDecoder;
+
+    /**
+     * Creates a new {@link CentralDogmaEndpointGroup}.
+     *
+     * @param watcher a {@link Watcher}
+     * @param endpointListDecoder an {@link EndpointListDecoder}
+     */
+    public static <T> CentralDogmaEndpointGroup<T> ofWatcher(Watcher<T> watcher,
+                                                             EndpointListDecoder<T> endpointListDecoder) {
+        return new CentralDogmaEndpointGroup<>(watcher, endpointListDecoder);
+    }
+
+    /**
+     * Creates a new {@link CentralDogmaEndpointGroup}.
+     *
+     * @param centralDogma a {@link CentralDogma}
+     * @param projectName a Central Dogma project name
+     * @param repositoryName a Central Dogma repository name
+     * @param query a {@link Query} to route file
+     * @param endpointListDecoder an {@link EndpointListDecoder}
+     */
+    public static <T> CentralDogmaEndpointGroup<T> of(CentralDogma centralDogma,
+                                                      String projectName, String repositoryName,
+                                                      Query<T> query,
+                                                      EndpointListDecoder<T> endpointListDecoder) {
+        return ofWatcher(centralDogma.fileWatcher(projectName, repositoryName, query), endpointListDecoder);
+    }
+
+    private CentralDogmaEndpointGroup(Watcher<T> instanceListWatcher,
+                                      EndpointListDecoder<T> endpointListDecoder) {
+        this.instanceListWatcher = requireNonNull(instanceListWatcher, "instanceListWatcher");
+        this.endpointListDecoder = requireNonNull(endpointListDecoder, "endpointListDecoder");
+        registerWatcher();
+    }
+
+    private void registerWatcher() {
+        instanceListWatcher.watch((revision, instances) -> {
+            try {
+                List<Endpoint> newEndpoints = endpointListDecoder.decode(instances);
+                if (newEndpoints.isEmpty()) {
+                    logger.info("Not refreshing the endpoint list of {} because it's empty. {}",
+                                instanceListWatcher, revision);
+                    return;
+                }
+                setEndpoints(newEndpoints);
+            } catch (Exception e) {
+                logger.warn("Failed to refresh the endpoint list.", e);
+            }
+        });
+        try {
+            instanceListWatcher.awaitInitialValue(WATCH_INITIALIZATION_TIMEOUT_MILLIS,
+                                                  TimeUnit.MILLISECONDS);
+        } catch (InterruptedException | TimeoutException e) {
+            logger.warn("Failed to initialize instance list on time.", e);
+        }
+    }
+
+    @Override
+    public void close() {
+        instanceListWatcher.close();
+    }
+}

client/java-armeria/src/main/java/com/linecorp/centraldogma/client/armeria/EndpointListCodecUtils.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2018 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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.linecorp.centraldogma.client.armeria;
+
+import static com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+
+import java.util.List;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import com.linecorp.armeria.client.Endpoint;
+
+final class EndpointListCodecUtils {
+    static final ObjectMapper objectMapper = new ObjectMapper().disable(FAIL_ON_UNKNOWN_PROPERTIES);
+
+    private EndpointListCodecUtils() {}
+
+    static List<Endpoint> convertToEndpointList(List<String> endpoints) {
+        return endpoints.stream()
+                        .map(String::trim)
+                        .map(Endpoint::parse)
+                        .collect(toImmutableList());
+    }
+}

client/java-armeria/src/main/java/com/linecorp/centraldogma/client/armeria/EndpointListDecoder.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2018 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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.linecorp.centraldogma.client.armeria;
+
+import java.util.List;
+
+import com.fasterxml.jackson.databind.JsonNode;
+
+import com.linecorp.armeria.client.Endpoint;
+
+/**
+ * Decodes the content of a file in Central Dogma into a list of {@link Endpoint}s.
+ *
+ * @param <T> the type of the file in Central Dogma
+ */
+@FunctionalInterface
+public interface EndpointListDecoder<T> {
+
+    /**
+     * Default {@link EndpointListDecoder} implementation for {@link JsonNode}.
+     * Retrieved object must be a JSON array (which has format as {@code "[\"segment1\", \"segment2\"]"})
+     * Each segment represents an endpoint whose format is
+     * {@code <host>[:<port_number>[:weight]]}, such as:
+     * <ul>
+     * <li>{@code "foo.com"} - default port number, default weight (1000)</li>
+     * <li>{@code "bar.com:8080} - port number 8080, default weight (1000)</li>
+     * <li>{@code "10.0.2.15:0:500} - default port number, weight 500</li>
+     * <li>{@code "192.168.1.2:8443:700} - port number 8443, weight 700</li>
+     * </ul>
+     * Note that the port number must be specified when you want to specify the weight.
+     */
+    EndpointListDecoder<JsonNode> JSON = new JsonEndpointListDecoder();
+
+    /**
+     * Default {@link EndpointListDecoder} implementation for {@link String}.
+     * Retrieved object must be a string which is a list of segments separated by a newline character.
+     * Each segment represents an endpoint whose format is
+     * {@code <host>[:<port_number>[:weight]]}, such as:
+     * <ul>
+     * <li>{@code "foo.com"} - default port number, default weight (1000)</li>
+     * <li>{@code "bar.com:8080} - port number 8080, default weight (1000)</li>
+     * <li>{@code "10.0.2.15:0:500} - default port number, weight 500</li>
+     * <li>{@code "192.168.1.2:8443:700} - port number 8443, weight 700</li>
+     * </ul>
+     * Note that the port number must be specified when you want to specify the weight.
+     */
+    EndpointListDecoder<String> TEXT = new TextEndpointListDecoder();
+
+    /**
+     * Decodes an object into a set of {@link Endpoint}s.
+     *
+     * @param object an object retrieved from Central Dogma.
+     * @return the list of {@link Endpoint}s
+     */
+    List<Endpoint> decode(T object);
+}

client/java-armeria/src/main/java/com/linecorp/centraldogma/client/armeria/JsonEndpointListDecoder.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2018 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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.linecorp.centraldogma.client.armeria;
+
+import static com.linecorp.centraldogma.client.armeria.EndpointListCodecUtils.convertToEndpointList;
+import static com.linecorp.centraldogma.client.armeria.EndpointListCodecUtils.objectMapper;
+
+import java.io.IOException;
+import java.util.List;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+
+import com.linecorp.armeria.client.Endpoint;
+
+final class JsonEndpointListDecoder implements EndpointListDecoder<JsonNode> {
+    @Override
+    public List<Endpoint> decode(JsonNode node) {
+        final List<String> endpoints;
+        try {
+            endpoints = objectMapper.readValue(node.traverse(),
+                                               new TypeReference<List<String>>() {});
+        } catch (IOException e) {
+            throw new IllegalArgumentException("invalid format: " + node);
+        }
+        return convertToEndpointList(endpoints);
+    }
+}

client/java-armeria/src/main/java/com/linecorp/centraldogma/client/armeria/TextEndpointListDecoder.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2018 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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.linecorp.centraldogma.client.armeria;
+
+import static com.linecorp.centraldogma.client.armeria.EndpointListCodecUtils.convertToEndpointList;
+
+import java.util.List;
+
+import com.google.common.base.CharMatcher;
+import com.google.common.base.Splitter;
+
+import com.linecorp.armeria.client.Endpoint;
+
+final class TextEndpointListDecoder implements EndpointListDecoder<String> {
+    private static final Splitter NEWLINE_SPLITTER = Splitter.on(CharMatcher.anyOf("\n\r"))
+                                                             .omitEmptyStrings()
+                                                             .trimResults();
+
+    @Override
+    public List<Endpoint> decode(String object) {
+        return convertToEndpointList(NEWLINE_SPLITTER.splitToList(object));
+    }
+}

client/java-armeria/src/test/java/com/linecorp/centraldogma/client/armeria/JsonEndpointListDecoderTest.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2018 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you 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:
+ *
+ *   https://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.linecorp.centraldogma.client.armeria;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+
+import org.junit.Test;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+
+import com.linecorp.armeria.client.Endpoint;
+
+public class JsonEndpointListDecoderTest {
+    private static final ObjectMapper objectMapper = new ObjectMapper();
+
+    static List<String> HOST_AND_PORT_LIST = ImmutableList.of(
+            "centraldogma-sample001.com",
+            "centraldogma-sample001.com:1234",
+            "1.2.3.4",
+            "1.2.3.4:5678"
+    );
+
+    static List<Endpoint> ENDPOINT_LIST = ImmutableList.of(
+            Endpoint.of("centraldogma-sample001.com"),
+            Endpoint.of("centraldogma-sample001.com", 1234),
+            Endpoint.of("1.2.3.4"),
+            Endpoint.of("1.2.3.4", 5678)
+    );
+
+    @Test
+    public void decode() throws Exception {
+        EndpointListDecoder<JsonNode> decoder = EndpointListDecoder.JSON;
+        List<Endpoint> decoded = decoder.decode(
+                objectMapper.readTree(objectMapper.writeValueAsString(HOST_AND_PORT_LIST)));
+
+        assertThat(decoded).hasSize(4);
+        assertThat(decoded).isEqualTo(ENDPOINT_LIST);
+    }
+}