Make Content.Chunk a RetainableByteBuffer

jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/GZIPContentDecoder.java

@@ -109,7 +109,7 @@ public RetainableByteBuffer decode(ByteBuffer compressed)
             RetainableByteBuffer result = acquire(length);
             for (RetainableByteBuffer buffer : _inflateds)
             {
-                BufferUtil.append(result.getByteBuffer(), buffer.getByteBuffer());
+                buffer.appendTo(result);
                 buffer.release();
             }
             _inflateds.clear();

jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/MultiPart.java

@@ -1114,7 +1114,7 @@ else if (type != HttpTokens.Type.SPACE && type != HttpTokens.Type.HTAB)
                     if (state == State.EPILOGUE)
                         notifyComplete();
                     else
-                        throw new EOFException("unexpected EOF");
+                        throw new EOFException("unexpected EOF in " + state);
                 }
             }
             catch (Throwable x)

jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/HTTP2Connection.java

@@ -478,6 +478,12 @@ public boolean canRetain()
             return retainable.canRetain();
         }
 
+        @Override
+        public boolean isRetained()
+        {
+            return retainable.isRetained();
+        }
+
         @Override
         public void retain()
         {

jetty-core/jetty-http2/jetty-http2-common/src/main/java/org/eclipse/jetty/http2/api/Stream.java

@@ -438,7 +438,7 @@ public default void onClosed(Stream stream)
     /**
      * <p>A {@link Retainable} wrapper of a {@link DataFrame}.</p>
      */
-    public abstract static class Data implements Retainable
+    abstract class Data implements Retainable
     {
         public static Data eof(int streamId)
         {

jetty-core/jetty-http3/jetty-http3-common/src/main/java/org/eclipse/jetty/http3/HTTP3StreamConnection.java

@@ -445,6 +445,12 @@ public boolean canRetain()
             return retainable.canRetain();
         }
 
+        @Override
+        public boolean isRetained()
+        {
+            return retainable.isRetained();
+        }
+
         @Override
         public void retain()
         {

jetty-core/jetty-http3/jetty-http3-common/src/main/java/org/eclipse/jetty/http3/api/Stream.java

@@ -384,7 +384,7 @@ public default void onFailure(Stream.Server stream, long error, Throwable failur
      *
      * @see Stream#readData()
      */
-    public abstract static class Data implements Retainable
+    abstract class Data implements Retainable
     {
         public static final Data EOF = new EOFData();
 

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/ArrayByteBufferPool.java

@@ -666,9 +666,19 @@ public Tracking(int minCapacity, int maxCapacity, int maxBucketSize)
             this(minCapacity, maxCapacity, maxBucketSize, -1L, -1L);
         }
 
+        public Tracking(int minCapacity, int factor, int maxCapacity, int maxBucketSize)
+        {
+            this(minCapacity, factor, maxCapacity, maxBucketSize, 0L, 0L);
+        }
+
         public Tracking(int minCapacity, int maxCapacity, int maxBucketSize, long maxHeapMemory, long maxDirectMemory)
         {
-            super(minCapacity, -1, maxCapacity, maxBucketSize, maxHeapMemory, maxDirectMemory);
+            this(minCapacity, -1, maxCapacity, maxBucketSize, maxHeapMemory, maxDirectMemory);
+        }
+
+        public Tracking(int minCapacity, int factor, int maxCapacity, int maxBucketSize, long maxHeapMemory, long maxDirectMemory)
+        {
+            super(minCapacity, factor, maxCapacity, maxBucketSize, maxHeapMemory, maxDirectMemory);
         }
 
         @Override

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/ChunkAccumulator.java

@@ -27,7 +27,9 @@
 /**
  * An accumulator of {@link Content.Chunk}s used to facilitate minimal copy
  * aggregation of multiple chunks.
+ * @deprecated use {@link Content.Source#asRetainableByteBuffer(Content.Source, ByteBufferPool, boolean, int)} instead.
  */
+@Deprecated
 public class ChunkAccumulator
 {
     private static final ByteBufferPool NON_POOLING = new ByteBufferPool.NonPooling();
@@ -109,7 +111,7 @@ public RetainableByteBuffer take(ByteBufferPool pool, boolean direct)
         for (Chunk chunk : _chunks)
         {
             offset += chunk.remaining();
-            BufferUtil.append(buffer.getByteBuffer(), chunk.getByteBuffer());
+            chunk.appendTo(buffer);
             chunk.release();
         }
         assert offset == _length;

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/Content.java

@@ -13,6 +13,7 @@
 
 package org.eclipse.jetty.io;
 
+import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -33,6 +34,7 @@
 import org.eclipse.jetty.io.internal.ContentCopier;
 import org.eclipse.jetty.io.internal.ContentSourceByteBuffer;
 import org.eclipse.jetty.io.internal.ContentSourceConsumer;
+import org.eclipse.jetty.io.internal.ContentSourceRetainableByteBuffer;
 import org.eclipse.jetty.io.internal.ContentSourceString;
 import org.eclipse.jetty.util.Blocker;
 import org.eclipse.jetty.util.BufferUtil;
@@ -192,7 +194,7 @@ static ByteBuffer asByteBuffer(Source source) throws IOException
          */
         static CompletableFuture<byte[]> asByteArrayAsync(Source source, int maxSize)
         {
-            return new ChunkAccumulator().readAll(source, maxSize);
+            return asRetainableByteBuffer(source, null, false, maxSize).thenApply(rbb -> rbb.getByteBuffer().array());
         }
 
         /**
@@ -230,7 +232,31 @@ static CompletableFuture<ByteBuffer> asByteBufferAsync(Source source, int maxSiz
          */
         static CompletableFuture<RetainableByteBuffer> asRetainableByteBuffer(Source source, ByteBufferPool pool, boolean direct, int maxSize)
         {
-            return new ChunkAccumulator().readAll(source, pool, direct, maxSize);
+            Promise.Completable<RetainableByteBuffer> promise = new Promise.Completable<>()
+            {
+                @Override
+                public void succeeded(RetainableByteBuffer result)
+                {
+                    result.retain();
+                    super.succeeded(result);
+                }
+            };
+            asRetainableByteBuffer(source, pool, direct, maxSize, promise);
+            return promise;
+        }
+
+        /**
+         * <p>Reads, non-blocking, the whole content source into a {@link RetainableByteBuffer}.</p>
+         *
+         * @param source the source to read
+         * @param pool The {@link ByteBufferPool} to acquire the buffer from, or null for a non {@link Retainable} buffer
+         * @param direct True if the buffer should be direct.
+         * @param maxSize The maximum size to read, or -1 for no limit
+         * @param promise the promise to notify when the whole content has been read into a RetainableByteBuffer.
+         */
+        static void asRetainableByteBuffer(Source source, ByteBufferPool pool, boolean direct, int maxSize, Promise<RetainableByteBuffer> promise)
+        {
+            new ContentSourceRetainableByteBuffer(source, pool, direct, maxSize, promise).run();
         }
 
         /**
@@ -470,6 +496,43 @@ default boolean rewind()
      */
     public interface Sink
     {
+        /**
+         * <p>Wraps the given {@link OutputStream} as a {@link Sink}.
+         * @param out The stream to wrap
+         * @return A sink wrapping the stream
+         */
+        static Sink from(OutputStream out)
+        {
+            return new Sink()
+            {
+                boolean closed;
+
+                @Override
+                public void write(boolean last, ByteBuffer byteBuffer, Callback callback)
+                {
+                    if (closed)
+                    {
+                        callback.failed(new EOFException());
+                        return;
+                    }
+                    try
+                    {
+                        BufferUtil.writeTo(byteBuffer, out);
+                        if (last)
+                        {
+                            closed = true;
+                            out.close();
+                        }
+                        callback.succeeded();
+                    }
+                    catch (Throwable t)
+                    {
+                        callback.failed(t);
+                    }
+                }
+            };
+        }
+
         /**
          * <p>Wraps the given content sink with a buffering sink.</p>
          *
@@ -561,7 +624,7 @@ static void write(Sink sink, boolean last, String utf8Content, Callback callback
      * to release the {@code ByteBuffer} back into a pool), or the
      * {@link #release()} method overridden.</p>
      */
-    public interface Chunk extends Retainable
+    public interface Chunk extends RetainableByteBuffer
     {
         /**
          * <p>An empty, non-last, chunk.</p>
@@ -804,11 +867,6 @@ static boolean isFailure(Chunk chunk, boolean last)
             return chunk != null && chunk.getFailure() != null && chunk.isLast() == last;
         }
 
-        /**
-         * @return the ByteBuffer of this Chunk
-         */
-        ByteBuffer getByteBuffer();
-
         /**
          * Get a failure (which may be from a {@link Source#fail(Throwable) failure} or
          * a {@link Source#fail(Throwable, boolean) warning}), if any, associated with the chunk.
@@ -831,59 +889,10 @@ default Throwable getFailure()
          */
         boolean isLast();
 
-        /**
-         * @return the number of bytes remaining in this Chunk
-         */
-        default int remaining()
-        {
-            return getByteBuffer().remaining();
-        }
-
-        /**
-         * @return whether this Chunk has remaining bytes
-         */
-        default boolean hasRemaining()
-        {
-            return getByteBuffer().hasRemaining();
-        }
-
-        /**
-         * <p>Copies the bytes from this Chunk to the given byte array.</p>
-         *
-         * @param bytes the byte array to copy the bytes into
-         * @param offset the offset within the byte array
-         * @param length the maximum number of bytes to copy
-         * @return the number of bytes actually copied
-         */
-        default int get(byte[] bytes, int offset, int length)
-        {
-            ByteBuffer b = getByteBuffer();
-            if (b == null || !b.hasRemaining())
-                return 0;
-            length = Math.min(length, b.remaining());
-            b.get(bytes, offset, length);
-            return length;
-        }
-
-        /**
-         * <p>Skips, advancing the ByteBuffer position, the given number of bytes.</p>
-         *
-         * @param length the maximum number of bytes to skip
-         * @return the number of bytes actually skipped
-         */
-        default int skip(int length)
-        {
-            if (length == 0)
-                return 0;
-            ByteBuffer byteBuffer = getByteBuffer();
-            length = Math.min(byteBuffer.remaining(), length);
-            byteBuffer.position(byteBuffer.position() + length);
-            return length;
-        }
-
         /**
          * @return an immutable version of this Chunk
          */
+        @Deprecated(forRemoval = true, since = "12.0.9")
         default Chunk asReadOnly()
         {
             if (getByteBuffer().isReadOnly())

jetty-core/jetty-io/src/main/java/org/eclipse/jetty/io/Retainable.java

@@ -62,6 +62,15 @@ default boolean canRetain()
         return false;
     }
 
+    /**
+     * @return whether this instance is retained
+     * @see ReferenceCounter#isRetained()
+     */
+    default boolean isRetained()
+    {
+        return false;
+    }
+
     /**
      * <p>Retains this resource, potentially incrementing a reference count if there are resources that will be released.</p>
      */
@@ -103,6 +112,12 @@ public boolean canRetain()
             return getWrapped().canRetain();
         }
 
+        @Override
+        public boolean isRetained()
+        {
+            return getWrapped().isRetained();
+        }
+
         @Override
         public void retain()
         {
@@ -141,7 +156,7 @@ public ReferenceCounter()
             this(1);
         }
 
-        protected ReferenceCounter(int initialCount)
+        public ReferenceCounter(int initialCount)
         {
             references = new AtomicInteger(initialCount);
         }
@@ -195,11 +210,7 @@ public boolean release()
             return ref == 0;
         }
 
-        /**
-         * <p>Returns whether {@link #retain()} has been called at least one more time than {@link #release()}.</p>
-         *
-         * @return whether this buffer is retained
-         */
+        @Override
         public boolean isRetained()
         {
             return references.get() > 1;