Work In Progress

jetty · Aug 27, 2021 · 7b78b90 · 7b78b90
1 parent 5834e5f
commit 7b78b90
Show file tree

Hide file tree

Showing 10 changed files with 324 additions and 242 deletions.
diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java b/jetty-client/src/main/java/org/eclipse/jetty/client/AbstractConnectionPool.java
@@ -70,14 +70,20 @@ protected AbstractConnectionPool(Destination destination, int maxConnections, Ca
 
     protected AbstractConnectionPool(HttpDestination destination, int maxConnections, boolean cache, Callback requester)
     {
-        this(destination, new Pool<>(Pool.StrategyType.FIRST, maxConnections, cache), requester);
+        this(destination, Pool.StrategyType.FIRST, maxConnections, cache, requester);
+    }
+
+    protected AbstractConnectionPool(HttpDestination destination, Pool.StrategyType strategy, int maxConnections, boolean cache, Callback requester)
+    {
+        this(destination, new Pool<>(strategy, maxConnections, cache), requester);
     }
 
     protected AbstractConnectionPool(HttpDestination destination, Pool<Connection> pool, Callback requester)
     {
         this.destination = destination;
         this.requester = requester;
         this.pool = pool;
+        pool.setMaxMultiplex(1); // Force the use of MultiUseEntry
         addBean(pool);
     }
 

diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/ConnectionPool.java b/jetty-client/src/main/java/org/eclipse/jetty/client/ConnectionPool.java
@@ -102,20 +102,4 @@ interface Factory
          */
         ConnectionPool newConnectionPool(HttpDestination destination);
     }
-
-    /**
-     * Marks a connection as supporting multiplexed requests.
-     */
-    interface Multiplexable
-    {
-        /**
-         * @return the max number of requests that can be multiplexed on a connection
-         */
-        int getMaxMultiplex();
-
-        /**
-         * @param maxMultiplex the max number of requests that can be multiplexed on a connection
-         */
-        void setMaxMultiplex(int maxMultiplex);
-    }
 }
diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/DuplexConnectionPool.java b/jetty-client/src/main/java/org/eclipse/jetty/client/DuplexConnectionPool.java
@@ -34,9 +34,10 @@ public DuplexConnectionPool(HttpDestination destination, int maxConnections, Cal
 
     public DuplexConnectionPool(HttpDestination destination, int maxConnections, boolean cache, Callback requester)
     {
-        this(destination, new Pool<>(Pool.StrategyType.FIRST, maxConnections, cache), requester);
+        super(destination, Pool.StrategyType.FIRST, maxConnections, cache, requester);
     }
 
+    @Deprecated
     public DuplexConnectionPool(HttpDestination destination, Pool<Connection> pool, Callback requester)
     {
         super(destination, pool, requester);

diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/MultiplexConnectionPool.java b/jetty-client/src/main/java/org/eclipse/jetty/client/MultiplexConnectionPool.java
@@ -34,12 +34,21 @@ public MultiplexConnectionPool(HttpDestination destination, int maxConnections,
 
     public MultiplexConnectionPool(HttpDestination destination, int maxConnections, boolean cache, Callback requester, int maxMultiplex)
     {
-        this(destination, new Pool<>(Pool.StrategyType.FIRST, maxConnections, cache), requester, maxMultiplex);
+        this(destination, Pool.StrategyType.FIRST, maxConnections, cache, requester, maxMultiplex);
     }
 
-    public MultiplexConnectionPool(HttpDestination destination, Pool<Connection> pool, Callback requester, int maxMultiplex)
+    public MultiplexConnectionPool(HttpDestination destination, Pool.StrategyType strategy, int maxConnections, boolean cache, Callback requester, int maxMultiplex)
     {
-        super(destination, pool, requester);
+        super(destination, new Pool<Connection>(strategy, maxConnections, cache)
+        {
+            @Override
+            protected int getMaxMultiplex(Connection connection)
+            {
+                if (connection instanceof Multiplexable)
+                    return ((Multiplexable)connection).getMaxMultiplex();
+                return super.getMaxMultiplex(connection);
+            }
+        }, requester);
         setMaxMultiplex(maxMultiplex);
     }
 
@@ -68,4 +77,15 @@ public void setMaxUsageCount(int maxUsageCount)
     {
         super.setMaxUsageCount(maxUsageCount);
     }
+
+    /**
+     * Marks a connection as supporting multiplexed requests.
+     */
+    public interface Multiplexable extends Connection
+    {
+        /**
+         * @return the max number of requests that can be multiplexed on a connection
+         */
+        int getMaxMultiplex();
+    }
 }
diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/RandomConnectionPool.java b/jetty-client/src/main/java/org/eclipse/jetty/client/RandomConnectionPool.java
@@ -31,6 +31,6 @@ public class RandomConnectionPool extends MultiplexConnectionPool
 {
     public RandomConnectionPool(HttpDestination destination, int maxConnections, Callback requester, int maxMultiplex)
     {
-        super(destination, new Pool<>(Pool.StrategyType.RANDOM, maxConnections, false), requester, maxMultiplex);
+        super(destination, Pool.StrategyType.RANDOM, maxConnections, false, requester, maxMultiplex);
     }
 }
diff --git a/jetty-client/src/main/java/org/eclipse/jetty/client/RoundRobinConnectionPool.java b/jetty-client/src/main/java/org/eclipse/jetty/client/RoundRobinConnectionPool.java
@@ -56,7 +56,7 @@ public RoundRobinConnectionPool(HttpDestination destination, int maxConnections,
 
     public RoundRobinConnectionPool(HttpDestination destination, int maxConnections, Callback requester, int maxMultiplex)
     {
-        super(destination, new Pool<>(Pool.StrategyType.ROUND_ROBIN, maxConnections, false), requester, maxMultiplex);
+        super(destination, Pool.StrategyType.ROUND_ROBIN, maxConnections, false, requester, maxMultiplex);
         // If there are queued requests and connections get
         // closed due to idle timeout or overuse, we want to
         // aggressively try to open new connections to replace

diff --git a/...-transport/src/main/java/org/eclipse/jetty/http2/client/http/HttpConnectionOverHTTP2.java b/...-transport/src/main/java/org/eclipse/jetty/http2/client/http/HttpConnectionOverHTTP2.java
@@ -28,22 +28,23 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import org.eclipse.jetty.client.ConnectionPool;
 import org.eclipse.jetty.client.HttpChannel;
 import org.eclipse.jetty.client.HttpConnection;
 import org.eclipse.jetty.client.HttpDestination;
 import org.eclipse.jetty.client.HttpExchange;
 import org.eclipse.jetty.client.HttpRequest;
+import org.eclipse.jetty.client.MultiplexConnectionPool;
 import org.eclipse.jetty.client.SendFailure;
 import org.eclipse.jetty.http.HttpVersion;
 import org.eclipse.jetty.http2.ErrorCode;
+import org.eclipse.jetty.http2.HTTP2Session;
 import org.eclipse.jetty.http2.api.Session;
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.util.thread.Sweeper;
 
-public class HttpConnectionOverHTTP2 extends HttpConnection implements Sweeper.Sweepable, ConnectionPool.Multiplexable
+public class HttpConnectionOverHTTP2 extends HttpConnection implements Sweeper.Sweepable, MultiplexConnectionPool.Multiplexable
 {
     private static final Logger LOG = Log.getLogger(HttpConnection.class);
 
@@ -53,7 +54,6 @@ public class HttpConnectionOverHTTP2 extends HttpConnection implements Sweeper.S
     private final AtomicInteger sweeps = new AtomicInteger();
     private final Session session;
     private boolean recycleHttpChannels = true;
-    private int maxMultiplex = 1;
 
     public HttpConnectionOverHTTP2(HttpDestination destination, Session session)
     {
@@ -76,14 +76,10 @@ public void setRecycleHttpChannels(boolean recycleHttpChannels)
         this.recycleHttpChannels = recycleHttpChannels;
     }
 
+    @Override
     public int getMaxMultiplex()
     {
-        return maxMultiplex;
-    }
-
-    public void setMaxMultiplex(int maxMultiplex)
-    {
-        this.maxMultiplex = maxMultiplex;
+        return ((HTTP2Session)session).getMaxLocalStreams();
     }
 
     @Override

diff --git a/...port/src/test/java/org/eclipse/jetty/http2/client/http/MultiplexedConnectionPoolTest.java b/...port/src/test/java/org/eclipse/jetty/http2/client/http/MultiplexedConnectionPoolTest.java
@@ -100,10 +100,12 @@ public void testMaxDurationConnectionsWithMultiplexedPool() throws Exception
         ConnectionPoolFactory factory = new ConnectionPoolFactory("duplex-maxDuration", destination ->
         {
             int maxConnections = destination.getHttpClient().getMaxConnectionsPerDestination();
-            Pool<Connection> pool = new Pool<>(Pool.StrategyType.FIRST, maxConnections, false);
-            poolRef.set(pool);
-            MultiplexConnectionPool connectionPool = new MultiplexConnectionPool(destination, pool, destination, MAX_MULTIPLEX)
+            MultiplexConnectionPool connectionPool = new MultiplexConnectionPool(destination, Pool.StrategyType.FIRST, maxConnections, false, destination, MAX_MULTIPLEX)
             {
+                {
+                    poolRef.set(getBean(Pool.class));
+                }
+
                 @Override
                 protected void onCreated(Connection connection)
                 {
@@ -161,22 +163,27 @@ public void testMaxDurationConnectionsWithMultiplexedPoolClosesExpiredConnection
         ConnectionPoolFactory factory = new ConnectionPoolFactory("duplex-maxDuration", destination ->
         {
             int maxConnections = destination.getHttpClient().getMaxConnectionsPerDestination();
-            Pool<Connection> pool = new Pool<>(Pool.StrategyType.FIRST, maxConnections, false);
-            poolRef.set(pool);
-            MultiplexConnectionPool connectionPool = new MultiplexConnectionPool(destination, pool, destination, MAX_MULTIPLEX)
-            {
-                @Override
-                protected void onCreated(Connection connection)
+            MultiplexConnectionPool connectionPool =
+                new MultiplexConnectionPool(destination, Pool.StrategyType.FIRST, maxConnections, false, destination, MAX_MULTIPLEX)
                 {
-                    poolCreateCounter.incrementAndGet();
-                }
+                    {
+                        poolRef.set(getBean(Pool.class));
+                    }
+
+                    @Override
+                    protected void onCreated(Connection connection)
+                    {
+                        poolCreateCounter.incrementAndGet();
+                    }
+
+                    @Override
+                    protected void removed(Connection connection)
+                    {
+                        poolRemoveCounter.incrementAndGet();
+                    }
+                };
+
 
-                @Override
-                protected void removed(Connection connection)
-                {
-                    poolRemoveCounter.incrementAndGet();
-                }
-            };
             connectionPool.setMaxDuration(maxDuration);
             return connectionPool;
         });