Merge pull request #5992 from eclipse-vertx/alt-svc-improvements

HTTP alternative services improvements

Author: vietj

vertx-core/src/main/java/io/vertx/core/http/impl/HttpClientImpl.java

@@ -35,6 +35,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 import java.util.regex.Pattern;
@@ -100,7 +101,7 @@ public class HttpClientImpl extends HttpClientBase implements HttpClientInternal
 
     this.tcpTransport = tcpTransport;
     this.quicTransport = quicTransport;
-    this.originEndpoints = new OriginResolver<>(vertx, resolveAll);
+    this.originEndpoints = new OriginResolver<>(vertx, resolveAll, this);
     this.resolver = (EndpointResolverInternal) resolver;
     this.originResolver = new EndpointResolverImpl<>(vertx, originEndpoints, resolveAll ? loadBalancer : LoadBalancer.FIRST, resolverKeepAlive.toMillis());
     this.poolOptions = poolOptions;
@@ -187,9 +188,10 @@ private Function<EndpointKey, SharedHttpClientConnectionGroup> httpEndpointProvi
       ClientMetrics clientMetrics = HttpClientImpl.this.httpMetrics != null ? HttpClientImpl.this.httpMetrics.createEndpointMetrics(address, maxPoolSize) : null;
       PoolMetrics poolMetrics = HttpClientImpl.this.httpMetrics != null ? vertx.metrics().createPoolMetrics("http", key.authority.toString(), maxPoolSize) : null;
       ProxyOptions proxyOptions = key.proxyOptions;
+      ClientSSLOptions sslOptions = key.sslOptions;
       if (proxyOptions != null && !key.ssl && proxyOptions.getType() == ProxyType.HTTP) {
         SocketAddress server = SocketAddress.inetSocketAddress(proxyOptions.getPort(), proxyOptions.getHost());
-        key = new EndpointKey(key.ssl, key.protocol, key.sslOptions, proxyOptions, server, key.authority);
+        key = new EndpointKey(key.ssl, key.protocol, sslOptions, proxyOptions, server, key.authority);
         proxyOptions = null;
       }
       HttpVersion protocol = key.protocol;
@@ -199,7 +201,7 @@ private Function<EndpointKey, SharedHttpClientConnectionGroup> httpEndpointProvi
       } else {
         protocols = List.of(protocol);
       }
-      HttpConnectParams params = new HttpConnectParams(protocols, key.sslOptions, proxyOptions, key.ssl);
+      HttpConnectParams params = new HttpConnectParams(protocols, sslOptions, proxyOptions, key.ssl);
       Function<SharedHttpClientConnectionGroup, SharedHttpClientConnectionGroup.Pool> p = group -> {
         int queueMaxSize = poolOptions.getMaxWaitQueueSize();
         int http1MaxSize = poolOptions.getHttp1MaxSize();
@@ -220,7 +222,7 @@ private Function<EndpointKey, SharedHttpClientConnectionGroup> httpEndpointProvi
               if (altSvc instanceof AltSvc.Clear) {
                 originEndpoints.clearAlternatives(evt.origin);
               } else if (altSvc instanceof AltSvc.ListOfValue) {
-                originEndpoints.updateAlternatives(evt.origin, (AltSvc.ListOfValue)altSvc);
+                originEndpoints.updateAlternatives(sslOptions, evt.origin, (AltSvc.ListOfValue)altSvc);
               }
             });
           }
@@ -593,17 +595,49 @@ private Future<HttpClientRequest> doRequest(
         // For HTTPS we must handle SNI to consider an alternative
         HostAndPort altUsed;
         if (followAlternativeServices && server instanceof Origin && ("https".equals((originServer = (Origin)server).scheme) && originServer.host.indexOf('.') > 0)) {
-          lookup = endpoint.selectServer(s -> {
-            OriginServer unwrap = (OriginServer) s.unwrap();
-            return protocol_ == unwrap.protocol;
-          });
-          protocol = protocol_;
+          if (protocol_ != null) {
+            ProtocolFilter filter;
+            switch (protocol_) {
+              case H3:
+                filter = ProtocolFilter.H3;
+                break;
+              case H2:
+                filter = ProtocolFilter.H2;
+                break;
+              case HTTP_1_1:
+                filter = ProtocolFilter.HTTP_1_1;
+                break;
+              case HTTP_1_0:
+                filter = ProtocolFilter.HTTP_1_0;
+                break;
+              default:
+                throw new AssertionError();
+            }
+            lookup = endpoint.selectServer(filter);
+            protocol = protocol_;
+          } else {
+            Set<String> protocols = endpoint.protocols();
+            if (!protocols.isEmpty()) {
+              List<ProtocolFilter> list = List.of(ProtocolFilter.H3, ProtocolFilter.H2, ProtocolFilter.HTTP_1_1, ProtocolFilter.HTTP_1_0);
+              lookup = null;
+              protocol = null;
+              for (ProtocolFilter candidate : list) {
+                if (protocols.contains(candidate.protocol.id())) {
+                  lookup = endpoint.selectServer(candidate);
+                  protocol = candidate.protocol;
+                }
+              }
+            } else {
+              lookup = null;
+              protocol = null;
+            }
+          }
           if (lookup == null) {
             altUsed = null;
             lookup = endpoint.selectServer();
           } else {
             OriginServer unwrap = (OriginServer) lookup.unwrap();
-            altUsed = unwrap.authority;
+            altUsed = unwrap.primary ? null : unwrap.authority;
           }
         } else {
           protocol = protocol_;
@@ -615,23 +649,32 @@ private Future<HttpClientRequest> doRequest(
           throw new IllegalStateException("No results for " + server);
         }
         SocketAddress address = lookup2.address();
-        EndpointKey key = new EndpointKey(useSSL, protocol != null ? protocol.version() : null, sslOptions, null, address, authority != null ? authority : HostAndPort.create(address.host(), address.port()));
-        return resourceManager.withResourceAsync(key, httpEndpointProvider(followAlternativeServices && useSSL, transport), (e, created) -> {
-          Future<Lease<HttpClientConnection>> fut2 = e.requestConnection(streamCtx, connectTimeout);
-          ServerInteraction endpointRequest = lookup2.newInteraction();
-          return fut2.andThen(ar -> {
-            if (ar.failed()) {
-              endpointRequest.reportFailure(ar.cause());
+        return getPool(
+          followAlternativeServices && useSSL && altUsed == null,
+          useSSL,
+          protocol,
+          sslOptions,
+          address,
+          authority != null ? authority : HostAndPort.create(address.host(), address.port()),
+          new Function<SharedHttpClientConnectionGroup, Future<ConnectionObtainedResult>>() {
+            @Override
+            public Future<ConnectionObtainedResult> apply(SharedHttpClientConnectionGroup pool) {
+              Future<Lease<HttpClientConnection>> fut2 = pool.requestConnection(streamCtx, connectTimeout);
+              ServerInteraction endpointRequest = lookup2.newInteraction();
+              return fut2.andThen(ar -> {
+                if (ar.failed()) {
+                  endpointRequest.reportFailure(ar.cause());
+                }
+              }).compose(lease -> {
+                HttpClientConnection conn = lease.get();
+                return conn.createStream(streamCtx).map(stream -> {
+                  HttpClientStream wrapped = new StatisticsGatheringHttpClientStream(stream, endpointRequest);
+                  wrapped.closeHandler(v -> lease.recycle());
+                  return new ConnectionObtainedResult(wrapped, lease, altUsed);
+                });
+              });
             }
-          }).compose(lease -> {
-            HttpClientConnection conn = lease.get();
-            return conn.createStream(streamCtx).map(stream -> {
-              HttpClientStream wrapped = new StatisticsGatheringHttpClientStream(stream, endpointRequest);
-              wrapped.closeHandler(v -> lease.recycle());
-              return new ConnectionObtainedResult(wrapped, lease, altUsed);
-            });
           });
-        });
       });
     if (future == null) {
       // I think this is not possible - so remove it
@@ -641,6 +684,42 @@ private Future<HttpClientRequest> doRequest(
     }
   }
 
+  Future<?> checkConnect(OriginServer primary, OriginAlternative alternative, OriginServer server, ClientSSLOptions sslOptions) {
+    return getPool(false, true, alternative.protocol, sslOptions, server.address, primary.authority, new Function<SharedHttpClientConnectionGroup, Future<Boolean>>() {
+      @Override
+      public Future<Boolean> apply(SharedHttpClientConnectionGroup group) {
+        if (group.size() > 0) {
+          return Future.succeededFuture();
+        } else {
+          // Get something better
+          Future<Lease<HttpClientConnection>> f = group.requestConnection(vertx.getOrCreateContext(), 10_000);
+          return f.map(lease -> {
+            lease.recycle();
+            return null;
+          });
+        }
+      }
+    });
+  }
+
+  <T> Future<T> getPool(boolean resolveOrigin,
+                        boolean useSSL,
+                        HttpProtocol protocol,
+                        ClientSSLOptions sslOptions,
+                        SocketAddress server,
+                        HostAndPort authority,
+                        Function<SharedHttpClientConnectionGroup, Future<T>> function) {
+    EndpointKey key = new EndpointKey(useSSL, protocol != null ? protocol.version() : null, sslOptions, null, server, authority);
+    HttpClientTransport transport;
+    if (protocol != null && protocol.version() == HttpVersion.HTTP_3) {
+      transport = quicTransport;
+    } else {
+      transport = tcpTransport;
+    }
+    Function<EndpointKey, SharedHttpClientConnectionGroup> provider = httpEndpointProvider(resolveOrigin, transport);
+    return resourceManager.withResourceAsync(key, provider, (group, created) -> function.apply(group));
+  }
+
   private Future<HttpClientRequest> wrap(HttpMethod method,
                                          String requestURI,
                                          MultiMap headers,

vertx-core/src/main/java/io/vertx/core/http/impl/OriginEndpoint.java

@@ -31,16 +31,16 @@ public class OriginEndpoint<L> {
   final L list;
   private final Map<OriginAlternative, OriginServer> alternatives;
 
-  Map<OriginAlternative, Long> update;
-  private volatile boolean valid;
+  Map<OriginAlternative, OriginServer> updates;
+  volatile boolean valid;
 
   OriginEndpoint(Origin origin, OriginServer primary, EndpointBuilder<L, OriginServer> builder, Map<OriginAlternative, OriginServer> alternatives) {
     this(origin, List.of(primary), builder, alternatives);
   }
 
   OriginEndpoint(Origin origin, List<OriginServer> primaries, EndpointBuilder<L, OriginServer> builder, Map<OriginAlternative, OriginServer> alternatives) {
 
-    L list = refresh(builder, primaries, alternatives);
+    L list = buildListOfServers(builder, primaries, alternatives);
 
     this.timestamp = System.currentTimeMillis();
     this.primary = primaries.get(0);
@@ -49,15 +49,20 @@ public class OriginEndpoint<L> {
     this.builder = builder;
     this.alternatives = alternatives;
     this.list = list;
+    this.updates = Collections.emptyMap();
     this.valid = true;
   }
 
-  private L refresh(EndpointBuilder<L, OriginServer> builder, List<OriginServer> primaries, Map<OriginAlternative, OriginServer> alternatives) {
+  private L buildListOfServers(EndpointBuilder<L, OriginServer> builder, List<OriginServer> primaries, Map<OriginAlternative, OriginServer> alternatives) {
     for (OriginServer primary : primaries) {
-      builder = builder.addServer(primary);
+      if (primary.available) {
+        builder = builder.addServer(primary);
+      }
     }
     for (OriginServer alternativeServer : alternatives.values()) {
-      builder.addServer(alternativeServer);
+      if (alternativeServer.available) {
+        builder.addServer(alternativeServer);
+      }
     }
     return builder.build();
   }
@@ -75,14 +80,14 @@ boolean validate() {
     return valid;
   }
 
-  void clearAlternatives() {
-    update = Collections.emptyMap();
-    valid = alternatives.isEmpty();
+  void updateAlternatives(Map<OriginAlternative, OriginServer> updates) {
+    this.updates = updates;
+    this.valid = false;
   }
 
-  void updateAlternatives(AltSvc.ListOfValue altSvc) {
+  Map<OriginAlternative, Long> shouldRefresh(AltSvc.ListOfValue altSvc) {
     long now = System.currentTimeMillis();
-    Map<OriginAlternative, Long> list = new LinkedHashMap<>();
+    Map<OriginAlternative, Long> updates = new LinkedHashMap<>();
     boolean valid = true;
     for (AltSvc.Value altSvcValue : altSvc) {
       HttpProtocol protocol = HttpProtocol.fromId(altSvcValue.protocolId());
@@ -113,18 +118,15 @@ void updateAlternatives(AltSvc.ListOfValue altSvc) {
           long value = now + maxAge * 1000 / 2;
           valid = (value < alternativeCachedExpiration);
         }
-        list.put(alternative, maxAge);
+        updates.put(alternative, maxAge);
       }
     }
     if (valid) {
       // 1. check now we don't have extra unwanted keys
       for (OriginAlternative alternative : alternatives.keySet()) {
-        valid &= list.containsKey(alternative);
+        valid &= updates.containsKey(alternative);
       }
     }
-    if (!valid) {
-      this.update = list;
-      this.valid = false;
-    }
+    return valid ? null : updates;
   }
 }