Merged branch 'jetty-9.4.x' into 'master'.

CONTRIBUTING.md

@@ -52,3 +52,13 @@ Create a new bug
 Be sure to search for existing bugs before you create another one. Remember that contributions are always welcome!
 
 - [https://github.com/eclipse/jetty.project/issues](https://github.com/eclipse/jetty.project/issues)
+
+Reporting Security Issues
+-----------------
+There are a number of avenues for reporting security issues to the Jetty project available. 
+If the issue is directly related to Jetty itself then reporting to the Jetty developers is encouraged. 
+The most direct method is to mail [security@webtide.com](mailto:security@webtide.com). 
+Webtide is comprised of the active committers of the Jetty project is our preferred reporting method. 
+We are flexible in how we work with reporters of security issues but we reserve the right to act in the interests of the Jetty project in all circumstances.
+
+If the issue is related to Eclipse or its Jetty integration then we encourage you to reach out to [security@eclipse.org](mailto:security@eclipse.org).

jetty-documentation/src/main/asciidoc/development/embedding/jetty-helloworld.adoc

@@ -23,8 +23,8 @@ This section provides a tutorial that shows how you can quickly develop embedded
 ==== Downloading the Jars
 
 Jetty is decomposed into many jars and dependencies to achieve a minimal footprint by selecting the minimal set of jars.
-Typically it is best to use something like Maven to manage jars, however this tutorial uses an aggregate Jar that contains all of the Jetty classes in one Jar.
-You can manually download the aggregate link:http://central.maven.org/maven2/org/eclipse/jetty/aggregate/jetty-all/{VERSION}/jetty-all-{VERSION}-uber.jar[`jetty-all.jar`] using `curl`) or a browser.
+Typically it is best to use something like link:#jetty-maven-helloworld[Maven] to manage jars, however this tutorial uses an aggregate Jar that contains all of the required Jetty classes in one Jar.
+You can manually download the aggregate link:http://central.maven.org/maven2/org/eclipse/jetty/aggregate/jetty-all/{VERSION}/jetty-all-{VERSION}-uber.jar[`jetty-all.jar`] using `curl` or a browser.
 
 ____
 [NOTE]
@@ -32,6 +32,12 @@ The central Maven repository has started to aggressively reject/deny access to t
 The administrators of the central maven repository have stated that the recommended command line download tool is now curl.
 ____
 
+____
+[IMPORTANT]
+The `jetty-all` jar referenced in this section is for example purposes only and should not be used outside of this context.
+Please consider using link:#jetty-maven-helloworld[Maven] to manage your project dependencies.
+____
+
 Use curl as follows:
 
 [source, screen, subs="{sub-order}"]

jetty-io/src/main/java/org/eclipse/jetty/io/FillInterest.java

@@ -88,8 +88,6 @@ public abstract class FillInterest
         
         try
         {
-            if (LOG.isDebugEnabled())
-                LOG.debug("{} register {}",this,callback);
             needsFillInterest();
         }
         catch (Throwable e)

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

@@ -281,7 +281,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
         {
             int filled = fillRequestBuffer();
             handled = parseRequestBuffer();
-            if (handled || filled<=0 || _channel.getRequest().getHttpInput().hasContent())
+            if (handled || filled<=0 || _input.hasContent())
                 break;
         }
         return handled;
@@ -395,7 +395,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
         else if (_parser.inContentState() && _generator.isPersistent())
         {
             // If we are async, then we have problems to complete neatly
-            if (_channel.getRequest().getHttpInput().isAsync())
+            if (_input.isAsync())
             {
                 if (LOG.isDebugEnabled())
                     LOG.debug("unconsumed async input {}", this);
@@ -406,7 +406,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
                 if (LOG.isDebugEnabled())
                     LOG.debug("unconsumed input {}", this);
                 // Complete reading the request
-                if (!_channel.getRequest().getHttpInput().consumeAll())
+                if (!_input.consumeAll())
                     _channel.abort(new IOException("unconsumed input"));
             }
         }
@@ -548,7 +548,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
 
     public void asyncReadFillInterested()
     {
-        getEndPoint().fillInterested(_asyncReadCallback);
+        getEndPoint().tryFillInterested(_asyncReadCallback);
     }
 
     public void blockingReadFillInterested()
@@ -624,7 +624,7 @@ public class HttpConnection extends AbstractConnection implements Runnable, Http
         {
             if (fillAndParseForContent())
                 _channel.handle();
-            else if (!_input.isFinished())
+            else if (!_input.isFinished() && !_input.hasContent())
                 asyncReadFillInterested();
         }
 

tests/test-http-client-transport/src/test/java/org/eclipse/jetty/http/client/AsyncIOServletTest.java

@@ -18,15 +18,21 @@
 
 package org.eclipse.jetty.http.client;
 
+import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.util.Deque;
 import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -49,6 +55,7 @@ import org.eclipse.jetty.client.api.Result;
 import org.eclipse.jetty.client.http.HttpConnectionOverHTTP;
 import org.eclipse.jetty.client.util.BufferingResponseListener;
 import org.eclipse.jetty.client.util.DeferredContentProvider;
+import org.eclipse.jetty.client.util.InputStreamContentProvider;
 import org.eclipse.jetty.client.util.StringContentProvider;
 import org.eclipse.jetty.http.HttpHeader;
 import org.eclipse.jetty.http.HttpMethod;
@@ -135,7 +142,7 @@ public class AsyncIOServletTest extends AbstractTest
         scope.set(null);
     }
 
-    private void sleep(long ms) throws IOException
+    private void sleep(long ms)
     {
         try
         {
@@ -143,7 +150,7 @@ public class AsyncIOServletTest extends AbstractTest
         }
         catch (InterruptedException e)
         {
-            throw new InterruptedIOException();
+            throw new UncheckedIOException(new InterruptedIOException());
         }
     }
 
@@ -1324,5 +1331,136 @@ public class AsyncIOServletTest extends AbstractTest
         
 
     }
-        
+
+    @Test
+    public void testWriteListenerFromOtherThread() throws Exception
+    {
+        start(new HttpServlet()
+        {
+            @Override
+            protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
+            {
+                AsyncContext asyncContext = request.startAsync();
+                asyncContext.setTimeout(0);
+                request.getInputStream().setReadListener(new Listener(asyncContext));
+            }
+        });
+
+        int cores = 4;
+        int iterations = 10;
+        CountDownLatch latch = new CountDownLatch(cores * iterations);
+        Deque<Throwable> failures = new LinkedBlockingDeque<>();
+        for (int i = 0; i < cores; ++i)
+        {
+            client.getExecutor().execute(() ->
+            {
+                for (int j = 0; j < iterations; ++j)
+                {
+                    try
+                    {
+                        ContentResponse response = client.newRequest(newURI())
+                                .method(HttpMethod.POST)
+                                .path(servletPath)
+                                .content(new InputStreamContentProvider(new ByteArrayInputStream(new byte[16 * 1024])
+                                {
+                                    @Override
+                                    public int read(byte[] b, int off, int len)
+                                    {
+                                        sleep(5);
+                                        return super.read(b, off, Math.min(len, 4242));
+                                    }
+                                }))
+                                .send();
+                        Assert.assertEquals(HttpStatus.OK_200, response.getStatus());
+                        latch.countDown();
+                    }
+                    catch (Throwable x)
+                    {
+                        failures.offer(x);
+                    }
+                }
+            });
+        }
+
+        Assert.assertTrue(latch.await(30, TimeUnit.SECONDS));
+        Assert.assertTrue(failures.isEmpty());
+    }
+
+    private class Listener implements ReadListener, WriteListener
+    {
+        private final Executor executor = Executors.newFixedThreadPool(32);
+        private final CompletableFuture<?> inputComplete = new CompletableFuture<>();
+        private final CompletableFuture<?> outputComplete = new CompletableFuture<>();
+        private final AtomicBoolean responseWritten = new AtomicBoolean();
+        private final AsyncContext asyncContext;
+        private final HttpServletResponse response;
+        private final ServletInputStream input;
+        private final ServletOutputStream output;
+
+        public Listener(AsyncContext asyncContext) throws IOException
+        {
+            this.asyncContext = asyncContext;
+            this.response = (HttpServletResponse)asyncContext.getResponse();
+            this.input = asyncContext.getRequest().getInputStream();
+            this.output = response.getOutputStream();
+            CompletableFuture.allOf(inputComplete, outputComplete)
+                    .whenComplete((ignoredResult, ignoredThrowable) -> asyncContext.complete());
+            // Dispatch setting the write listener to another thread.
+            executor.execute(() -> output.setWriteListener(this));
+        }
+
+        @Override
+        public void onDataAvailable() throws IOException
+        {
+            byte[] buffer = new byte[16 * 1024];
+            while (input.isReady())
+            {
+                if (input.read(buffer) < 0)
+                    return;
+            }
+        }
+
+        @Override
+        public void onAllDataRead() throws IOException
+        {
+            inputComplete.complete(null);
+        }
+
+        @Override
+        public void onWritePossible() throws IOException
+        {
+            // Dispatch OWP to another thread.
+            executor.execute(() ->
+            {
+                while (output.isReady())
+                {
+                    if (responseWritten.compareAndSet(false, true))
+                    {
+                        try
+                        {
+                            response.setStatus(HttpServletResponse.SC_OK);
+                            response.setContentType("text/plain;charset=utf-8");
+                            output.write("Hello world".getBytes());
+                        }
+                        catch (IOException x)
+                        {
+                            throw new UncheckedIOException(x);
+                        }
+                    }
+                    else
+                    {
+                        outputComplete.complete(null);
+                        return;
+                    }
+                }
+            });
+        }
+
+        @Override
+        public void onError(Throwable t)
+        {
+            response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+            asyncContext.complete();
+        }
+    }
 }