Fix #4461 HttpOutput Aggregation

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

@@ -378,6 +378,19 @@ private boolean updateApiState(Throwable failure)
         return wake;
     }
 
+    private int getAggregateSpace()
+    {
+        // If no aggregate, we can allocate one of bufferSize
+        if (_aggregate == null)
+            return getBufferSize();
+
+        // if the position is not zero, compact to avoid empty at capacity
+        if (_aggregate.position() != 0)
+            BufferUtil.compact(_aggregate);
+
+        return BufferUtil.space(_aggregate);
+    }
+
     public void softClose()
     {
         synchronized (_channelState)
@@ -723,6 +736,9 @@ private void checkWritable() throws EofException
     @Override
     public void write(byte[] b, int off, int len) throws IOException
     {
+        if (LOG.isDebugEnabled())
+            LOG.debug("write(array {})", BufferUtil.toDetailString(ByteBuffer.wrap(b, off, len)));
+
         boolean last;
         boolean aggregate;
         boolean flush;
@@ -733,7 +749,7 @@ public void write(byte[] b, int off, int len) throws IOException
         {
             checkWritable();
             long written = _written + len;
-            int space = _aggregate == null ? getBufferSize() : BufferUtil.space(_aggregate);
+            int space = getAggregateSpace();
             last = _channel.getResponse().isAllContentWritten(written);
             // Write will be aggregated if:
             //  + it is smaller than the commitSize
@@ -777,14 +793,23 @@ public void write(byte[] b, int off, int len) throws IOException
 
                 // return if we are not complete, not full and filled all the content
                 if (!flush)
+                {
+                    if (LOG.isDebugEnabled())
+                        LOG.debug("write(array) {} aggregated !flush {}",
+                            stateString(), BufferUtil.toDetailString(_aggregate));
                     return;
+                }
 
                 // adjust offset/length
                 off += filled;
                 len -= filled;
             }
         }
 
+        if (LOG.isDebugEnabled())
+            LOG.debug("write(array) {} last={} agg={} flush=true async={}, len={} {}",
+                stateString(), last, aggregate, async, len, BufferUtil.toDetailString(_aggregate));
+
         if (async)
         {
             // Do the asynchronous writing from the callback
@@ -801,9 +826,10 @@ public void write(byte[] b, int off, int len) throws IOException
                 channelWrite(_aggregate, last && len == 0);
 
                 // should we fill aggregate again from the buffer?
-                if (len > 0 && !last && len <= _commitSize && len <= BufferUtil.space(_aggregate))
+                if (len > 0 && !last && len <= _commitSize && len <= getAggregateSpace())
                 {
                     BufferUtil.append(_aggregate, b, off, len);
+                    onWriteComplete(false, null);
                     return;
                 }
             }
@@ -929,7 +955,7 @@ public void write(int b) throws IOException
         {
             checkWritable();
             long written = _written + 1;
-            int space = _aggregate == null ? getBufferSize() : BufferUtil.space(_aggregate);
+            int space = getAggregateSpace();
             last = _channel.getResponse().isAllContentWritten(written);
             flush = last || space == 1;
 
@@ -1602,7 +1628,7 @@ private class AsyncWrite extends ChannelWriteCB
         private final ByteBuffer _buffer;
         private final ByteBuffer _slice;
         private final int _len;
-        volatile boolean _completed;
+        private boolean _completed;
 
         AsyncWrite(byte[] b, int off, int len, boolean last)
         {
@@ -1639,7 +1665,7 @@ protected Action process() throws Exception
             }
 
             // Can we just aggregate the remainder?
-            if (!_last && _len < BufferUtil.space(_aggregate) && _len < _commitSize)
+            if (!_last && _aggregate != null && _len < getAggregateSpace() && _len < _commitSize)
             {
                 int position = BufferUtil.flipToFill(_aggregate);
                 BufferUtil.put(_buffer, _aggregate);
@@ -1670,15 +1696,6 @@ protected Action process() throws Exception
                 return Action.SCHEDULED;
             }
 
-            // all content written, but if we have not yet signal completion, we
-            // need to do so
-            if (_last && !_completed)
-            {
-                _completed = true;
-                channelWrite(BufferUtil.EMPTY_BUFFER, true, this);
-                return Action.SCHEDULED;
-            }
-
             if (LOG.isDebugEnabled() && _completed)
                 LOG.debug("EOF of {}", this);
 

jetty-server/src/main/java/org/eclipse/jetty/server/handler/gzip/GzipHttpOutputInterceptor.java

@@ -381,7 +381,7 @@ protected Action process() throws Exception
                         int len = slice.remaining();
                         _crc.update(array, off, len);
                         _deflater.setInput(array, off, len);  // TODO use ByteBuffer API in Jetty-10
-                        BufferUtil.clear(slice);
+                        slice.position(slice.position() + len);
                         if (_last && BufferUtil.isEmpty(_content))
                             _deflater.finish();
                     }