Improve DelayedHandler. #9051

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

@@ -489,15 +489,18 @@ public void setMaxParts(long maxParts)
          */
         public void configure(MultiPartConfig config)
         {
-            parser.setMaxParts(config.getMaxParts());
-            maxMemoryFileSize = config.getMaxMemoryPartSize();
-            maxFileSize = config.getMaxPartSize();
-            maxLength = config.getMaxSize();
-            parser.setPartHeadersMaxLength(config.getMaxHeadersSize());
-            useFilesForPartsWithoutFileName = config.isUseFilesForPartsWithoutFileName();
-            filesDirectory = config.getLocation();
-            complianceListener = config.getViolationListener();
-            compliance = config.getMultiPartCompliance();
+            if (config != null)
+            {
+                parser.setMaxParts(config.getMaxParts());
+                maxMemoryFileSize = config.getMaxMemoryPartSize();
+                maxFileSize = config.getMaxPartSize();
+                maxLength = config.getMaxSize();
+                parser.setPartHeadersMaxLength(config.getMaxHeadersSize());
+                useFilesForPartsWithoutFileName = config.isUseFilesForPartsWithoutFileName();
+                filesDirectory = config.getLocation();
+                complianceListener = config.getViolationListener();
+                compliance = config.getMultiPartCompliance();
+            }
         }
 
         // Only used for testing.

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

@@ -52,12 +52,9 @@ protected AbstractConnection(EndPoint endPoint, Executor executor)
         _readCallback = new ReadCallback();
     }
 
-    @Deprecated
     @Override
     public InvocationType getInvocationType()
     {
-        // TODO consider removing the #fillInterested method from the connection and only use #fillInterestedCallback
-        //      so a connection need not be Invocable
         return Invocable.super.getInvocationType();
     }
 
@@ -170,7 +167,7 @@ public boolean isFillInterested()
     protected void onFillInterestedFailed(Throwable cause)
     {
         if (LOG.isDebugEnabled())
-            LOG.debug("{} onFillInterestedFailed {}", this, cause);
+            LOG.debug("onFillInterestedFailed {}", this, cause);
         if (_endPoint.isOpen())
         {
             boolean close = true;

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/HttpConnectionFactory.java

@@ -15,14 +15,11 @@
 
 import java.util.Objects;
 
-import org.eclipse.jetty.http.ComplianceViolation;
 import org.eclipse.jetty.http.HttpVersion;
 import org.eclipse.jetty.io.Connection;
 import org.eclipse.jetty.io.EndPoint;
 import org.eclipse.jetty.server.internal.HttpConnection;
 import org.eclipse.jetty.util.annotation.Name;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * A Connection Factory for HTTP Connections.
@@ -32,10 +29,7 @@
  */
 public class HttpConnectionFactory extends AbstractConnectionFactory implements HttpConfiguration.ConnectionFactory
 {
-    private static final Logger LOG = LoggerFactory.getLogger(HttpConnectionFactory.class);
     private final HttpConfiguration _config;
-    private boolean _useInputDirectByteBuffers;
-    private boolean _useOutputDirectByteBuffers;
 
     public HttpConnectionFactory()
     {
@@ -57,45 +51,26 @@ public HttpConfiguration getHttpConfiguration()
         return _config;
     }
 
-    /**
-     * @deprecated use {@link HttpConfiguration#getComplianceViolationListeners()} instead to know if there
-     * are any {@link ComplianceViolation.Listener} to notify.  this method will be removed in Jetty 12.1.0
-     */
-    @Deprecated(since = "12.0.6", forRemoval = true)
-    public boolean isRecordHttpComplianceViolations()
-    {
-        return !_config.getComplianceViolationListeners().isEmpty();
-    }
-
-    /**
-     * Does nothing.
-     * @deprecated use {@link HttpConfiguration#addComplianceViolationListener(ComplianceViolation.Listener)} instead.
-     * this method will be removed in Jetty 12.1.0
-     */
-    @Deprecated(since = "12.0.6", forRemoval = true)
-    public void setRecordHttpComplianceViolations(boolean recordHttpComplianceViolations)
-    {
-        _config.addComplianceViolationListener(new ComplianceViolation.LoggingListener());
-    }
-
     public boolean isUseInputDirectByteBuffers()
     {
-        return _useInputDirectByteBuffers;
+        return _config.isUseInputDirectByteBuffers();
     }
 
+    @Deprecated(forRemoval = true, since = "12.1.0")
     public void setUseInputDirectByteBuffers(boolean useInputDirectByteBuffers)
     {
-        _useInputDirectByteBuffers = useInputDirectByteBuffers;
+        _config.setUseInputDirectByteBuffers(useInputDirectByteBuffers);
     }
 
     public boolean isUseOutputDirectByteBuffers()
     {
-        return _useOutputDirectByteBuffers;
+        return _config.isUseOutputDirectByteBuffers();
     }
 
+    @Deprecated(forRemoval = true, since = "12.1.0")
     public void setUseOutputDirectByteBuffers(boolean useOutputDirectByteBuffers)
     {
-        _useOutputDirectByteBuffers = useOutputDirectByteBuffers;
+        _config.setUseOutputDirectByteBuffers(useOutputDirectByteBuffers);
     }
 
     @Override

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/DelayedHandler.java

@@ -17,14 +17,14 @@
 import java.nio.charset.StandardCharsets;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicReference;
 
 import org.eclipse.jetty.http.HttpField;
 import org.eclipse.jetty.http.HttpHeader;
 import org.eclipse.jetty.http.HttpHeaderValue;
 import org.eclipse.jetty.http.HttpStatus;
 import org.eclipse.jetty.http.MimeTypes;
-import org.eclipse.jetty.io.Content;
+import org.eclipse.jetty.http.MultiPartConfig;
+import org.eclipse.jetty.http.MultiPartFormData;
 import org.eclipse.jetty.server.FormFields;
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.Request;
@@ -100,19 +100,27 @@ protected DelayedProcess newDelayedProcess(boolean contentExpected, String conte
         if (!contentExpected)
             return null;
 
-        // if we are not configured to delay dispatch, then no delay
-        if (!request.getConnectionMetaData().getHttpConfiguration().isDelayDispatchUntilContent())
-            return null;
-
-        // If there is no known content type, then delay only until content is available
+        // if no mimeType, then no delay
         if (mimeType == null)
-            return new UntilContentDelayedProcess(handler, request, response, callback);
+            return null;
 
         // Otherwise, delay until a known content type is fully read; or if the type is not known then until the content is available
         return switch (mimeType)
         {
             case FORM_ENCODED -> new UntilFormDelayedProcess(handler, request, response, callback, contentType);
-            default -> new UntilContentDelayedProcess(handler, request, response, callback);
+            case MULTIPART_FORM_DATA ->
+            {
+                MultiPartConfig config;
+                if (request.getContext().getAttribute(MultiPartConfig.class.getName()) instanceof MultiPartConfig mpc)
+                    config = mpc;
+                else if (getHandler().getServer().getAttribute(MultiPartConfig.class.getName()) instanceof MultiPartConfig mpc)
+                    config = mpc;
+                else
+                    yield null;
+
+                yield new UntilMultipartDelayedProcess(handler, request, response, callback, contentType, config);
+            }
+            default -> null;
         };
     }
 
@@ -167,96 +175,82 @@ protected void process()
         protected abstract void delay() throws Exception;
     }
 
-    protected static class UntilContentDelayedProcess extends DelayedProcess
+    protected static class UntilFormDelayedProcess extends DelayedProcess
     {
-        public UntilContentDelayedProcess(Handler handler, Request request, Response response, Callback callback)
+        private final Charset _charset;
+
+        public UntilFormDelayedProcess(Handler handler, Request request, Response response, Callback callback, String contentType)
         {
             super(handler, request, response, callback);
+
+            String cs = MimeTypes.getCharsetFromContentType(contentType);
+            _charset = StringUtil.isEmpty(cs) ? StandardCharsets.UTF_8 : Charset.forName(cs);
         }
 
         @Override
         protected void delay()
         {
-            Content.Chunk chunk = super.getRequest().read();
-            if (chunk == null)
-            {
-                getRequest().demand(this::onContent);
-            }
-            else
-            {
-                RewindChunkRequest request = new RewindChunkRequest(getRequest(), chunk);
-                try
-                {
-                    getHandler().handle(request, getResponse(), getCallback());
-                }
-                catch (Throwable x)
-                {
-                    // Use the wrapped request so that the error handling can
-                    // consume the request content and release the already read chunk.
-                    Response.writeError(request, getResponse(), getCallback(), x);
-                }
-            }
+            CompletableFuture<Fields> futureFormFields = FormFields.from(getRequest(), _charset);
+
+            // if we are done already, then we are still in the scope of the original process call and can
+            // process directly, otherwise we must execute a call to process as we are within a serialized
+            // demand callback.
+            futureFormFields.whenComplete(futureFormFields.isDone() ? this::process : this::executeProcess);
         }
 
-        public void onContent()
+        private void process(Fields fields, Throwable x)
         {
-            // We must execute here, because demand callbacks are serialized and process may block on a demand callback
-            getRequest().getContext().execute(this::process);
+            if (x == null)
+                super.process();
+            else
+                Response.writeError(getRequest(), getResponse(), getCallback(), x);
         }
 
-        private static class RewindChunkRequest extends Request.Wrapper
+        private void executeProcess(Fields fields, Throwable x)
         {
-            private final AtomicReference<Content.Chunk> _chunk;
-
-            public RewindChunkRequest(Request wrapped, Content.Chunk chunk)
-            {
-                super(wrapped);
-                _chunk = new AtomicReference<>(chunk);
-            }
-
-            @Override
-            public Content.Chunk read()
-            {
-                Content.Chunk chunk = _chunk.getAndSet(null);
-                if (chunk != null)
-                    return chunk;
-                return super.read();
-            }
+            if (x == null)
+                // We must execute here as even though we have consumed all the input, we are probably
+                // invoked in a demand runnable that is serialized with any write callbacks that might be done in process
+                getRequest().getContext().execute(super::process);
+            else
+                Response.writeError(getRequest(), getResponse(), getCallback(), x);
         }
     }
 
-    protected static class UntilFormDelayedProcess extends DelayedProcess
+    protected static class UntilMultipartDelayedProcess extends DelayedProcess
     {
-        private final Charset _charset;
+        private final String _contentType;
+        private final MultiPartConfig _config;
 
-        public UntilFormDelayedProcess(Handler handler, Request wrapped, Response response, Callback callback, String contentType)
+        public UntilMultipartDelayedProcess(Handler handler, Request request, Response response, Callback callback, String contentType, MultiPartConfig config)
         {
-            super(handler, wrapped, response, callback);
-
-            String cs = MimeTypes.getCharsetFromContentType(contentType);
-            _charset = StringUtil.isEmpty(cs) ? StandardCharsets.UTF_8 : Charset.forName(cs);
+            super(handler, request, response, callback);
+            _contentType = contentType;
+            _config = config;
         }
 
         @Override
         protected void delay()
         {
-            CompletableFuture<Fields> futureFormFields = FormFields.from(getRequest(), _charset);
+            Request request = getRequest();
+
+            CompletableFuture<MultiPartFormData.Parts> futureMultiPart = MultiPartFormData.from(request, request, _contentType, _config);
 
             // if we are done already, then we are still in the scope of the original process call and can
             // process directly, otherwise we must execute a call to process as we are within a serialized
             // demand callback.
-            futureFormFields.whenComplete(futureFormFields.isDone() ? this::process : this::executeProcess);
+            futureMultiPart.whenComplete(futureMultiPart.isDone() ? this::process : this::executeProcess);
         }
 
-        private void process(Fields fields, Throwable x)
+        private void process(MultiPartFormData.Parts parts, Throwable failure)
         {
-            if (x == null)
+            if (failure == null)
                 super.process();
             else
-                Response.writeError(getRequest(), getResponse(), getCallback(), x);
+                Response.writeError(getRequest(), getResponse(), getCallback(), failure);
         }
 
-        private void executeProcess(Fields fields, Throwable x)
+        private void executeProcess(MultiPartFormData.Parts parts, Throwable x)
         {
             if (x == null)
                 // We must execute here as even though we have consumed all the input, we are probably

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpChannelState.java

@@ -690,18 +690,22 @@ public void run()
         @Override
         public void succeeded()
         {
-            HttpStream stream;
-            boolean completeStream;
+            HttpStream completeStream = null;
+            Throwable failure = null;
             try (AutoLock ignored = _lock.lock())
             {
                 assert _callbackCompleted;
                 _streamSendState = StreamSendState.LAST_COMPLETE;
-                completeStream = _handling == null;
-                stream = _stream;
+                if (_handling == null)
+                {
+                    completeStream = _stream;
+                    _stream = null;
+                    failure = _callbackFailure;
+                }
             }
 
-            if (completeStream)
-                completeStream(stream, null);
+            if (completeStream != null)
+                completeStream(completeStream, failure);
         }
 
         /**