client-api: Add metadata as field on ServiceDiscovererEvent

servicetalk-client-api/src/main/java/io/servicetalk/client/api/ServiceDiscovererEvent.java

@@ -16,6 +16,7 @@
 package io.servicetalk.client.api;
 
 import java.util.Locale;
+import java.util.Map;
 
 /**
  * Notification from the Service Discovery system that availability for an address has changed.
@@ -50,6 +51,7 @@
  * @param <ResolvedAddress> the type of address after resolution.
  */
 public interface ServiceDiscovererEvent<ResolvedAddress> {
+
     /**
      * Get the resolved address which is the subject of this event.
      * @return a resolved address that can be used for connecting.
@@ -63,6 +65,15 @@ public interface ServiceDiscovererEvent<ResolvedAddress> {
      */
     Status status();
 
+    /**
+     * The raw meta-data associated with this ServiceDiscovererEvent.
+     * Note: the result will be an unmodifiable collection.
+     * @return the raw meta-data associated with this ServiceDiscovererEvent.
+     */
+    default Map<String, Object> metadata() {
+        return ServiceDiscovererMetadata.EMPTY_MAP;
+    }
+
     /**
      * Status provided by the {@link ServiceDiscoverer} system that guides the actions of {@link LoadBalancer} upon the
      * bound {@link ServiceDiscovererEvent#address()} (via {@link ServiceDiscovererEvent}).

servicetalk-client-api/src/main/java/io/servicetalk/client/api/ServiceDiscovererMetadata.java

@@ -0,0 +1,124 @@
+/*
+ * Copyright © 2024 Apple Inc. and the ServiceTalk project authors
+ *
+ * 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 io.servicetalk.client.api;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Utilities helpful for extracting meta-data from {@link ServiceDiscovererEvent}s.
+ */
+public final class ServiceDiscovererMetadata {
+
+    public static final Map<String, Object> EMPTY_MAP = Collections.unmodifiableMap(new HashMap<>(0));
+
+    /**
+     * Metadata that describes the relative weight of an endpoint.
+     */
+    public static final Key<Double> WEIGHT = new ServiceDiscovererMetadata.Key<>(Double.class, "endpoint.weight", 1d);
+
+    /**
+     * Metadata describing the priority class of an endpoint.
+     */
+    public static final Key<Integer> PRIORITY = new ServiceDiscovererMetadata.Key<>(
+            Integer.class, "endpoint.priority", 0);
+
+    /**
+     * An extractor of meta-data to user with {@link ServiceDiscovererEvent} instances.
+     *
+     * A {@link ServiceDiscovererEvent} can carry additional metadata, but this data is not type safe. The key type
+     * exists to provide a uniform way to define meta-data extractors that can properly extract and cast meta-data
+     * while also providing a default.
+     * @param <T> the expected type of the meta-data.
+     */
+    public static final class Key<T> {
+
+        private final Class<T> clazz;
+        private final String name;
+        private final T defaultValue;
+
+        public Key(final Class<T> clazz, final String name, final T defaultValue) {
+            this.clazz = requireNonNull(clazz, "clazz");
+            this.name = requireNonNull(name, "name");
+            this.defaultValue = requireNonNull(defaultValue, "defaultValue");
+        }
+
+        /**
+         * Get the name associated with the meta-data.
+         * @return the name associated with the meta-data.
+         */
+        public String name() {
+            return name;
+        }
+
+        /**
+         * The java class that is expected to be associated with the meta-data.
+         * @return the java class that is expected to be associated with the meta-data.
+         */
+        public Class<T> clazz() {
+            return clazz;
+        }
+
+        /**
+         * Determine whether the meta-data both contains an entry with the keys name and that entry is the correct type.
+         * @param event the {@link ServiceDiscovererEvent} for which to check if the meta-data exists.
+         * @return true if the meta-data contains an entry with the keys name and the value is the correct type.
+         */
+        public boolean contains(ServiceDiscovererEvent<?> event) {
+            return clazz.isInstance(event.metadata().get(name));
+        }
+
+        /**
+         * Extract the meta-data from a {@link ServiceDiscovererEvent}, or get the default.
+         * @param event the {@link ServiceDiscovererEvent} from which to extract the meta-data.
+         * @return the value contained in the meta-data, or the default if it doesn't exist or has the wrong type.
+         */
+        public T getValue(ServiceDiscovererEvent<?> event) {
+            Object result = event.metadata().get(name);
+            if (clazz.isInstance(result)) {
+                return (T) result;
+            } else {
+                return defaultValue;
+            }
+        }
+
+        @Override
+        public int hashCode() {
+            return name.hashCode();
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null || obj.getClass() != Key.class) {
+                return false;
+            }
+            Key<?> other = (Key<?>) obj;
+            return other.clazz.equals(clazz) && other.name.equals(name);
+        }
+
+        @Override
+        public String toString() {
+            return "Key<" + clazz.getName() + ">(" + name + ", " + defaultValue + ")";
+        }
+    }
+
+    private ServiceDiscovererMetadata() {
+        // no instances.
+    }
+}

servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultHost.java

@@ -79,6 +79,7 @@ private enum State {
 
     private final String lbDescription;
     private final Addr address;
+    private final double weight;
     @Nullable
     private final HealthCheckConfig healthCheckConfig;
     @Nullable
@@ -91,12 +92,14 @@ private enum State {
     private volatile ConnState connState = new ConnState(emptyList(), State.ACTIVE, 0, null);
 
     DefaultHost(final String lbDescription, final Addr address,
+                final double weight,
                 final ConnectionPoolStrategy<C> connectionPoolStrategy,
                 final ConnectionFactory<Addr, ? extends C> connectionFactory,
                 final HostObserver hostObserver, final @Nullable HealthCheckConfig healthCheckConfig,
                 final @Nullable HealthIndicator healthIndicator) {
         this.lbDescription = requireNonNull(lbDescription, "lbDescription");
         this.address = requireNonNull(address, "address");
+        this.weight = weight;
         this.healthIndicator = healthIndicator;
         this.connectionPoolStrategy = requireNonNull(connectionPoolStrategy, "connectionPoolStrategy");
         requireNonNull(connectionFactory, "connectionFactory");
@@ -113,6 +116,11 @@ public Addr address() {
         return address;
     }
 
+    @Override
+    public double weight() {
+        return weight;
+    }
+
     @Override
     public boolean markActiveIfNotClosed() {
         final ConnState oldState = connStateUpdater.getAndUpdate(this, oldConnState -> {

servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultLoadBalancer.java

@@ -55,6 +55,7 @@
 import static io.servicetalk.client.api.ServiceDiscovererEvent.Status.AVAILABLE;
 import static io.servicetalk.client.api.ServiceDiscovererEvent.Status.EXPIRED;
 import static io.servicetalk.client.api.ServiceDiscovererEvent.Status.UNAVAILABLE;
+import static io.servicetalk.client.api.ServiceDiscovererMetadata.WEIGHT;
 import static io.servicetalk.concurrent.api.AsyncCloseables.newCompositeCloseable;
 import static io.servicetalk.concurrent.api.AsyncCloseables.toAsyncCloseable;
 import static io.servicetalk.concurrent.api.Processors.newPublisherProcessorDropHeadOnOverflow;
@@ -309,7 +310,7 @@ private void sequentialOnNext(Collection<? extends ServiceDiscovererEvent<Resolv
                     } else {
                         // It's a new host, so the set changed.
                         hostSetChanged = true;
-                        nextHosts.add(createHost(event.address()));
+                        nextHosts.add(createHost(event.address(), WEIGHT.getValue(event)));
                     }
                 } else if (EXPIRED.equals(event.status())) {
                     if (!host.markExpired()) {
@@ -336,7 +337,7 @@ private void sequentialOnNext(Collection<? extends ServiceDiscovererEvent<Resolv
                 if (AVAILABLE.equals(event.status())) {
                     sendReadyEvent = true;
                     hostSetChanged = true;
-                    nextHosts.add(createHost(event.address()));
+                    nextHosts.add(createHost(event.address(), WEIGHT.getValue(event)));
                 }
             }
 
@@ -380,15 +381,15 @@ private void sequentialOnNext(Collection<? extends ServiceDiscovererEvent<Resolv
             }
         }
 
-        private Host<ResolvedAddress, C> createHost(ResolvedAddress addr) {
+        private Host<ResolvedAddress, C> createHost(ResolvedAddress addr, double weight) {
             final LoadBalancerObserver.HostObserver hostObserver = loadBalancerObserver.hostObserver(addr);
             // All hosts will share the health check config of the parent load balancer.
             final HealthIndicator indicator = outlierDetector.newHealthIndicator(addr, hostObserver);
             // We don't need the host level health check if we are either not health checking at all or if the
             // failed connect threshold is negative, meaning disabled.
             final HealthCheckConfig hostHealthCheckConfig =
                     healthCheckConfig == null || healthCheckConfig.failedThreshold < 0 ? null : healthCheckConfig;
-            final Host<ResolvedAddress, C> host = new DefaultHost<>(lbDescription, addr, connectionPoolStrategy,
+            final Host<ResolvedAddress, C> host = new DefaultHost<>(lbDescription, addr, weight, connectionPoolStrategy,
                     connectionFactory, hostObserver, hostHealthCheckConfig, indicator);
             if (indicator != null) {
                 indicator.setHost(host);

servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/Host.java

@@ -51,6 +51,12 @@ interface Host<ResolvedAddress, C extends LoadBalancedConnection> extends Listen
      */
     ResolvedAddress address();
 
+    /**
+     * The relative weight of the endpoint.
+     * @return the relative weight of the endpoint.
+     */
+    double weight();
+
     /**
      * Determine the health status of this host.
      * @return whether the host considers itself healthy enough to serve traffic. This is best effort and does not
@@ -78,12 +84,4 @@ interface Host<ResolvedAddress, C extends LoadBalancedConnection> extends Listen
      * @return true if the host is now in the closed state, false otherwise.
      */
     boolean markExpired();
-
-    /**
-     * The weight of the host, relative to the weights of associated hosts.
-     * @return the relative weight of the host.
-     */
-    default double weight() {
-        return 1.0;
-    }
 }

servicetalk-loadbalancer-experimental/src/test/java/io/servicetalk/loadbalancer/DefaultHostTest.java

@@ -48,6 +48,7 @@
 class DefaultHostTest {
 
     private static final String DEFAULT_ADDRESS = "address";
+    private static final double DEFAULT_WEIGHT = 1.0;
 
     @RegisterExtension
     final ExecutorExtension<TestExecutor> executor = ExecutorExtension.withTestExecutor();
@@ -81,7 +82,7 @@ void cleanup() {
     }
 
     private void buildHost(@Nullable HealthIndicator healthIndicator) {
-        host = new DefaultHost<>("lbDescription", DEFAULT_ADDRESS,
+        host = new DefaultHost<>("lbDescription", DEFAULT_ADDRESS, DEFAULT_WEIGHT,
                 LinearSearchConnectionPoolStrategy.<TestLoadBalancedConnection>factory(DEFAULT_LINEAR_SEARCH_SPACE)
                         .buildStrategy("resource"),
                 connectionFactory, mockHostObserver, healthCheckConfig, healthIndicator);