Implement HttpContent.writeTo() async API

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/InputStreamRequestContent.java

@@ -30,28 +30,52 @@ public class InputStreamRequestContent extends InputStreamContentSource implemen
 {
     private final String contentType;
 
+    /**
+     * @deprecated use {@link #InputStreamRequestContent(String, InputStream, ByteBufferPool.Sized)} instead.
+     */
+    @Deprecated
     public InputStreamRequestContent(InputStream stream)
     {
         this(stream, 4096);
     }
 
+    /**
+     * @deprecated use {@link #InputStreamRequestContent(String, InputStream, ByteBufferPool.Sized)} instead.
+     */
+    @Deprecated
     public InputStreamRequestContent(InputStream stream, int bufferSize)
     {
         this("application/octet-stream", stream, bufferSize);
     }
 
+    /**
+     * @deprecated use {@link #InputStreamRequestContent(String, InputStream, ByteBufferPool.Sized)} instead.
+     */
+    @Deprecated
     public InputStreamRequestContent(String contentType, InputStream stream, int bufferSize)
     {
-        this(contentType, stream);
-        setBufferSize(bufferSize);
+        this(contentType, stream, new ByteBufferPool.Sized(null, false, bufferSize));
     }
 
+    /**
+     * @deprecated use {@link #InputStreamRequestContent(String, InputStream, ByteBufferPool.Sized)} instead.
+     */
+    @Deprecated
     public InputStreamRequestContent(String contentType, InputStream stream)
     {
         this(contentType, stream, null);
     }
 
+    /**
+     * @deprecated use {@link #InputStreamRequestContent(String, InputStream, ByteBufferPool.Sized)} instead.
+     */
+    @Deprecated
     public InputStreamRequestContent(String contentType, InputStream stream, ByteBufferPool bufferPool)
+    {
+        this(contentType, stream, new ByteBufferPool.Sized(bufferPool));
+    }
+
+    public InputStreamRequestContent(String contentType, InputStream stream, ByteBufferPool.Sized bufferPool)
     {
         super(stream, bufferPool);
         this.contentType = contentType;

jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/content/CachingHttpContentFactory.java

@@ -14,7 +14,6 @@
 package org.eclipse.jetty.http.content;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.time.Instant;
 import java.util.Set;
 import java.util.SortedSet;
@@ -30,9 +29,12 @@
 import org.eclipse.jetty.http.MimeTypes;
 import org.eclipse.jetty.http.PreEncodedHttpField;
 import org.eclipse.jetty.io.ByteBufferPool;
+import org.eclipse.jetty.io.Content;
 import org.eclipse.jetty.io.IOResources;
 import org.eclipse.jetty.io.Retainable;
 import org.eclipse.jetty.io.RetainableByteBuffer;
+import org.eclipse.jetty.util.BufferUtil;
+import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.NanoTime;
 import org.eclipse.jetty.util.StringUtil;
 import org.eclipse.jetty.util.resource.Resource;
@@ -328,30 +330,19 @@ public CachedHttpContent(String key, HttpContent httpContent)
             boolean isValid = true;
 
             // Read the content into memory if the HttpContent does not already have a buffer.
-            RetainableByteBuffer buffer;
-            ByteBuffer byteBuffer = httpContent.getByteBuffer();
-            if (byteBuffer == null)
+            RetainableByteBuffer buffer = null;
+            try
             {
-                try
-                {
-                    if (_contentLengthValue <= _maxCachedFileSize)
-                        buffer = IOResources.toRetainableByteBuffer(httpContent.getResource(), _bufferPool, _useDirectByteBuffers);
-                    else
-                        buffer = null;
-                }
-                catch (Throwable t)
-                {
-                    buffer = null;
-                    isValid = false;
-                    if (LOG.isDebugEnabled())
-                        LOG.warn("Failed to read Resource: {}", httpContent.getResource(), t);
-                    else
-                        LOG.warn("Failed to read Resource: {} - {}", httpContent.getResource(), t.toString());
-                }
+                if (_contentLengthValue <= _maxCachedFileSize)
+                    buffer = IOResources.toRetainableByteBuffer(httpContent.getResource(), _bufferPool, _useDirectByteBuffers);
             }
-            else
+            catch (Throwable t)
             {
-                buffer = RetainableByteBuffer.wrap(byteBuffer);
+                isValid = false;
+                if (LOG.isDebugEnabled())
+                    LOG.warn("Failed to read Resource: {}", httpContent.getResource(), t);
+                else
+                    LOG.warn("Failed to read Resource: {} - {}", httpContent.getResource(), t.toString());
             }
 
             _buffer = buffer;
@@ -373,12 +364,6 @@ public long getContentLengthValue()
             return _contentLengthValue;
         }
 
-        @Override
-        public ByteBuffer getByteBuffer()
-        {
-            return _buffer == null ? null : _buffer.getByteBuffer().asReadOnlyBuffer();
-        }
-
         @Override
         public long getBytesOccupied()
         {
@@ -403,6 +388,20 @@ public String getKey()
             return _cacheKey;
         }
 
+        @Override
+        public void writeTo(Content.Sink sink, long offset, long length, Callback callback)
+        {
+            try
+            {
+                sink.write(true, BufferUtil.slice(_buffer.getByteBuffer(), (int)offset, (int)length), callback);
+            }
+            catch (Throwable x)
+            {
+                // BufferUtil.slice() may fail if offset and/or length are out of bounds.
+                callback.failed(x);
+            }
+        }
+
         @Override
         public boolean retain()
         {
@@ -596,9 +595,9 @@ public Resource getResource()
         }
 
         @Override
-        public ByteBuffer getByteBuffer()
+        public void writeTo(Content.Sink sink, long offset, long length, Callback callback)
         {
-            return null;
+            sink.write(true, BufferUtil.EMPTY_BUFFER, callback);
         }
 
         @Override

jetty-core/jetty-http/src/main/java/org/eclipse/jetty/http/content/FileMappingHttpContentFactory.java

@@ -15,20 +15,25 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.file.Path;
 import java.util.Objects;
 
+import org.eclipse.jetty.io.Content;
 import org.eclipse.jetty.util.BufferUtil;
-import org.eclipse.jetty.util.thread.AutoLock;
+import org.eclipse.jetty.util.Callback;
+import org.eclipse.jetty.util.IteratingNestedCallback;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class FileMappingHttpContentFactory implements HttpContent.Factory
 {
     private static final Logger LOG = LoggerFactory.getLogger(FileMappingHttpContentFactory.class);
     private static final int DEFAULT_MIN_FILE_SIZE = 16 * 1024;
+    private static final int DEFAULT_MAX_BUFFER_SIZE = Integer.MAX_VALUE;
 
     private final HttpContent.Factory _factory;
     private final int _minFileSize;
+    private final int _maxBufferSize;
 
     /**
      * Construct a {@link FileMappingHttpContentFactory} which can use file mapped buffers.
@@ -39,19 +44,21 @@ public class FileMappingHttpContentFactory implements HttpContent.Factory
      */
     public FileMappingHttpContentFactory(HttpContent.Factory factory)
     {
-        this(factory, DEFAULT_MIN_FILE_SIZE);
+        this(factory, DEFAULT_MIN_FILE_SIZE, DEFAULT_MAX_BUFFER_SIZE);
     }
 
     /**
      * Construct a {@link FileMappingHttpContentFactory} which can use file mapped buffers.
      *
      * @param factory the wrapped {@link HttpContent.Factory} to use.
      * @param minFileSize the minimum size of an {@link HttpContent} before trying to use a file mapped buffer.
+     * @param maxBufferSize the maximum size of the memory mapped buffers
      */
-    public FileMappingHttpContentFactory(HttpContent.Factory factory, int minFileSize)
+    public FileMappingHttpContentFactory(HttpContent.Factory factory, int minFileSize, int maxBufferSize)
     {
         _factory = Objects.requireNonNull(factory);
         _minFileSize = minFileSize;
+        _maxBufferSize = maxBufferSize;
     }
 
     @Override
@@ -60,71 +67,147 @@ public HttpContent getContent(String path) throws IOException
         HttpContent content = _factory.getContent(path);
         if (content != null)
         {
-            long contentLength = content.getContentLengthValue();
-            if (contentLength > _minFileSize && contentLength < Integer.MAX_VALUE)
-                return new FileMappedHttpContent(content);
+            try
+            {
+                long contentLength = content.getContentLengthValue();
+                if (contentLength < _minFileSize)
+                    return content;
+                return contentLength <= _maxBufferSize ? new SingleBufferFileMappedHttpContent(content) : new MultiBufferFileMappedHttpContent(content, _maxBufferSize);
+            }
+            catch (IOException e)
+            {
+                if (LOG.isDebugEnabled())
+                    LOG.debug("Error getting Mapped Buffer", e);
+                // Fall through to return the content gotten from the factory.
+            }
         }
         return content;
     }
 
-    private static class FileMappedHttpContent extends HttpContent.Wrapper
+    private static class SingleBufferFileMappedHttpContent extends HttpContent.Wrapper
     {
-        private static final ByteBuffer SENTINEL_BUFFER = BufferUtil.allocate(0);
-
-        private final AutoLock _lock = new AutoLock();
-        private final HttpContent _content;
-        private volatile ByteBuffer _buffer;
+        private final ByteBuffer _buffer;
 
-        public FileMappedHttpContent(HttpContent content)
+        private SingleBufferFileMappedHttpContent(HttpContent content) throws IOException
         {
             super(content);
-            this._content = content;
+            Path path = content.getResource().getPath();
+            if (path == null)
+                throw new IOException("Cannot memory map Content whose Resource is not backed by a Path: " + content.getResource());
+            _buffer = BufferUtil.toMappedBuffer(path);
         }
 
         @Override
-        public ByteBuffer getByteBuffer()
+        public void writeTo(Content.Sink sink, long offset, long length, Callback callback)
         {
-            ByteBuffer buffer = _buffer;
-            if (buffer != null)
-                return (buffer == SENTINEL_BUFFER) ? super.getByteBuffer() : buffer.asReadOnlyBuffer();
-
-            try (AutoLock lock = _lock.lock())
+            try
+            {
+                sink.write(true, BufferUtil.slice(_buffer, (int)offset, (int)length), callback);
+            }
+            catch (Throwable x)
             {
-                if (_buffer == null)
-                    _buffer = getMappedByteBuffer();
-                return (_buffer == SENTINEL_BUFFER) ? super.getByteBuffer() : _buffer.asReadOnlyBuffer();
+                callback.failed(x);
             }
         }
 
         @Override
         public long getBytesOccupied()
         {
-            ByteBuffer buffer = _buffer;
-            if (buffer != null)
-                return (buffer == SENTINEL_BUFFER) ? super.getBytesOccupied() : 0;
+            return _buffer.remaining();
+        }
+    }
+
+    private static class MultiBufferFileMappedHttpContent extends HttpContent.Wrapper
+    {
+        private final ByteBuffer[] _buffers;
+        private final int maxBufferSize;
+        private final long _bytesOccupied;
+
+        private MultiBufferFileMappedHttpContent(HttpContent content, int maxBufferSize) throws IOException
+        {
+            super(content);
+            this.maxBufferSize = maxBufferSize;
+            Path path = content.getResource().getPath();
+            if (path == null)
+                throw new IOException("Cannot memory map Content whose Resource is not backed by a Path: " + content.getResource());
 
-            try (AutoLock lock = _lock.lock())
+            long contentLength = content.getContentLengthValue();
+            int bufferCount = Math.toIntExact(contentLength / maxBufferSize);
+            _buffers = new ByteBuffer[bufferCount];
+            long currentPos = 0L;
+            long total = 0L;
+            for (int i = 0; i < _buffers.length; i++)
             {
-                if (_buffer == null)
-                    _buffer = getMappedByteBuffer();
-                return (_buffer == SENTINEL_BUFFER) ? super.getBytesOccupied() : 0;
+                long len = Math.min(contentLength - currentPos, maxBufferSize);
+                _buffers[i] = BufferUtil.toMappedBuffer(path, currentPos, len);
+                currentPos += len;
+                total += _buffers[i].remaining();
             }
+            _bytesOccupied = total;
         }
 
-        private ByteBuffer getMappedByteBuffer()
+        @Override
+        public void writeTo(Content.Sink sink, long offset, long length, Callback callback)
         {
             try
             {
-                ByteBuffer byteBuffer = BufferUtil.toMappedBuffer(_content.getResource().getPath());
-                return (byteBuffer == null) ? SENTINEL_BUFFER : byteBuffer;
+                if (offset > getBytesOccupied())
+                    throw new IllegalArgumentException("Offset outside of mapped file range");
+                if (length > -1 && length + offset > getBytesOccupied())
+                    throw new IllegalArgumentException("Offset / length outside of mapped file range");
+
+                int beginIndex = Math.toIntExact(offset / maxBufferSize);
+                int firstOffset = Math.toIntExact(offset % maxBufferSize);
+
+                int endIndex = calculateEndIndex(offset, length);
+                int lastLen = calculateLastLen(offset, length);
+                new IteratingNestedCallback(callback)
+                {
+                    int index = beginIndex;
+                    @Override
+                    protected Action process()
+                    {
+                        if (index > endIndex)
+                            return Action.SUCCEEDED;
+
+                        ByteBuffer currentBuffer = _buffers[index];
+                        int offset = index == beginIndex ? firstOffset : 0;
+                        int len = index == endIndex ? lastLen : -1;
+                        boolean last = index == endIndex;
+                        index++;
+                        sink.write(last, BufferUtil.slice(currentBuffer, offset, len), this);
+                        return Action.SCHEDULED;
+                    }
+                }.iterate();
             }
-            catch (Throwable t)
+            catch (Throwable x)
             {
-                if (LOG.isDebugEnabled())
-                    LOG.debug("Error getting Mapped Buffer", t);
+                callback.failed(x);
             }
+        }
+
+        private int calculateLastLen(long offset, long length)
+        {
+            if (length == 0)
+                return 0;
+            int lastLen = length < 0 ? -1 : Math.toIntExact((length + offset) % maxBufferSize);
+            if (Math.toIntExact((length + offset) / maxBufferSize) == _buffers.length)
+                lastLen = -1;
+            return lastLen;
+        }
+
+        private int calculateEndIndex(long offset, long length)
+        {
+            int endIndex = length < 0 ? (_buffers.length - 1) : Math.toIntExact((length + offset) / maxBufferSize);
+            if (endIndex == _buffers.length)
+                endIndex--;
+            return endIndex;
+        }
 
-            return SENTINEL_BUFFER;
+        @Override
+        public long getBytesOccupied()
+        {
+            return _bytesOccupied;
         }
     }
 }