diff --git a/servicetalk-client-api/src/main/java/io/servicetalk/client/api/DefaultServiceDiscovererEvent.java b/servicetalk-client-api/src/main/java/io/servicetalk/client/api/DefaultServiceDiscovererEvent.java index 3f298fc271..bc8c7a5a06 100644 --- a/servicetalk-client-api/src/main/java/io/servicetalk/client/api/DefaultServiceDiscovererEvent.java +++ b/servicetalk-client-api/src/main/java/io/servicetalk/client/api/DefaultServiceDiscovererEvent.java @@ -23,6 +23,7 @@ */ public final class DefaultServiceDiscovererEvent implements ServiceDiscovererEvent { private final T address; + private final double weight; private final Status status; /** @@ -31,7 +32,21 @@ public final class DefaultServiceDiscovererEvent implements ServiceDiscoverer * @param status Value returned by {@link #status()}. */ public DefaultServiceDiscovererEvent(T address, Status status) { + this(address, 1.0, status); + } + + /** + * Create a new instance. + * @param address The address returned by {@link #address()}. + * @param weight The relative weight of the address. + * @param status Value returned by {@link #status()}. + */ + public DefaultServiceDiscovererEvent(T address, double weight, Status status) { + if (weight < 0 || Double.isNaN(weight) || !Double.isFinite(weight)) { + throw new IllegalArgumentException("Weight value most be a finite positive number: " + weight); + } this.address = requireNonNull(address); + this.weight = weight; this.status = requireNonNull(status); } @@ -40,6 +55,11 @@ public T address() { return address; } + @Override + public double weight() { + return weight; + } + @Override public Status status() { return status; diff --git a/servicetalk-client-api/src/main/java/io/servicetalk/client/api/ServiceDiscovererEvent.java b/servicetalk-client-api/src/main/java/io/servicetalk/client/api/ServiceDiscovererEvent.java index c08ecbbf6d..04c0a89b0a 100644 --- a/servicetalk-client-api/src/main/java/io/servicetalk/client/api/ServiceDiscovererEvent.java +++ b/servicetalk-client-api/src/main/java/io/servicetalk/client/api/ServiceDiscovererEvent.java @@ -39,6 +39,14 @@ public interface ServiceDiscovererEvent { */ Status status(); + /** + * The weight of endpoint. + * @return the weight of endpoint. + */ + default double weight() { + return 1.0; + } + /** * Status provided by the {@link ServiceDiscoverer} system that guides the actions of {@link LoadBalancer} upon the * bound {@link ServiceDiscovererEvent#address()} (via {@link ServiceDiscovererEvent}). diff --git a/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultHost.java b/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultHost.java index e10a81be09..73e5b29bce 100644 --- a/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultHost.java +++ b/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 @@ -90,13 +91,14 @@ private enum State { private final ListenableAsyncCloseable closeable; private volatile ConnState connState = new ConnState(emptyList(), State.ACTIVE, 0, null); - DefaultHost(final String lbDescription, final Addr address, + DefaultHost(final String lbDescription, final Addr address, final double weight, final ConnectionPoolStrategy connectionPoolStrategy, final ConnectionFactory 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 +115,11 @@ public Addr address() { return address; } + @Override + public double weight() { + return weight; + } + @Override public boolean markActiveIfNotClosed() { final ConnState oldState = connStateUpdater.getAndUpdate(this, oldConnState -> { diff --git a/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultLoadBalancer.java b/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultLoadBalancer.java index b389b9414c..0690681384 100644 --- a/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultLoadBalancer.java +++ b/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultLoadBalancer.java @@ -309,7 +309,7 @@ private void sequentialOnNext(Collection createHost(ResolvedAddress addr) { + private Host 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); @@ -388,7 +388,7 @@ private Host createHost(ResolvedAddress addr) { // failed connect threshold is negative, meaning disabled. final HealthCheckConfig hostHealthCheckConfig = healthCheckConfig == null || healthCheckConfig.failedThreshold < 0 ? null : healthCheckConfig; - final Host host = new DefaultHost<>(lbDescription, addr, connectionPoolStrategy, + final Host host = new DefaultHost<>(lbDescription, addr, weight, connectionPoolStrategy, connectionFactory, hostObserver, hostHealthCheckConfig, indicator); if (indicator != null) { indicator.setHost(host); diff --git a/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/Host.java b/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/Host.java index a94c1f3c1b..c68e8f3e88 100644 --- a/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/Host.java +++ b/servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/Host.java @@ -51,6 +51,8 @@ interface Host extends Listen */ ResolvedAddress address(); + 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