Port ChunkedWriteHandler

codec/src/main/java/io/netty/handler/codec/embedder/EmbeddedChannel.java

@@ -16,7 +16,6 @@
 package io.netty.handler.codec.embedder;
 
 import io.netty.buffer.ChannelBuffer;
-import io.netty.buffer.ChannelBuffers;
 import io.netty.channel.AbstractChannel;
 import io.netty.channel.ChannelBufferHolder;
 import io.netty.channel.ChannelBufferHolders;
@@ -37,8 +36,7 @@ class EmbeddedChannel extends AbstractChannel {
     private int state; // 0 = OPEN, 1 = ACTIVE, 2 = CLOSED
 
     EmbeddedChannel(Queue<Object> productQueue) {
-        super(null, null, ChannelBufferHolders.catchAllBuffer(
+        super(null, null, ChannelBufferHolders.catchAllBuffer());
-                productQueue, ChannelBuffers.dynamicBuffer()));
         this.productQueue = productQueue;
     }
 
@@ -105,7 +103,12 @@ class EmbeddedChannel extends AbstractChannel {
             productQueue.add(byteBuf.readBytes(byteBufLen));
             byteBuf.clear();
         }
-        // We do nothing for message buffer because it's actually productQueue.
+
+        Queue<Object> msgBuf = buf.messageBuffer();
+        if (!msgBuf.isEmpty()) {
+            productQueue.addAll(msgBuf);
+            msgBuf.clear();
+        }
     }
 
     @Override

handler/pom.xml

@@ -39,6 +39,11 @@
       <artifactId>netty-transport</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>netty-codec</artifactId>
+      <version>${project.version}</version>
+    </dependency>
   </dependencies>
 
   <build>

handler/src/main/java/io/netty/handler/stream/ChunkedFile.java

@@ -21,8 +21,6 @@ import java.io.File;
 import java.io.IOException;
 import java.io.RandomAccessFile;
 
-import io.netty.channel.FileRegion;
-
 /**
  * A {@link ChunkedInput} that fetches data from a file chunk by chunk.
  * <p>

handler/src/main/java/io/netty/handler/stream/ChunkedNioFile.java

@@ -23,8 +23,6 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 
-import io.netty.channel.FileRegion;
-
 /**
  * A {@link ChunkedInput} that fetches data from a file chunk by chunk using
  * NIO {@link FileChannel}.

handler/src/main/java/io/netty/handler/stream/ChunkedWriteHandler.java

@@ -17,20 +17,25 @@ package io.netty.handler.stream;
 
 import io.netty.buffer.ChannelBuffers;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelBufferHolder;
+import io.netty.channel.ChannelBufferHolders;
+import io.netty.channel.ChannelException;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerContext;
 import io.netty.channel.ChannelPipeline;
 import io.netty.logging.InternalLogger;
 import io.netty.logging.InternalLoggerFactory;
 import io.netty.util.internal.QueueFactory;
 
-import java.io.IOException;
-import java.nio.channels.Channels;
 import java.nio.channels.ClosedChannelException;
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * A {@link ChannelHandler} that adds support for writing a large data stream
@@ -65,16 +70,36 @@ import java.util.concurrent.atomic.AtomicBoolean;
  * @apiviz.landmark
  * @apiviz.has io.netty.handler.stream.ChunkedInput oneway - - reads from
  */
-public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDownstreamHandler, LifeCycleAwareChannelHandler {
+public class ChunkedWriteHandler extends ChannelHandlerAdapter<Object, Object> {
 
     private static final InternalLogger logger =
         InternalLoggerFactory.getInstance(ChunkedWriteHandler.class);
 
-    private final Queue<MessageEvent> queue = QueueFactory.createQueue(MessageEvent.class);
+    private static final int MAX_PENDING_WRITES = 4;
+
+    private final Queue<Object> queue = QueueFactory.createQueue();
 
     private volatile ChannelHandlerContext ctx;
+    private final AtomicInteger pendingWrites = new AtomicInteger();
     private final AtomicBoolean flush = new AtomicBoolean(false);
-    private MessageEvent currentEvent;
+    private Object currentEvent;
+
+    @Override
+    public ChannelBufferHolder<Object> newInboundBuffer(
+            ChannelInboundHandlerContext<Object> ctx) throws Exception {
+        this.ctx = ctx;
+        return ChannelBufferHolders.inboundBypassBuffer(ctx);
+    }
+
+    @Override
+    public ChannelBufferHolder<Object> newOutboundBuffer(
+            ChannelOutboundHandlerContext<Object> ctx) throws Exception {
+        return ChannelBufferHolders.messageBuffer(queue);
+    }
+
+    private boolean isWritable() {
+        return pendingWrites.get() < MAX_PENDING_WRITES;
+    }
 
     /**
      * Continues to fetch the chunks from the input.
@@ -86,7 +111,7 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
         }
 
         try {
-            flush(ctx, false);
+            doFlush(ctx, false);
         } catch (Exception e) {
             if (logger.isWarnEnabled()) {
                 logger.warn("Unexpected exception while sending chunks.", e);
@@ -94,50 +119,25 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
         }
     }
 
-    public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent e)
+    @Override
-            throws Exception {
+    public void flush(ChannelOutboundHandlerContext<Object> ctx, ChannelFuture future) throws Exception {
-        if (!(e instanceof MessageEvent)) {
+        queue.add(future);
-            ctx.sendDownstream(e);
+        if (isWritable() || !ctx.channel().isActive()) {
-            return;
+            doFlush(ctx, false);
-        }
-
-        boolean offered = queue.offer((MessageEvent) e);
-        assert offered;
-
-        final Channel channel = ctx.getChannel();
-        // call flush if the channel is writable or not connected. flush(..) will take care of the rest
-
-        if (channel.isWritable() || !channel.isConnected()) {
-            this.ctx = ctx;
-            flush(ctx, false);
         }
     }
 
-    public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
+    @Override
-            throws Exception {
+    public void channelInactive(ChannelInboundHandlerContext<Object> ctx) throws Exception {
-        if (e instanceof ChannelStateEvent) {
+        doFlush(ctx, true);
-            ChannelStateEvent cse = (ChannelStateEvent) e;
+        super.channelInactive(ctx);
-            switch (cse.getState()) {
-            case INTEREST_OPS:
-                // Continue writing when the channel becomes writable.
-                flush(ctx, true);
-                break;
-            case OPEN:
-                if (!Boolean.TRUE.equals(cse.getValue())) {
-                    // Fail all pending writes
-                    flush(ctx, true);
-                }
-                break;
-            }
-        }
-        ctx.sendUpstream(e);
     }
 
-    private void discard(ChannelHandlerContext ctx, boolean fireNow) {
+    private void discard(final ChannelHandlerContext ctx, final Throwable cause) {
-        ClosedChannelException cause = null;
 
+        boolean fireExceptionCaught = false;
         for (;;) {
-            MessageEvent currentEvent = this.currentEvent;
+            Object currentEvent = this.currentEvent;
 
             if (this.currentEvent == null) {
                 currentEvent = queue.poll();
@@ -149,45 +149,39 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
                 break;
             }
 
-
+            if (currentEvent instanceof ChunkedInput) {
-            Object m = currentEvent.getMessage();
+                closeInput((ChunkedInput) currentEvent);
-            if (m instanceof ChunkedInput) {
+            } else if (currentEvent instanceof ChannelFuture) {
-                closeInput((ChunkedInput) m);
+                fireExceptionCaught = true;
+                ((ChannelFuture) currentEvent).setFailure(cause);
+            }
         }
 
-            // Trigger a ClosedChannelException
+        if (fireExceptionCaught) {
-            if (cause == null) {
+            if (ctx.eventLoop().inEventLoop()) {
-                cause = new ClosedChannelException();
+                ctx.fireExceptionCaught(cause);
-            }
-            currentEvent.getFuture().setFailure(cause);
-
-            currentEvent = null;
-        }
-
-
-        if (cause != null) {
-            if (fireNow) {
-                Channels.fireExceptionCaught(ctx.getChannel(), cause);
             } else {
-                Channels.fireExceptionCaughtLater(ctx.getChannel(), cause);
+                ctx.eventLoop().execute(new Runnable() {
+                    @Override
+                    public void run() {
+                        ctx.fireExceptionCaught(cause);
+                    }
+                });
             }
         }
     }
 
-    private void flush(ChannelHandlerContext ctx, boolean fireNow) throws Exception {
+    private void doFlush(final ChannelHandlerContext ctx, boolean fireNow) throws Exception {
         boolean acquired = false;
-        final Channel channel = ctx.getChannel();
+        Channel channel = ctx.channel();
-
+        // use CAS to see if the have flush already running, if so we don't need to take further actions
-        // use CAS to see if the have flush already running, if so we don't need to take futher actions
         if (acquired = flush.compareAndSet(false, true)) {
             try {
-
+                if (!channel.isActive()) {
-                if (!channel.isConnected()) {
+                    discard(ctx, new ClosedChannelException());
-                    discard(ctx, fireNow);
                     return;
                 }
-
+                while (isWritable()) {
-                while (channel.isWritable()) {
                     if (currentEvent == null) {
                         currentEvent = queue.poll();
                     }
@@ -196,15 +190,12 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
                         break;
                     }
 
-                    if (currentEvent.getFuture().isDone()) {
+                    final Object currentEvent = this.currentEvent;
-                        // Skip the current request because the previous partial write
+                    if (currentEvent instanceof ChannelFuture) {
-                        // attempt for the current request has been failed.
+                        this.currentEvent = null;
-                        currentEvent = null;
+                        ctx.flush((ChannelFuture) currentEvent);
-                    } else {
+                    } else if (currentEvent instanceof ChunkedInput) {
-                        final MessageEvent currentEvent = this.currentEvent;
+                        final ChunkedInput chunks = (ChunkedInput) currentEvent;
-                        Object m = currentEvent.getMessage();
-                        if (m instanceof ChunkedInput) {
-                            final ChunkedInput chunks = (ChunkedInput) m;
                         Object chunk;
                         boolean endOfInput;
                         boolean suspend;
@@ -218,14 +209,18 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
                             } else {
                                 suspend = false;
                             }
-                            } catch (Throwable t) {
+                        } catch (final Throwable t) {
                             this.currentEvent = null;
 
-                                currentEvent.getFuture().setFailure(t);
+                            if (ctx.eventLoop().inEventLoop()) {
-                                if (fireNow) {
+                                ctx.fireExceptionCaught(t);
-                                    fireExceptionCaught(ctx, t);
                             } else {
-                                    fireExceptionCaughtLater(ctx, t);
+                                ctx.eventLoop().execute(new Runnable() {
+                                    @Override
+                                    public void run() {
+                                        ctx.fireExceptionCaught(t);
+                                    }
+                                });
                             }
 
                             closeInput(chunks);
@@ -237,48 +232,54 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
                             // not reached at the end of input.  Let's wait until
                             // more chunks arrive.  Nothing to write or notify.
                             break;
-                            } else {
+                        }
-                                ChannelFuture writeFuture;
+
+                        pendingWrites.incrementAndGet();
+                        ctx.nextOutboundMessageBuffer().add(chunk);
+                        ChannelFuture f = ctx.flush();
                         if (endOfInput) {
                             this.currentEvent = null;
-                                    writeFuture = currentEvent.getFuture();
 
                             // Register a listener which will close the input once the write is complete. This is needed because the Chunk may have
                             // some resource bound that can not be closed before its not written
                             //
                             // See https://github.com/netty/netty/issues/303
-                                    writeFuture.addListener(new ChannelFutureListener() {
+                            f.addListener(new ChannelFutureListener() {
-
                                 @Override
                                 public void operationComplete(ChannelFuture future) throws Exception {
+                                    pendingWrites.decrementAndGet();
                                     closeInput(chunks);
                                 }
                             });
-                                } else {
+                        } else if (isWritable()) {
-                                    writeFuture = future(channel);
+                            f.addListener(new ChannelFutureListener() {
-                                    writeFuture.addListener(new ChannelFutureListener() {
                                 @Override
                                 public void operationComplete(ChannelFuture future) throws Exception {
+                                    pendingWrites.decrementAndGet();
                                     if (!future.isSuccess()) {
-                                                currentEvent.getFuture().setFailure(future.getCause());
+                                        closeInput((ChunkedInput) currentEvent);
-                                                closeInput((ChunkedInput) currentEvent.getMessage());
+                                    }
+                                }
+                            });
+                        } else {
+                            f.addListener(new ChannelFutureListener() {
+                                @Override
+                                public void operationComplete(ChannelFuture future) throws Exception {
+                                    pendingWrites.decrementAndGet();
+                                    if (!future.isSuccess()) {
+                                        closeInput((ChunkedInput) currentEvent);
+                                    } else if (isWritable()) {
+                                        resumeTransfer();
                                     }
                                 }
                             });
                         }
-
-                                Channels.write(
-                                        ctx, writeFuture, chunk,
-                                        currentEvent.getRemoteAddress());
-                            }
                     } else {
-                            this.currentEvent = null;
+                        ctx.nextOutboundMessageBuffer().add(currentEvent);
-                            ctx.sendDownstream(currentEvent);
-                        }
                     }
 
-                    if (!channel.isConnected()) {
+                    if (!channel.isActive()) {
-                        discard(ctx, fireNow);
+                        discard(ctx, new ClosedChannelException());
                         return;
                     }
                 }
@@ -286,11 +287,10 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
                 // mark the flush as done
                 flush.set(false);
             }
-
         }
 
-        if (acquired && (!channel.isConnected() || channel.isWritable() && !queue.isEmpty())) {
+        if (acquired && (!channel.isActive() || isWritable() && !queue.isEmpty())) {
-            flush(ctx, fireNow);
+            doFlush(ctx, fireNow);
         }
     }
 
@@ -304,61 +304,19 @@ public class ChunkedWriteHandler implements ChannelUpstreamHandler, ChannelDowns
         }
     }
 
-    public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
+    @Override
-        // nothing to do
-
-    }
-
-    public void afterAdd(ChannelHandlerContext ctx) throws Exception {
-        // nothing to do
-
-    }
-
     public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
         // try to flush again a last time.
         //
         // See #304
-        flush(ctx, false);
+        doFlush(ctx, false);
     }
 
     // This method should not need any synchronization as the ChunkedWriteHandler will not receive any new events
+    @Override
     public void afterRemove(ChannelHandlerContext ctx) throws Exception {
         // Fail all MessageEvent's that are left. This is needed because otherwise we would never notify the
         // ChannelFuture and the registered FutureListener. See #304
-        //
+        discard(ctx, new ChannelException(ChunkedWriteHandler.class.getSimpleName() + " removed from pipeline."));
-        Throwable cause = null;
-        boolean fireExceptionCaught = false;
-
-        for (;;) {
-            MessageEvent currentEvent = this.currentEvent;
-
-            if (this.currentEvent == null) {
-                currentEvent = queue.poll();
-            } else {
-                this.currentEvent = null;
-            }
-
-            if (currentEvent == null) {
-                break;
-            }
-
-            Object m = currentEvent.getMessage();
-            if (m instanceof ChunkedInput) {
-                closeInput((ChunkedInput) m);
-            }
-
-            // Create exception
-            if (cause == null) {
-                cause = new IOException("Unable to flush event, discarding");
-            }
-            currentEvent.getFuture().setFailure(cause);
-            fireExceptionCaught = true;
-
-            currentEvent = null;
-        }
-
-        if (fireExceptionCaught) {
-            Channels.fireExceptionCaughtLater(ctx.getChannel(), cause);
-        }
     }
 }

handler/src/test/java/io/netty/handler/stream/ChunkedWriteHandlerTest.java

@@ -16,6 +16,7 @@
 package io.netty.handler.stream;
 
 import io.netty.buffer.ChannelBuffer;
+import io.netty.handler.codec.embedder.EncoderEmbedder;
 
 import java.io.ByteArrayInputStream;
 import java.io.File;
@@ -89,7 +90,9 @@ public class ChunkedWriteHandlerTest {
     }
 
     private static void check(ChunkedInput... inputs) {
-        EncoderEmbedder<ChannelBuffer> embedder = new EncoderEmbedder<ChannelBuffer>(new ChunkedWriteHandler());
+        EncoderEmbedder<ChannelBuffer> embedder =
+                new EncoderEmbedder<ChannelBuffer>(new ChunkedWriteHandler());
+
         for (ChunkedInput input: inputs) {
             embedder.offer(input);
         }