Merge branch 'master' into javawebsocket-jsr

jetty-client/src/main/java/org/eclipse/jetty/client/util/DeferredContentProvider.java

@@ -23,6 +23,7 @@ import java.util.Iterator;
 import java.util.NoSuchElementException;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.eclipse.jetty.client.AsyncContentProvider;
@@ -34,6 +35,10 @@ import org.eclipse.jetty.client.api.Response;
  * A {@link ContentProvider} that allows to add content after {@link Request#send(Response.CompleteListener)}
  * has been called, therefore providing the request content at a later time.
  * <p />
+ * {@link DeferredContentProvider} can only be used in conjunction with
+ * {@link Request#send(Response.CompleteListener)} (and not with its blocking counterpart {@link Request#send()})
+ * because it provides content asynchronously.
+ * <p />
  * The deferred content is provided once and then fully consumed.
  * Invocations to the {@link #iterator()} method after the first will return an "empty" iterator
  * because the stream has been consumed on the first invocation.
@@ -79,6 +84,7 @@ public class DeferredContentProvider implements AsyncContentProvider, AutoClosea
     private final Queue<ByteBuffer> chunks = new ConcurrentLinkedQueue<>();
     private final AtomicReference<Listener> listener = new AtomicReference<>();
     private final Iterator<ByteBuffer> iterator = new DeferredContentProviderIterator();
+    private final AtomicBoolean closed = new AtomicBoolean();
 
     /**
      * Creates a new {@link DeferredContentProvider} with the given initial content
@@ -124,8 +130,11 @@ public class DeferredContentProvider implements AsyncContentProvider, AutoClosea
      */
     public void close()
     {
-        chunks.offer(CLOSE);
+        if (closed.compareAndSet(false, true))
-        notifyListener();
+        {
+            chunks.offer(CLOSE);
+            notifyListener();
+        }
     }
 
     private void notifyListener()

jetty-client/src/main/java/org/eclipse/jetty/client/util/OutputStreamContentProvider.java

@@ -0,0 +1,132 @@
+//
+//  ========================================================================
+//  Copyright (c) 1995-2013 Mort Bay Consulting Pty. Ltd.
+//  ------------------------------------------------------------------------
+//  All rights reserved. This program and the accompanying materials
+//  are made available under the terms of the Eclipse Public License v1.0
+//  and Apache License v2.0 which accompanies this distribution.
+//
+//      The Eclipse Public License is available at
+//      http://www.eclipse.org/legal/epl-v10.html
+//
+//      The Apache License v2.0 is available at
+//      http://www.opensource.org/licenses/apache2.0.php
+//
+//  You may elect to redistribute this code under either of these licenses.
+//  ========================================================================
+//
+
+package org.eclipse.jetty.client.util;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Iterator;
+
+import org.eclipse.jetty.client.AsyncContentProvider;
+import org.eclipse.jetty.client.api.ContentProvider;
+import org.eclipse.jetty.client.api.Request;
+import org.eclipse.jetty.client.api.Response;
+
+/**
+ * A {@link ContentProvider} that provides content asynchronously through an {@link OutputStream}
+ * similar to {@link DeferredContentProvider}.
+ * <p />
+ * {@link OutputStreamContentProvider} can only be used in conjunction with
+ * {@link Request#send(Response.CompleteListener)} (and not with its blocking counterpart {@link Request#send()})
+ * because it provides content asynchronously.
+ * <p />
+ * The deferred content is provided once by writing to the {@link #getOutputStream() output stream}
+ * and then fully consumed.
+ * Invocations to the {@link #iterator()} method after the first will return an "empty" iterator
+ * because the stream has been consumed on the first invocation.
+ * However, it is possible for subclasses to support multiple invocations of {@link #iterator()}
+ * by overriding {@link #write(ByteBuffer)} and {@link #close()}, copying the bytes and making them
+ * available for subsequent invocations.
+ * <p />
+ * Content must be provided by writing to the {@link #getOutputStream() output stream}, that must be
+ * {@link OutputStream#close() closed} when all content has been provided.
+ * <p />
+ * Example usage:
+ * <pre>
+ * HttpClient httpClient = ...;
+ *
+ * // Use try-with-resources to autoclose the output stream
+ * OutputStreamContentProvider content = new OutputStreamContentProvider();
+ * try (OutputStream output = content.getOutputStream())
+ * {
+ *     httpClient.newRequest("localhost", 8080)
+ *             .content(content)
+ *             .send(new Response.CompleteListener()
+ *             {
+ *                 &#64Override
+ *                 public void onComplete(Result result)
+ *                 {
+ *                     // Your logic here
+ *                 }
+ *             });
+ *
+ *     // At a later time...
+ *     output.write("some content".getBytes());
+ * }
+ * </pre>
+ */
+public class OutputStreamContentProvider implements AsyncContentProvider
+{
+    private final DeferredContentProvider deferred = new DeferredContentProvider();
+    private final OutputStream output = new DeferredOutputStream();
+
+    @Override
+    public long getLength()
+    {
+        return deferred.getLength();
+    }
+
+    @Override
+    public Iterator<ByteBuffer> iterator()
+    {
+        return deferred.iterator();
+    }
+
+    @Override
+    public void setListener(Listener listener)
+    {
+        deferred.setListener(listener);
+    }
+
+    public OutputStream getOutputStream()
+    {
+        return output;
+    }
+
+    protected void write(ByteBuffer buffer)
+    {
+        deferred.offer(buffer);
+    }
+
+    protected void close()
+    {
+        deferred.close();
+    }
+
+    private class DeferredOutputStream extends OutputStream
+    {
+        @Override
+        public void write(int b) throws IOException
+        {
+            write(new byte[]{(byte)b}, 0, 1);
+        }
+
+        @Override
+        public void write(byte[] b, int off, int len) throws IOException
+        {
+            OutputStreamContentProvider.this.write(ByteBuffer.wrap(b, off, len));
+        }
+
+        @Override
+        public void close() throws IOException
+        {
+            OutputStreamContentProvider.this.close();
+        }
+    }
+}

jetty-client/src/test/java/org/eclipse/jetty/client/HttpClientStreamTest.java

@@ -47,6 +47,7 @@ import org.eclipse.jetty.client.api.Result;
 import org.eclipse.jetty.client.util.BufferingResponseListener;
 import org.eclipse.jetty.client.util.DeferredContentProvider;
 import org.eclipse.jetty.client.util.InputStreamResponseListener;
+import org.eclipse.jetty.client.util.OutputStreamContentProvider;
 import org.eclipse.jetty.server.handler.AbstractHandler;
 import org.eclipse.jetty.toolchain.test.MavenTestingUtils;
 import org.eclipse.jetty.toolchain.test.annotation.Slow;
@@ -366,6 +367,7 @@ public class HttpClientStreamTest extends AbstractHttpClientServerTest
                         }
                     });
 
+            // Make sure we provide the content *after* the request has been "sent".
             Thread.sleep(1000);
 
             try (ByteArrayInputStream input = new ByteArrayInputStream(new byte[1024]))
@@ -505,4 +507,46 @@ public class HttpClientStreamTest extends AbstractHttpClientServerTest
 
         Assert.assertTrue(latch.await(5, TimeUnit.SECONDS));
     }
+
+    @Test
+    public void testUploadWithOutputStream() throws Exception
+    {
+        start(new AbstractHandler()
+        {
+            @Override
+            public void handle(String target, org.eclipse.jetty.server.Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
+            {
+                baseRequest.setHandled(true);
+                IO.copy(request.getInputStream(), response.getOutputStream());
+            }
+        });
+
+        final byte[] data = new byte[512];
+        final CountDownLatch latch = new CountDownLatch(1);
+        OutputStreamContentProvider content = new OutputStreamContentProvider();
+        client.newRequest("localhost", connector.getLocalPort())
+                .scheme(scheme)
+                .content(content)
+                .send(new BufferingResponseListener()
+                {
+                    @Override
+                    public void onComplete(Result result)
+                    {
+                        if (result.isSucceeded() &&
+                                result.getResponse().getStatus() == 200 &&
+                                Arrays.equals(data, getContent()))
+                            latch.countDown();
+                    }
+                });
+
+        // Make sure we provide the content *after* the request has been "sent".
+        Thread.sleep(1000);
+
+        try (OutputStream output = content.getOutputStream())
+        {
+            output.write(data);
+        }
+
+        Assert.assertTrue(latch.await(5, TimeUnit.SECONDS));
+    }
 }

jetty-client/src/test/java/org/eclipse/jetty/client/api/Usage.java

@@ -20,6 +20,7 @@ package org.eclipse.jetty.client.api;
 
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
+import java.io.OutputStream;
 import java.net.HttpCookie;
 import java.net.URI;
 import java.nio.ByteBuffer;
@@ -35,6 +36,7 @@ import org.eclipse.jetty.client.util.DeferredContentProvider;
 import org.eclipse.jetty.client.util.FutureResponseListener;
 import org.eclipse.jetty.client.util.InputStreamContentProvider;
 import org.eclipse.jetty.client.util.InputStreamResponseListener;
+import org.eclipse.jetty.client.util.OutputStreamContentProvider;
 import org.eclipse.jetty.http.HttpMethod;
 import org.eclipse.jetty.http.HttpVersion;
 import org.junit.Assert;
@@ -274,6 +276,33 @@ public class Usage
         Assert.assertEquals(200, response.getStatus());
     }
 
+    @Test
+    public void testRequestOutputStream() throws Exception
+    {
+        HttpClient client = new HttpClient();
+        client.start();
+
+        OutputStreamContentProvider content = new OutputStreamContentProvider();
+        try (OutputStream output = content.getOutputStream())
+        {
+            client.newRequest("localhost", 8080)
+                    .content(content)
+                    .send(new Response.CompleteListener()
+                    {
+                        @Override
+                        public void onComplete(Result result)
+                        {
+                            Assert.assertEquals(200, result.getResponse().getStatus());
+                        }
+                    });
+
+            output.write(new byte[1024]);
+            output.write(new byte[512]);
+            output.write(new byte[256]);
+            output.write(new byte[128]);
+        }
+    }
+
     @Test
     public void testProxyUsage() throws Exception
     {

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

@@ -21,7 +21,6 @@ package org.eclipse.jetty.server;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-
 import javax.servlet.AsyncContext;
 import javax.servlet.AsyncEvent;
 import javax.servlet.AsyncListener;
@@ -39,8 +38,8 @@ import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
 import org.eclipse.jetty.util.thread.Scheduler;
 
-/* ------------------------------------------------------------ */
+/**
-/** Implementation of AsyncContext interface that holds the state of request-response cycle.
+ * Implementation of AsyncContext interface that holds the state of request-response cycle.
  *
  * <table>
  * <tr><th>STATE</th><th colspan=6>ACTION</th></tr>
@@ -56,7 +55,6 @@ import org.eclipse.jetty.util.thread.Scheduler;
  * <tr><th align=right>COMPLETING:</th>    <td>COMPLETING</td>  <td></td>            <td></td>            <td></td>             <td></td>                <td>COMPLETED</td></tr>
  * <tr><th align=right>COMPLETED:</th>     <td></td>            <td></td>            <td></td>            <td></td>             <td></td>                <td></td></tr>
  * </table>
- *
  */
 public class HttpChannelState implements AsyncContext
 {
@@ -78,12 +76,10 @@ public class HttpChannelState implements AsyncContext
         COMPLETED      // Request is complete
     }
 
-    /* ------------------------------------------------------------ */
     private final HttpChannel<?> _channel;
     private List<AsyncListener> _lastAsyncListeners;
     private List<AsyncListener> _asyncListeners;
 
-    /* ------------------------------------------------------------ */
     private State _state;
     private boolean _initial;
     private boolean _dispatched;
@@ -92,7 +88,6 @@ public class HttpChannelState implements AsyncContext
     private long _timeoutMs=DEFAULT_TIMEOUT;
     private AsyncEventState _event;
 
-    /* ------------------------------------------------------------ */
     protected HttpChannelState(HttpChannel<?> channel)
     {
         _channel=channel;
@@ -100,7 +95,6 @@ public class HttpChannelState implements AsyncContext
         _initial=true;
     }
 
-    /* ------------------------------------------------------------ */
     public State getState()
     {
         synchronized(this)
@@ -109,7 +103,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public void addListener(AsyncListener listener)
     {
@@ -121,7 +114,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public void addListener(AsyncListener listener,ServletRequest request, ServletResponse response)
     {
@@ -134,7 +126,6 @@ public class HttpChannelState implements AsyncContext
     }
 
 
-    /* ------------------------------------------------------------ */
     @Override
     public void setTimeout(long ms)
     {
@@ -144,7 +135,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public long getTimeout()
     {
@@ -154,7 +144,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     public AsyncEventState getAsyncEventState()
     {
         synchronized(this)
@@ -163,7 +152,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public String toString()
     {
@@ -173,7 +161,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     public String getStatusString()
     {
         synchronized (this)
@@ -185,7 +172,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     /**
      * @return true if the handling of the request should proceed
      */
@@ -228,10 +214,10 @@ public class HttpChannelState implements AsyncContext
 
             _responseWrapped=false;
             return true;
-            
+
         }
     }
-    /* ------------------------------------------------------------ */
+
     public void startAsync()
     {
         synchronized (this)
@@ -272,11 +258,7 @@ public class HttpChannelState implements AsyncContext
             }
         }
     }
-    
+
-    /* ------------------------------------------------------------ */
-    /* (non-Javadoc)
-     * @see javax.servlet.ServletRequest#suspend(long)
-     */
     public void startAsync(final ServletContext context,final ServletRequest request,final ServletResponse response)
     {
         synchronized (this)
@@ -319,7 +301,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     protected void error(Throwable th)
     {
         synchronized (this)
@@ -329,7 +310,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     /**
      * Signal that the HttpConnection has finished handling the request.
      * For blocking connectors, this call may block if the request has
@@ -382,11 +362,10 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public void dispatch()
     {
-        boolean dispatch=false;
+        boolean dispatch;
         synchronized (this)
         {
             switch(_state)
@@ -417,10 +396,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
-    /**
-     * @see Continuation#isDispatched()
-     */
     public boolean isDispatched()
     {
         synchronized (this)
@@ -428,8 +403,7 @@ public class HttpChannelState implements AsyncContext
             return _dispatched;
         }
     }
-    
+
-    /* ------------------------------------------------------------ */
     protected void expired()
     {
         final List<AsyncListener> aListeners;
@@ -442,7 +416,6 @@ public class HttpChannelState implements AsyncContext
                     aListeners=_asyncListeners;
                     break;
                 default:
-                    aListeners=null;
                     return;
             }
             _expired=true;
@@ -463,29 +436,31 @@ public class HttpChannelState implements AsyncContext
             }
         }
 
-
+        boolean complete;
         synchronized (this)
         {
             switch(_state)
             {
                 case ASYNCSTARTED:
                 case ASYNCWAIT:
-                    complete();
+                    complete = true;
+                    break;
+                default:
+                    complete = false;
+                    break;
             }
         }
+        if (complete)
+            complete();
 
         scheduleDispatch();
     }
 
-    /* ------------------------------------------------------------ */
-    /* (non-Javadoc)
-     * @see javax.servlet.ServletRequest#complete()
-     */
     @Override
     public void complete()
     {
         // just like resume, except don't set _dispatched=true;
-        boolean dispatch=false;
+        boolean dispatch;
         synchronized (this)
         {
             switch(_state)
@@ -516,13 +491,11 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public <T extends AsyncListener> T createListener(Class<T> clazz) throws ServletException
     {
         try
         {
-            // TODO inject
             return clazz.newInstance();
         }
         catch(Exception e)
@@ -531,11 +504,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-
-    /* ------------------------------------------------------------ */
-    /* (non-Javadoc)
-     * @see javax.servlet.ServletRequest#complete()
-     */
     protected void completed()
     {
         final List<AsyncListener> aListeners;
@@ -549,7 +517,6 @@ public class HttpChannelState implements AsyncContext
                     break;
 
                 default:
-                    aListeners=null;
                     throw new IllegalStateException(this.getStatusString());
             }
         }
@@ -577,7 +544,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     protected void recycle()
     {
         synchronized (this)
@@ -600,7 +566,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     public void cancel()
     {
         synchronized (this)
@@ -609,13 +574,11 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     protected void scheduleDispatch()
     {
         _channel.execute(_channel);
     }
 
-    /* ------------------------------------------------------------ */
     protected void scheduleTimeout()
     {
         Scheduler scheduler = _channel.getScheduler();
@@ -623,7 +586,6 @@ public class HttpChannelState implements AsyncContext
             _event._timeout=scheduler.schedule(new AsyncTimeout(),_timeoutMs,TimeUnit.MILLISECONDS);
     }
 
-    /* ------------------------------------------------------------ */
     protected void cancelTimeout()
     {
         AsyncEventState event=_event;
@@ -635,10 +597,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
-    /* (non-Javadoc)
-     * @see javax.servlet.ServletRequest#isInitial()
-     */
     public boolean isInitial()
     {
         synchronized(this)
@@ -647,7 +605,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     public boolean isSuspended()
     {
         synchronized(this)
@@ -666,7 +623,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     boolean isCompleting()
     {
         synchronized (this)
@@ -675,13 +631,12 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     public boolean isAsync()
     {
         synchronized (this)
         {
             switch(_state)
-            {  
+            {
                 case ASYNCSTARTED:
                 case REDISPATCHING:
                 case ASYNCWAIT:
@@ -696,7 +651,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public void dispatch(ServletContext context, String path)
     {
@@ -705,7 +659,6 @@ public class HttpChannelState implements AsyncContext
         dispatch();
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public void dispatch(String path)
     {
@@ -713,13 +666,11 @@ public class HttpChannelState implements AsyncContext
         dispatch();
     }
 
-    /* ------------------------------------------------------------ */
     public Request getBaseRequest()
     {
         return _channel.getRequest();
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public ServletRequest getRequest()
     {
@@ -728,7 +679,6 @@ public class HttpChannelState implements AsyncContext
         return _channel.getRequest();
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public ServletResponse getResponse()
     {
@@ -737,7 +687,6 @@ public class HttpChannelState implements AsyncContext
         return _channel.getResponse();
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public void start(final Runnable run)
     {
@@ -755,7 +704,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     @Override
     public boolean hasOriginalRequestAndResponse()
     {
@@ -765,7 +713,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
     public ContextHandler getContextHandler()
     {
         final AsyncEventState event=_event;
@@ -774,11 +721,6 @@ public class HttpChannelState implements AsyncContext
         return null;
     }
 
-
-    /* ------------------------------------------------------------ */
-    /**
-     * @see org.eclipse.jetty.continuation.Continuation#getServletResponse()
-     */
     public ServletResponse getServletResponse()
     {
         if (_responseWrapped && _event!=null && _event.getSuppliedResponse()!=null)
@@ -786,35 +728,21 @@ public class HttpChannelState implements AsyncContext
         return _channel.getResponse();
     }
 
-    /* ------------------------------------------------------------ */
-    /**
-     * @see org.eclipse.jetty.continuation.Continuation#getAttribute(java.lang.String)
-     */
     public Object getAttribute(String name)
     {
         return _channel.getRequest().getAttribute(name);
     }
 
-    /* ------------------------------------------------------------ */
-    /**
-     * @see org.eclipse.jetty.continuation.Continuation#removeAttribute(java.lang.String)
-     */
     public void removeAttribute(String name)
     {
         _channel.getRequest().removeAttribute(name);
     }
 
-    /* ------------------------------------------------------------ */
-    /**
-     * @see org.eclipse.jetty.continuation.Continuation#setAttribute(java.lang.String, java.lang.Object)
-     */
     public void setAttribute(String name, Object attribute)
     {
         _channel.getRequest().setAttribute(name,attribute);
     }
 
-    /* ------------------------------------------------------------ */
-    /* ------------------------------------------------------------ */
     public class AsyncTimeout implements Runnable
     {
         @Override
@@ -824,8 +752,6 @@ public class HttpChannelState implements AsyncContext
         }
     }
 
-    /* ------------------------------------------------------------ */
-    /* ------------------------------------------------------------ */
     public class AsyncEventState extends AsyncEvent
     {
         final private ServletContext _suspendedContext;
@@ -838,7 +764,7 @@ public class HttpChannelState implements AsyncContext
         {
             super(HttpChannelState.this, request,response);
             _suspendedContext=context;
-            
+
             // Get the base request So we can remember the initial paths
             Request r=_channel.getRequest();
 
@@ -884,7 +810,6 @@ public class HttpChannelState implements AsyncContext
             return _dispatchContext==null?_suspendedContext:_dispatchContext;
         }
 
-        /* ------------------------------------------------------------ */
         /**
          * @return The path in the context
          */

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

@@ -23,7 +23,6 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
-
 import javax.servlet.RequestDispatcher;
 import javax.servlet.ServletOutputStream;
 import javax.servlet.ServletRequest;
@@ -31,7 +30,6 @@ import javax.servlet.ServletResponse;
 
 import org.eclipse.jetty.http.HttpContent;
 import org.eclipse.jetty.http.HttpHeader;
-import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.log.Log;
 import org.eclipse.jetty.util.log.Logger;
@@ -78,31 +76,27 @@ public class HttpOutput extends ServletOutputStream
     public void reset()
     {
         _written = 0;
-        _closed = false;
+        reopen();
     }
 
     public void reopen()
     {
         _closed = false;
     }
-    
+
-    /** Called by the HttpChannel if the output was closed 
+    /** Called by the HttpChannel if the output was closed
      * externally (eg by a 500 exception handling).
      */
-    void closed() 
+    void closed()
     {
         _closed = true;
-        if (_aggregate != null)
+        releaseBuffer();
-        {
-            _channel.getConnector().getByteBufferPool().release(_aggregate);
-            _aggregate = null;
-        }
     }
 
     @Override
-    public void close() 
+    public void close()
     {
-        if (!_closed)
+        if (!isClosed())
         {
             try
             {
@@ -117,7 +111,11 @@ public class HttpOutput extends ServletOutputStream
                 LOG.ignore(e);
             }
         }
-        _closed = true;
+        closed();
+    }
+
+    private void releaseBuffer()
+    {
         if (_aggregate != null)
         {
             _channel.getConnector().getByteBufferPool().release(_aggregate);
@@ -133,8 +131,8 @@ public class HttpOutput extends ServletOutputStream
     @Override
     public void flush() throws IOException
     {
-        if (_closed)
+        if (isClosed())
-            throw new EofException();
+            return;
 
         if (BufferUtil.hasContent(_aggregate))
             _channel.write(_aggregate, false);
@@ -150,8 +148,8 @@ public class HttpOutput extends ServletOutputStream
     @Override
     public void write(byte[] b, int off, int len) throws IOException
     {
-        if (_closed)
+        if (isClosed())
-            throw new EOFException();
+            throw new EOFException("Closed");
 
         // Do we have an aggregate buffer already ?
         if (_aggregate == null)
@@ -201,16 +199,16 @@ public class HttpOutput extends ServletOutputStream
             _channel.write(_aggregate, false);
     }
 
-    
+
     @Override
     public void write(int b) throws IOException
     {
-        if (_closed)
+        if (isClosed())
-            throw new EOFException();
+            throw new EOFException("Closed");
 
         if (_aggregate == null)
             _aggregate = _channel.getByteBufferPool().acquire(getBufferSize(), OUTPUT_BUFFER_DIRECT);
-        
+
         BufferUtil.append(_aggregate, (byte)b);
         _written++;
 
@@ -224,6 +222,7 @@ public class HttpOutput extends ServletOutputStream
     {
         if (isClosed())
             throw new IOException("Closed");
+
         write(s.getBytes(_channel.getResponse().getCharacterEncoding()));
     }
 
@@ -256,7 +255,7 @@ public class HttpOutput extends ServletOutputStream
             String etag=httpContent.getETag();
             if (etag!=null)
                 response.getHttpFields().put(HttpHeader.ETAG,etag);
-            
+
             content = httpContent.getDirectBuffer();
             if (content == null)
                 content = httpContent.getIndirectBuffer();

jetty-server/src/test/java/org/eclipse/jetty/server/ResponseTest.java

@@ -18,12 +18,6 @@
 
 package org.eclipse.jetty.server;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.io.LineNumberReader;
@@ -35,8 +29,8 @@ import java.util.Collections;
 import java.util.Enumeration;
 import java.util.Iterator;
 import java.util.Locale;
-
 import javax.servlet.ServletException;
+import javax.servlet.ServletOutputStream;
 import javax.servlet.http.Cookie;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
@@ -61,32 +55,37 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
 public class ResponseTest
 {
     private Server _server;
-    private HttpChannel<?> _channel;
+    private HttpChannel<ByteBuffer> _channel;
-    private Scheduler _scheduler;
 
     @Before
     public void init() throws Exception
     {
         _server = new Server();
-        _scheduler = new TimerScheduler();
+        Scheduler _scheduler = new TimerScheduler();
         HttpConfiguration config = new HttpConfiguration();
-        LocalConnector connector = new LocalConnector(_server,null,_scheduler,null,1,new HttpConnectionFactory(config));
+        LocalConnector connector = new LocalConnector(_server,null, _scheduler,null,1,new HttpConnectionFactory(config));
         _server.addConnector(connector);
         _server.setHandler(new DumpHandler());
         _server.start();
 
         AbstractEndPoint endp = new ByteArrayEndPoint(_scheduler, 5000);
         ByteBufferHttpInput input = new ByteBufferHttpInput();
-        _channel = new HttpChannel<ByteBuffer>(connector, new HttpConfiguration(), endp, new HttpTransport()
+        _channel = new HttpChannel<>(connector, new HttpConfiguration(), endp, new HttpTransport()
         {
             @Override
             public void send(HttpGenerator.ResponseInfo info, ByteBuffer content, boolean lastContent) throws IOException
             {
             }
-            
+
             @Override
             public void send(ResponseInfo info, ByteBuffer content, boolean lastContent, Callback callback)
             {
@@ -435,7 +434,7 @@ public class ResponseTest
                 {"http://myhost:8888/other/location;jsessionid=12345?name=value","http://myhost:8888/other/location;jsessionid=12345?name=value"},
                 {"/other/location;jsessionid=12345?name=value","http://myhost:8888/other/location;jsessionid=12345?name=value"},
                 {"./location;jsessionid=12345?name=value","http://myhost:8888/path/location;jsessionid=12345?name=value"},
-                
+
                 // From cookie
                 {"/other/location","http://myhost:8888/other/location"},
                 {"/other/l%20cation", "http://myhost:8888/other/l%20cation"},
@@ -532,7 +531,7 @@ public class ResponseTest
             socket.getOutputStream().write("HEAD / HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes());
             socket.getOutputStream().write("GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n".getBytes());
             socket.getOutputStream().flush();
-                        
+
             LineNumberReader reader = new LineNumberReader(new InputStreamReader(socket.getInputStream()));
             String line = reader.readLine();
             Assert.assertThat(line, Matchers.startsWith("HTTP/1.1 200 OK"));
@@ -576,11 +575,11 @@ public class ResponseTest
 
         assertEquals("name=value;Path=/path;Domain=domain;Secure;HttpOnly;Comment=comment", set);
     }
-    
+
-    
+
     @Test
     public void testCookiesWithReset() throws Exception
-    { 
+    {
         Response response = newResponse();
 
         Cookie cookie=new Cookie("name","value");
@@ -589,14 +588,14 @@ public class ResponseTest
         cookie.setSecure(true);
         cookie.setComment("comment__HTTP_ONLY__");
         response.addCookie(cookie);
-        
+
         Cookie cookie2=new Cookie("name2", "value2");
         cookie2.setDomain("domain");
         cookie2.setPath("/path");
         response.addCookie(cookie2);
 
         //keep the cookies
-        response.reset(true);        
+        response.reset(true);
 
         Enumeration<String> set = response.getHttpFields().getValues("Set-Cookie");
 
@@ -605,14 +604,27 @@ public class ResponseTest
         assertEquals(2, list.size());
         assertTrue(list.contains("name=value;Path=/path;Domain=domain;Secure;HttpOnly;Comment=comment"));
         assertTrue(list.contains("name2=value2;Path=/path;Domain=domain"));
-        
+
         //get rid of the cookies
         response.reset();
-        
+
         set = response.getHttpFields().getValues("Set-Cookie");
         assertFalse(set.hasMoreElements());
     }
 
+    @Test
+    public void testFlushAfterFullContent() throws Exception
+    {
+        Response response = _channel.getResponse();
+        byte[] data = new byte[]{(byte)0xCA, (byte)0xFE};
+        ServletOutputStream output = response.getOutputStream();
+        response.setContentLength(data.length);
+        // Write the whole content
+        output.write(data);
+        // Must not throw
+        output.flush();
+    }
+
     private Response newResponse()
     {
         _channel.reset();