Experiment for aborting callbacks

jetty-core/jetty-client/src/main/java/org/eclipse/jetty/client/transport/HttpDestination.java

@@ -188,7 +188,7 @@ protected void doStart() throws Exception
     protected void doStop() throws Exception
     {
         requestTimeouts.destroy();
-        abort(new AsynchronousCloseException());
+        abortExchanges(new AsynchronousCloseException());
         Sweeper connectionPoolSweeper = client.getBean(Sweeper.class);
         if (connectionPoolSweeper != null && connectionPool instanceof Sweeper.Sweepable)
             connectionPoolSweeper.remove((Sweeper.Sweepable)connectionPool);
@@ -294,7 +294,7 @@ public void succeeded()
     @Override
     public void failed(Throwable x)
     {
-        abort(x);
+        abortExchanges(x);
     }
 
     @Override
@@ -513,7 +513,7 @@ public boolean remove(Connection connection)
      *
      * @param cause the abort cause
      */
-    public void abort(Throwable cause)
+    private void abortExchanges(Throwable cause)
     {
         // Copy the queue of exchanges and fail only those that are queued at this moment.
         // The application may queue another request from the failure/complete listener

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

@@ -18,7 +18,6 @@
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Response;
 import org.eclipse.jetty.util.Callback;
-import org.eclipse.jetty.util.thread.Invocable;
 
 public class ContextResponse extends Response.Wrapper
 {
@@ -33,24 +32,18 @@ public ContextResponse(ContextHandler.ScopedContext context, Request request, Re
     @Override
     public void write(boolean last, ByteBuffer content, Callback callback)
     {
-        Callback contextCallback = new Callback()
+        Callback contextCallback = new Callback.Nested(callback)
         {
             @Override
-            public void succeeded()
+            protected void onCompleteSuccess()
             {
                 _context.run(callback::succeeded, getRequest());
             }
 
             @Override
-            public void failed(Throwable x)
+            protected void onCompleteFailure(Throwable cause)
             {
-                _context.accept(callback::failed, x, getRequest());
-            }
-
-            @Override
-            public InvocationType getInvocationType()
-            {
-                return Invocable.getInvocationType(callback);
+                _context.accept(callback::failed, cause, getRequest());
             }
         };
         super.write(last, content, contextCallback);

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

@@ -1147,7 +1147,7 @@ private Runnable lockedFailWrite(Throwable x)
                 _writeFailure = x;
             else
                 ExceptionUtil.addSuppressedIfNotAssociated(_writeFailure, x);
-            return () -> HttpChannelState.failed(writeCallback, x);
+            return () -> HttpChannelState.cancel(writeCallback, x);
         }
 
         public long getContentBytesWritten()
@@ -1916,6 +1916,27 @@ public void onComplianceViolation(ComplianceViolation.Event event)
         }
     }
 
+    /**
+     * Invoke a callback cancel, handling any {@link Throwable} thrown
+     * by adding the passed {@code failure} as a suppressed with
+     * {@link ExceptionUtil#addSuppressedIfNotAssociated(Throwable, Throwable)}.
+     * @param callback The callback to fail
+     * @param failure The failure
+     * @throws RuntimeException If thrown, will have the {@code failure} added as a suppressed.
+     */
+    private static void cancel(Callback callback, Throwable failure)
+    {
+        try
+        {
+            callback.abort(failure);
+        }
+        catch (Throwable t)
+        {
+            ExceptionUtil.addSuppressedIfNotAssociated(t, failure);
+            throw t;
+        }
+    }
+
     /**
      * Invoke a callback failure, handling any {@link Throwable} thrown
      * by adding the passed {@code failure} as a suppressed with

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

@@ -1509,7 +1509,8 @@ public void succeeded()
             {
                 if (LOG.isDebugEnabled())
                     LOG.debug("abort due to pending read {} {} ", this, getEndPoint());
-                abort(new IOException("Pending read in onCompleted"));
+                Throwable failure = new IOException("Pending read in onCompleted");
+                getEndPoint().close(failure);
                 return;
             }
 
@@ -1603,12 +1604,7 @@ public void failed(Throwable x)
             }
             if (LOG.isDebugEnabled())
                 LOG.debug("aborting", x);
-            abort(x);
-        }
-
-        private void abort(Throwable failure)
-        {
-            getEndPoint().close(failure);
+            getEndPoint().close(x);
         }
 
         @Override