Migrate XdsEndpointGroup to client.endpoint (xDS-endpoint pt 2)

xds/src/main/java/com/linecorp/armeria/xds/ClusterSnapshot.java

@@ -17,6 +17,7 @@
 package com.linecorp.armeria.xds;
 
 import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
 
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.annotation.UnstableApi;
@@ -89,6 +90,26 @@ int index() {
         return index;
     }
 
+    @Override
+    public boolean equals(Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+        final ClusterSnapshot that = (ClusterSnapshot) object;
+        return index == that.index && Objects.equal(clusterXdsResource, that.clusterXdsResource) &&
+               Objects.equal(endpointSnapshot, that.endpointSnapshot) &&
+               Objects.equal(virtualHost, that.virtualHost) &&
+               Objects.equal(route, that.route);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(clusterXdsResource, endpointSnapshot, virtualHost, route, index);
+    }
+
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)

xds/src/main/java/com/linecorp/armeria/xds/ConfigSourceClient.java

@@ -30,6 +30,7 @@
 import com.linecorp.armeria.client.retry.Backoff;
 import com.linecorp.armeria.common.SessionProtocol;
 import com.linecorp.armeria.common.util.SafeCloseable;
+import com.linecorp.armeria.xds.client.endpoint.XdsEndpointGroup;
 
 import io.envoyproxy.envoy.config.core.v3.ApiConfigSource;
 import io.envoyproxy.envoy.config.core.v3.ApiConfigSource.ApiType;
@@ -65,7 +66,7 @@ final class ConfigSourceClient implements SafeCloseable {
         final ClusterSnapshot clusterSnapshot = bootstrapClusters.clusterSnapshot(clusterName);
         checkArgument(clusterSnapshot != null, "Unable to find static cluster '%s'", clusterName);
 
-        endpointGroup = new XdsEndpointGroup(clusterSnapshot);
+        endpointGroup = XdsEndpointGroup.of(clusterSnapshot);
         final boolean ads = apiConfigSource.getApiType() == ApiType.AGGREGATED_GRPC;
         final UpstreamTlsContext tlsContext = clusterSnapshot.xdsResource().upstreamTlsContext();
         final SessionProtocol sessionProtocol =

xds/src/main/java/com/linecorp/armeria/xds/EndpointSnapshot.java

@@ -17,6 +17,7 @@
 package com.linecorp.armeria.xds;
 
 import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
 
 import com.linecorp.armeria.common.annotation.UnstableApi;
 
@@ -38,6 +39,23 @@ public EndpointXdsResource xdsResource() {
         return endpoint;
     }
 
+    @Override
+    public boolean equals(Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+        final EndpointSnapshot that = (EndpointSnapshot) object;
+        return Objects.equal(endpoint, that.endpoint);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(endpoint);
+    }
+
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)

xds/src/main/java/com/linecorp/armeria/xds/ListenerSnapshot.java

@@ -17,6 +17,7 @@
 package com.linecorp.armeria.xds;
 
 import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
 
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.annotation.UnstableApi;
@@ -51,6 +52,24 @@ public RouteSnapshot routeSnapshot() {
         return routeSnapshot;
     }
 
+    @Override
+    public boolean equals(Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+        final ListenerSnapshot that = (ListenerSnapshot) object;
+        return Objects.equal(listenerXdsResource, that.listenerXdsResource) &&
+               Objects.equal(routeSnapshot, that.routeSnapshot);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(listenerXdsResource, routeSnapshot);
+    }
+
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)

xds/src/main/java/com/linecorp/armeria/xds/RouteSnapshot.java

@@ -23,6 +23,7 @@
 import java.util.Map;
 
 import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
 
 import com.linecorp.armeria.common.annotation.UnstableApi;
 
@@ -74,6 +75,24 @@ public Map<VirtualHost, List<ClusterSnapshot>> virtualHostMap() {
         return virtualHostMap;
     }
 
+    @Override
+    public boolean equals(Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+        final RouteSnapshot that = (RouteSnapshot) object;
+        return Objects.equal(routeXdsResource, that.routeXdsResource) &&
+               Objects.equal(clusterSnapshots, that.clusterSnapshots);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(routeXdsResource, clusterSnapshots);
+    }
+
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)

xds/src/main/java/com/linecorp/armeria/xds/XdsConverterUtil.java

@@ -17,81 +17,17 @@
 package com.linecorp.armeria.xds;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.collect.ImmutableList.toImmutableList;
-import static com.linecorp.armeria.xds.XdsConstants.SUBSET_LOAD_BALANCING_FILTER_NAME;
 
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.function.Predicate;
-
-import com.google.common.base.Strings;
-import com.google.protobuf.Struct;
-import com.google.protobuf.Value;
-
-import com.linecorp.armeria.client.Endpoint;
 import com.linecorp.armeria.common.annotation.Nullable;
 
 import io.envoyproxy.envoy.config.core.v3.ApiConfigSource;
 import io.envoyproxy.envoy.config.core.v3.ApiConfigSource.ApiType;
 import io.envoyproxy.envoy.config.core.v3.ConfigSource;
-import io.envoyproxy.envoy.config.core.v3.SocketAddress;
-import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
-import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint;
 
 final class XdsConverterUtil {
 
     private XdsConverterUtil() {}
 
-    static List<Endpoint> convertEndpoints(ClusterLoadAssignment clusterLoadAssignment) {
-        return convertEndpoints(clusterLoadAssignment, lbEndpoint -> true);
-    }
-
-    static List<Endpoint> convertEndpoints(ClusterLoadAssignment clusterLoadAssignment, Struct filterMetadata) {
-        checkArgument(filterMetadata.getFieldsCount() > 0,
-                      "filterMetadata.getFieldsCount(): %s (expected: > 0)", filterMetadata.getFieldsCount());
-        final Predicate<LbEndpoint> lbEndpointPredicate = lbEndpoint -> {
-            final Struct endpointMetadata = lbEndpoint.getMetadata().getFilterMetadataOrDefault(
-                    SUBSET_LOAD_BALANCING_FILTER_NAME, Struct.getDefaultInstance());
-            if (endpointMetadata.getFieldsCount() == 0) {
-                return false;
-            }
-            return containsFilterMetadata(filterMetadata, endpointMetadata);
-        };
-        return convertEndpoints(clusterLoadAssignment, lbEndpointPredicate);
-    }
-
-    private static List<Endpoint> convertEndpoints(ClusterLoadAssignment clusterLoadAssignment,
-                                                   Predicate<LbEndpoint> lbEndpointPredicate) {
-        return clusterLoadAssignment.getEndpointsList().stream().flatMap(
-                localityLbEndpoints -> localityLbEndpoints
-                        .getLbEndpointsList()
-                        .stream()
-                        .filter(lbEndpointPredicate)
-                        .map(lbEndpoint -> {
-                            final SocketAddress socketAddress =
-                                    lbEndpoint.getEndpoint().getAddress().getSocketAddress();
-                            final String hostname = lbEndpoint.getEndpoint().getHostname();
-                            if (!Strings.isNullOrEmpty(hostname)) {
-                                return Endpoint.of(hostname, socketAddress.getPortValue())
-                                               .withIpAddr(socketAddress.getAddress());
-                            } else {
-                                return Endpoint.of(socketAddress.getAddress(), socketAddress.getPortValue());
-                            }
-                        })).collect(toImmutableList());
-    }
-
-    private static boolean containsFilterMetadata(Struct filterMetadata, Struct endpointMetadata) {
-        final Map<String, Value> endpointMetadataMap = endpointMetadata.getFieldsMap();
-        for (Entry<String, Value> entry : filterMetadata.getFieldsMap().entrySet()) {
-            final Value value = endpointMetadataMap.get(entry.getKey());
-            if (value == null || !value.equals(entry.getValue())) {
-                return false;
-            }
-        }
-        return true;
-    }
-
     static void validateConfigSource(@Nullable ConfigSource configSource) {
         if (configSource == null || configSource.equals(ConfigSource.getDefaultInstance())) {
             return;

xds/src/main/java/com/linecorp/armeria/xds/XdsConstants.java → xds/src/main/java/com/linecorp/armeria/xds/client/endpoint/XdsConstants.java

@@ -14,7 +14,7 @@
  * under the License.
  */
 
-package com.linecorp.armeria.xds;
+package com.linecorp.armeria.xds.client.endpoint;
 
 final class XdsConstants {
 

xds/src/main/java/com/linecorp/armeria/xds/XdsEndpointGroup.java → xds/src/main/java/com/linecorp/armeria/xds/client/endpoint/XdsEndpointGroup.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2023 LINE Corporation
+ * Copyright 2024 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
@@ -14,11 +14,11 @@
  * under the License.
  */
 
-package com.linecorp.armeria.xds;
+package com.linecorp.armeria.xds.client.endpoint;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static com.linecorp.armeria.xds.XdsConstants.SUBSET_LOAD_BALANCING_FILTER_NAME;
-import static com.linecorp.armeria.xds.XdsConverterUtil.convertEndpoints;
+import static com.linecorp.armeria.xds.client.endpoint.XdsConstants.SUBSET_LOAD_BALANCING_FILTER_NAME;
+import static com.linecorp.armeria.xds.client.endpoint.XdsEndpointUtil.convertEndpoints;
 import static java.util.Objects.requireNonNull;
 
 import java.util.List;
@@ -37,11 +37,19 @@
 import com.linecorp.armeria.client.endpoint.EndpointGroup;
 import com.linecorp.armeria.common.annotation.UnstableApi;
 import com.linecorp.armeria.common.util.SafeCloseable;
+import com.linecorp.armeria.xds.ClusterSnapshot;
+import com.linecorp.armeria.xds.EndpointSnapshot;
+import com.linecorp.armeria.xds.ListenerRoot;
+import com.linecorp.armeria.xds.ListenerSnapshot;
+import com.linecorp.armeria.xds.RouteSnapshot;
+import com.linecorp.armeria.xds.SnapshotWatcher;
+import com.linecorp.armeria.xds.XdsBootstrap;
 
 import io.envoyproxy.envoy.config.cluster.v3.Cluster;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.LbSubsetConfig;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.LbSubsetConfig.LbSubsetFallbackPolicy;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster.LbSubsetConfig.LbSubsetSelector;
+import io.envoyproxy.envoy.config.core.v3.GrpcService;
 import io.envoyproxy.envoy.config.core.v3.SocketAddress;
 import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
 import io.envoyproxy.envoy.config.route.v3.Route;
@@ -76,6 +84,17 @@ public static EndpointGroup of(ListenerRoot listenerRoot) {
         return new XdsEndpointGroup(listenerRoot);
     }
 
+    /**
+     * Creates a {@link XdsEndpointGroup} based on the specified {@link ClusterSnapshot}.
+     * This may be useful if one would like to create an {@link EndpointGroup} based on
+     * a {@link GrpcService}.
+     */
+    @UnstableApi
+    public static EndpointGroup of(ClusterSnapshot clusterSnapshot) {
+        requireNonNull(clusterSnapshot, "clusterSnapshot");
+        return new XdsEndpointGroup(clusterSnapshot);
+    }
+
     XdsEndpointGroup(ListenerRoot listenerRoot) {
         final SnapshotWatcher<ListenerSnapshot> watcher = update -> {
             final RouteSnapshot routeSnapshot = update.routeSnapshot();

xds/src/main/java/com/linecorp/armeria/xds/client/endpoint/XdsEndpointUtil.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2024 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.armeria.xds.client.endpoint;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.linecorp.armeria.xds.client.endpoint.XdsConstants.SUBSET_LOAD_BALANCING_FILTER_NAME;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.function.Predicate;
+
+import com.google.common.base.Strings;
+import com.google.protobuf.Struct;
+import com.google.protobuf.Value;
+
+import com.linecorp.armeria.client.Endpoint;
+
+import io.envoyproxy.envoy.config.core.v3.SocketAddress;
+import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
+import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint;
+
+final class XdsEndpointUtil {
+
+    static List<Endpoint> convertEndpoints(ClusterLoadAssignment clusterLoadAssignment) {
+        return convertEndpoints(clusterLoadAssignment, lbEndpoint -> true);
+    }
+
+    static List<Endpoint> convertEndpoints(ClusterLoadAssignment clusterLoadAssignment, Struct filterMetadata) {
+        checkArgument(filterMetadata.getFieldsCount() > 0,
+                      "filterMetadata.getFieldsCount(): %s (expected: > 0)", filterMetadata.getFieldsCount());
+        final Predicate<LbEndpoint> lbEndpointPredicate = lbEndpoint -> {
+            final Struct endpointMetadata = lbEndpoint.getMetadata().getFilterMetadataOrDefault(
+                    SUBSET_LOAD_BALANCING_FILTER_NAME, Struct.getDefaultInstance());
+            if (endpointMetadata.getFieldsCount() == 0) {
+                return false;
+            }
+            return containsFilterMetadata(filterMetadata, endpointMetadata);
+        };
+        return convertEndpoints(clusterLoadAssignment, lbEndpointPredicate);
+    }
+
+    private static List<Endpoint> convertEndpoints(ClusterLoadAssignment clusterLoadAssignment,
+                                                   Predicate<LbEndpoint> lbEndpointPredicate) {
+        return clusterLoadAssignment.getEndpointsList().stream().flatMap(
+                localityLbEndpoints -> localityLbEndpoints
+                        .getLbEndpointsList()
+                        .stream()
+                        .filter(lbEndpointPredicate)
+                        .map(lbEndpoint -> {
+                            final SocketAddress socketAddress =
+                                    lbEndpoint.getEndpoint().getAddress().getSocketAddress();
+                            final String hostname = lbEndpoint.getEndpoint().getHostname();
+                            if (!Strings.isNullOrEmpty(hostname)) {
+                                return Endpoint.of(hostname, socketAddress.getPortValue())
+                                               .withIpAddr(socketAddress.getAddress());
+                            } else {
+                                return Endpoint.of(socketAddress.getAddress(), socketAddress.getPortValue());
+                            }
+                        })).collect(toImmutableList());
+    }
+
+    private static boolean containsFilterMetadata(Struct filterMetadata, Struct endpointMetadata) {
+        final Map<String, Value> endpointMetadataMap = endpointMetadata.getFieldsMap();
+        for (Entry<String, Value> entry : filterMetadata.getFieldsMap().entrySet()) {
+            final Value value = endpointMetadataMap.get(entry.getKey());
+            if (value == null || !value.equals(entry.getValue())) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private XdsEndpointUtil() {}
+}