Let FileRegion extend ReferenceCounted and add ChannelGroup.flush() , ChannelGroup.sendFile(..)

transport/src/main/java/io/netty/channel/AbstractChannel.java

@@ -522,7 +522,7 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
                     // the first FileRegion to flush so trigger it now!
                     doFlushFileRegion(region, promise);
                 } catch (Throwable cause) {
-                    region.close();
+                    region.release();
                     promise.setFailure(cause);
                 }
                 return;

transport/src/main/java/io/netty/channel/DefaultFileRegion.java

@@ -15,6 +15,7 @@
  */
 package io.netty.channel;
 
+import io.netty.buffer.AbstractReferenceCounted;
 import io.netty.logging.InternalLogger;
 import io.netty.logging.InternalLoggerFactory;
 
@@ -25,7 +26,7 @@ import java.nio.channels.WritableByteChannel;
 /**
  * Default {@link FileRegion} implementation which transfer data from a {@link FileChannel}.
  */
-public class DefaultFileRegion implements FileRegion {
+public class DefaultFileRegion extends AbstractReferenceCounted implements FileRegion {
 
     private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultFileRegion.class);
 
@@ -81,7 +82,7 @@ public class DefaultFileRegion implements FileRegion {
     }
 
     @Override
-    public void close() {
+    protected void deallocate() {
         try {
             file.close();
         } catch (IOException e) {

transport/src/main/java/io/netty/channel/FileRegion.java

@@ -15,6 +15,8 @@
  */
 package io.netty.channel;
 
+import io.netty.buffer.ReferenceCounted;
+
 import java.io.IOException;
 import java.nio.channels.FileChannel;
 import java.nio.channels.WritableByteChannel;
@@ -50,7 +52,7 @@ import java.nio.channels.WritableByteChannel;
  *
  * Currently, the NIO transport is the only transport that supports {@link FileRegion}.
  */
-public interface FileRegion {
+public interface FileRegion extends ReferenceCounted {
 
     /**
      * Returns the offset in the file where the transfer began.
@@ -73,9 +75,4 @@ public interface FileRegion {
      *                  byte of the region transferred.
      */
     long transferTo(WritableByteChannel target, long position) throws IOException;
-
-    /**
-     * Close the {@link FileRegion}. After calling this method accessing the {@link FileRegion} may fail.
-     */
-    void close();
 }

transport/src/main/java/io/netty/channel/group/ChannelGroup.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012 The Netty Project
+ * Copyright 2013 The Netty Project
  *
  * The Netty Project licenses this file to you under the Apache License,
  * version 2.0 (the "License"); you may not use this file except in compliance
@@ -21,6 +21,7 @@ import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelStateHandlerAdapter;
+import io.netty.channel.FileRegion;
 import io.netty.channel.ServerChannel;
 import io.netty.util.CharsetUtil;
 
@@ -116,6 +117,25 @@ public interface ChannelGroup extends Set<Channel>, Comparable<ChannelGroup> {
      */
     ChannelGroupFuture write(Object message);
 
+    /**
+     * Writes the specified {@link FileRegion} to all {@link Channel}s in this
+     * group. Please note that this operation is asynchronous as
+     * {@link Channel#sendFile(FileRegion)} is.
+     *
+     * @return the {@link ChannelGroupFuture} instance that notifies when
+     *         the operation is done for all channels
+     */
+    ChannelGroupFuture sendFile(FileRegion region);
+
+    /**
+     * Flush all {@link Channel} in this group. Please note that this operation
+     * is asynchronous as {@link Channel#flush()} is.
+     *
+     * @return the {@link ChannelGroupFuture} instance that notifies when
+     *         the operation is done for all channels
+     */
+    ChannelGroupFuture flush();
+
     /**
      * Disconnects all {@link Channel}s in this group from their remote peers.
      *

transport/src/main/java/io/netty/channel/group/DefaultChannelGroup.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012 The Netty Project
+ * Copyright 2013 The Netty Project
  *
  * The Netty Project licenses this file to you under the Apache License,
  * version 2.0 (the "License"); you may not use this file except in compliance
@@ -19,6 +19,7 @@ import io.netty.buffer.BufUtil;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.FileRegion;
 import io.netty.channel.ServerChannel;
 
 import java.util.AbstractSet;
@@ -221,6 +222,32 @@ public class DefaultChannelGroup extends AbstractSet<Channel> implements Channel
         return new DefaultChannelGroupFuture(this, futures);
     }
 
+    @Override
+    public ChannelGroupFuture sendFile(FileRegion region) {
+        if (region == null) {
+            throw new NullPointerException("region");
+        }
+
+        Map<Integer, ChannelFuture> futures = new LinkedHashMap<Integer, ChannelFuture>(size());
+        for (Channel c: nonServerChannels.values()) {
+            BufUtil.retain(region);
+            futures.put(c.id(), c.sendFile(region));
+        }
+
+        BufUtil.release(region);
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
+    @Override
+    public ChannelGroupFuture flush() {
+        Map<Integer, ChannelFuture> futures = new LinkedHashMap<Integer, ChannelFuture>(size());
+        for (Channel c: nonServerChannels.values()) {
+            futures.put(c.id(), c.flush());
+        }
+
+        return new DefaultChannelGroupFuture(this, futures);
+    }
+
     @Override
     public int hashCode() {
         return System.identityHashCode(this);

transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java

@@ -192,14 +192,14 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
                     } else {
                         writtenBytes += localWrittenBytes;
                         if (writtenBytes >= region.count()) {
-                            region.close();
+                            region.release();
                             promise.setSuccess();
                             return;
                         }
                     }
                 }
             } catch (Throwable cause) {
-                region.close();
+                region.release();
                 promise.setFailure(cause);
             }
         }
@@ -217,7 +217,7 @@ public abstract class AbstractNioByteChannel extends AbstractNioChannel {
             }
 
             if (writtenBytes < region.count()) {
-                region.close();
+                region.release();
                 if (!isOpen()) {
                     promise.setFailure(new ClosedChannelException());
                 } else {

transport/src/main/java/io/netty/channel/oio/OioByteStreamChannel.java

@@ -117,7 +117,7 @@ public abstract class OioByteStreamChannel extends AbstractOioByteChannel {
             long localWritten = region.transferTo(outChannel, written);
             if (localWritten == -1) {
                 checkEOF(region, written);
-                region.close();
+                region.release();
                 promise.setSuccess();
                 return;
             }

transport/src/main/java/io/netty/channel/socket/aio/AioSocketChannel.java

@@ -552,7 +552,7 @@ public class AioSocketChannel extends AbstractAioChannel implements SocketChanne
                         written += result;
 
                         if (written >= region.count()) {
-                            region.close();
+                            region.release();
                             promise.setSuccess();
                             return;
                         }
@@ -562,14 +562,14 @@ public class AioSocketChannel extends AbstractAioChannel implements SocketChanne
                             region.transferTo(WritableByteChannelAdapter.this, written);
                         }
                     } catch (Throwable cause) {
-                        region.close();
+                        region.release();
                         promise.setFailure(cause);
                     }
                 }
 
                 @Override
                 public void failed(Throwable exc, Object attachment) {
-                    region.close();
+                    region.release();
                     promise.setFailure(exc);
                 }
             });