Upgraded to Netty 4

In addition, made the following changes to the way the handler works:

* Messages are buffered in a hash map by sequence ID if it's not the
  right time for the sequence.
* Removed subsequence ordering - writes for a given request are required
  to be sequential without the pipelining handler, so with it we can
  assume the same.
* Replaced buffer limit with back pressure - if the number of in flight
  requests exceeds a high water mark, the handler now exerts back
  pressure on the client by blocking read completed events, and unblocks
  them, and requests a new read, when a low watermark is reached.

Author: jroper

Diff for: README.md

@@ -4,12 +4,7 @@ netty-http-pipelining
 This library adds http pipelining capability to Netty.
 
 Inserting the HttpPipeliningHandler into a pipeline will cause message events containing an HttpRequest to become transformed
-into OrderedUpstreamMessageEvents. The OrderedUpstreamMessageEvent retains context such that a handler further upstream can
-compose it and reply with an OrderedDownstreamChannelEvent in any order and in parallel. The HttpPipeliningHandler will
+into SequencedHttpRequest. The SequencedHttpRequest retains context such that a handler further upstream can
+compose it and reply with an SequencedOutboundMessage in any order and in parallel. The HttpPipeliningHandler will
 ensure that http replies are sent back in the order that the http pipelining specification requires i.e. the order in which
 replies are returned must correlate to the order in which requests are made.
-
-The chunking of http replies is handled. Limits are also available within the handler to cap the buffering of replies
-in order to avoid memory exhaustion.
-
-Please refer to the HttpPipeliningHandlerTest for a comprehensive illustration of usage.

Diff for: pom.xml

@@ -41,8 +41,8 @@
     <dependencies>
         <dependency>
             <groupId>io.netty</groupId>
-            <artifactId>netty</artifactId>
-            <version>3.10.1.Final</version>
+            <artifactId>netty-codec-http</artifactId>
+            <version>4.0.29.Final</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>

Diff for: src/main/java/com/typesafe/netty/http/pipelining/HttpPipeliningHandler.java

@@ -1,110 +1,208 @@
 package com.typesafe.netty.http.pipelining;
 
-import org.jboss.netty.channel.*;
-import org.jboss.netty.handler.codec.http.HttpRequest;
+
+import io.netty.channel.*;
+import io.netty.handler.codec.http.*;
 
 import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Implements HTTP pipelining ordering, ensuring that responses are completely served in the same order as their
- * corresponding requests. NOTE: A side effect of using this handler is that upstream HttpRequest objects will
- * cause the original message event to be effectively transformed into an OrderedUpstreamMessageEvent. Conversely
- * OrderedDownstreamChannelEvent objects are expected to be received for the correlating response objects.
+ * corresponding requests.
+ *
+ * Each incoming request is assigned a sequence number, which is provided in the wrapping {@link SequencedHttpRequest}.
+ * When a handler writes a response and a response body, it must wrap each of those messages in a
+ * {@link SequencedOutboundMessage} with the corresponding sequence from that request.  It must send at least one
+ * message in response to the request, and the last, and only the last message it sends must have the last flag set.
+ *
+ * If messages are sent after the last message is sent, these messages may end up being buffered until the sequence
+ * number overflows and cycles back. If messages for a given sequence number are not sent sequentially, similar
+ * behaviour could also occur.
+ *
+ * There is no limit to the amount of messages that this handler will buffer for a particular sequence number.  It is
+ * the responsibility of the handler sending the outbound messages to handle back pressure via promises associated
+ * with each write event - if this is done, the buffering will be inherently bounded by back pressure.
  *
- * @author Christopher Hunt
+ * This handler does however put a bound on the maximum number of in flight requests that it will handle, configured by
+ * inFlightRequestsLowWatermark and inFlightRequestsHighWatermark. When the high watermark is exceeded, the handler will
+ * push back on the client. When the low watermark is reached, the handler will start reading again.
  */
-public class HttpPipeliningHandler extends SimpleChannelHandler {
+public class HttpPipeliningHandler extends ChannelDuplexHandler {
 
-    public static final int INITIAL_EVENTS_HELD = 3;
-    public static final int MAX_EVENTS_HELD = 10000;
+    /**
+     * The sequence of received HTTP requests.
+     */
+    volatile private int receiveSequence = 0;
 
-    private final int maxEventsHeld;
+    /**
+     * The currently sending sequence of HTTP requests.
+     */
+    volatile private int currentlySendingSequence = 1;
 
-    private int sequence;
-    private int nextRequiredSequence;
-    private int nextRequiredSubsequence;
+    /**
+     * Whether the high watermark has been exceeded.
+     */
+    volatile private boolean highWatermarkExceeded = false;
 
-    private final Queue<OrderedDownstreamChannelEvent> holdingQueue;
+    /**
+     * The number of requests in flight
+     */
+    private final AtomicInteger inFlight = new AtomicInteger();
 
-    public HttpPipeliningHandler() {
-        this(MAX_EVENTS_HELD);
+    /**
+     * A write message with a promise of when it's written.
+     */
+    private static class WriteMessage {
+        /**
+         * The written message.
+         */
+        final SequencedOutboundMessage message;
+
+        /**
+         * The future that is redeemed once the message is written.
+         */
+        final ChannelPromise promise;
+
+        public WriteMessage(SequencedOutboundMessage message, ChannelPromise promise) {
+            this.message = message;
+            this.promise = promise;
+        }
     }
 
     /**
-     * @param maxEventsHeld the maximum number of channel events that will be retained prior to aborting the channel
-     *                      connection. This is required as events cannot queue up indefintely; we would run out of
-     *                      memory if this was the case.
+     * The buffered events, by sequence
      */
-    public HttpPipeliningHandler(final int maxEventsHeld) {
-        this.maxEventsHeld = maxEventsHeld;
-
-        holdingQueue = new PriorityQueue<OrderedDownstreamChannelEvent>(INITIAL_EVENTS_HELD, new Comparator<OrderedDownstreamChannelEvent>() {
-            @Override
-            public int compare(OrderedDownstreamChannelEvent o1, OrderedDownstreamChannelEvent o2) {
-                final int delta = o1.getOrderedUpstreamMessageEvent().getSequence() - o2.getOrderedUpstreamMessageEvent().getSequence();
-                if (delta == 0) {
-                    return o1.getSubsequence() - o2.getSubsequence();
-                } else {
-                    return delta;
-                }
-            }
-        });
+    private final Map<Integer, List<WriteMessage>> bufferedEvents = new HashMap<>();
+
+    private final int inFlightRequestsLowWatermark;
+    private final int inFlightRequestsHighWatermark;
+
+    /**
+     * Create the pipelining handler with low and high watermarks of 2 and 4.
+     */
+    public HttpPipeliningHandler() {
+        this(2, 4);
     }
 
-    public int getMaxEventsHeld() {
-        return maxEventsHeld;
+    /**
+     * Create the pipelining handler.
+     *
+     * @param inFlightRequestsLowWatermark The soft limit for the number of in flight requests. If this limit is exceeded,
+     *                                the handler will start responding with 503 errors.  Must be at least 1.
+     * @param inFlightRequestsHighWatermark The hard limit for the number of in flight requests. If this limit is exceeded,
+     *                                the handler will close the connection. Must be at least
+     *                                {@code inFlightRequestsLowWatermark}.
+     */
+    public HttpPipeliningHandler(int inFlightRequestsLowWatermark, int inFlightRequestsHighWatermark) {
+        if (inFlightRequestsLowWatermark < 0) {
+            throw new IllegalArgumentException("inFlightRequestsLowWatermark must be an least 0, was " + inFlightRequestsLowWatermark);
+        }
+        if (inFlightRequestsHighWatermark <= inFlightRequestsLowWatermark) {
+            throw new IllegalArgumentException("inFlightRequestsHighWatermark must be greater than inFlightRequestsLowWatermark, but was " + inFlightRequestsHighWatermark);
+        }
+        this.inFlightRequestsLowWatermark = inFlightRequestsLowWatermark;
+        this.inFlightRequestsHighWatermark = inFlightRequestsHighWatermark;
     }
 
     @Override
-    public void messageReceived(final ChannelHandlerContext ctx, final MessageEvent e) {
-        final Object msg = e.getMessage();
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        // Only forward read complete if we haven't exceeded the high watermark
+        if (!highWatermarkExceeded) {
+            ctx.fireChannelReadComplete();
+        }    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         if (msg instanceof HttpRequest) {
-            ctx.sendUpstream(new OrderedUpstreamMessageEvent(sequence++, e.getChannel(), msg, e.getRemoteAddress()));
+            receiveSequence++;
+
+            if (inFlight.incrementAndGet() > inFlightRequestsHighWatermark) {
+                highWatermarkExceeded = true;
+            }
+            HttpRequest request = (HttpRequest) msg;
+            ctx.fireChannelRead(new SequencedHttpRequest(receiveSequence, request));
+
         } else {
-            ctx.sendUpstream(e);
+            ctx.fireChannelRead(msg);
         }
     }
 
     @Override
-    public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent e)
-            throws Exception {
-        if (e instanceof OrderedDownstreamChannelEvent) {
-
-            boolean channelShouldClose = false;
-
-            synchronized (holdingQueue) {
-                if (holdingQueue.size() < maxEventsHeld) {
-
-                    final OrderedDownstreamChannelEvent currentEvent = (OrderedDownstreamChannelEvent) e;
-                    holdingQueue.add(currentEvent);
-
-                    while (!holdingQueue.isEmpty()) {
-                        final OrderedDownstreamChannelEvent nextEvent = holdingQueue.peek();
-                        if (nextEvent.getOrderedUpstreamMessageEvent().getSequence() != nextRequiredSequence |
-                                nextEvent.getSubsequence() != nextRequiredSubsequence) {
-                            break;
-                        }
-                        holdingQueue.remove();
-                        ctx.sendDownstream(nextEvent.getChannelEvent());
-                        if (nextEvent.isLast()) {
-                            ++nextRequiredSequence;
-                            nextRequiredSubsequence = 0;
-                        } else {
-                            ++nextRequiredSubsequence;
-                        }
-                    }
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        if (msg instanceof SequencedOutboundMessage) {
+            SequencedOutboundMessage sequenced = (SequencedOutboundMessage) msg;
+
+            synchronizedWrite(ctx, sequenced, promise);
+        } else {
+            ctx.write(msg, promise);
+        }
+    }
+
+    private void progressToNextSendingSequence(ChannelHandlerContext ctx) {
+        currentlySendingSequence++;
+
+        int inFlight = this.inFlight.decrementAndGet();
+        // If we're now at the low water mark, issue a read for the next request
+        if (highWatermarkExceeded && inFlight == inFlightRequestsLowWatermark) {
+            ctx.read();
+            highWatermarkExceeded = false;
+        }
+    }
+
+    /**
+     * Write the next sequences, if buffered.
+     */
+    private void flushNextSequences(ChannelHandlerContext ctx) {
+        synchronized (bufferedEvents) {
+
+            progressToNextSendingSequence(ctx);
 
+            List<WriteMessage> toFlush = bufferedEvents.get(currentlySendingSequence);
+
+            // Loop while we still have a sequence to flush
+            while (toFlush != null) {
+
+                bufferedEvents.remove(currentlySendingSequence);
+
+                WriteMessage lastWritten = null;
+
+                // Flush each event
+                for (WriteMessage message: toFlush) {
+                    ctx.write(message.message.getMessage(), message.promise);
+                    lastWritten = message;
+                }
+
+                // If the last message that we wrote was the last message for that sequence,
+                // then increment the sequence and maybe get the next sequence from the buffer.
+                if (lastWritten != null && lastWritten.message.isLast()) {
+                    progressToNextSendingSequence(ctx);
+                    toFlush = bufferedEvents.get(currentlySendingSequence);
                 } else {
-                    channelShouldClose = true;
+                    toFlush = null;
                 }
             }
+        }
+    }
 
-            if (channelShouldClose) {
-                Channels.close(e.getChannel());
+    /**
+     * Perform a write, synchronized on the buffer.
+     */
+    private void synchronizedWrite(ChannelHandlerContext ctx, SequencedOutboundMessage sequenced, ChannelPromise promise) {
+        synchronized (bufferedEvents) {
+            if (sequenced.getSequence() == currentlySendingSequence) {
+                ctx.write(sequenced.getMessage(), promise);
+                if (sequenced.isLast()) {
+                    flushNextSequences(ctx);
+                }
+            } else {
+                List<WriteMessage> sequenceBuffer = bufferedEvents.get(sequenced.getSequence());
+                if (sequenceBuffer == null) {
+                    sequenceBuffer = new ArrayList<>();
+                    bufferedEvents.put(sequenced.getSequence(), sequenceBuffer);
+                }
+                sequenceBuffer.add(new WriteMessage(sequenced, promise));
             }
-        } else {
-            super.handleDownstream(ctx, e);
         }
     }
-
 }