Merge pull request #32 from normanmaurer/master

Push fixes from 3.2 branch to master

src/main/java/org/jboss/netty/channel/ChannelEvent.java

@@ -71,6 +71,7 @@ import org.jboss.netty.channel.socket.ServerSocketChannel;
  * <td>{@code "channelOpen"}</td>
  * <td>{@link ChannelStateEvent}<br/>(state = {@link ChannelState#OPEN OPEN}, value = {@code true})</td>
  * <td>a {@link Channel} is open, but not bound nor connected</td>
+ * <td><strong>Be aware that this event is fired from within the Boss-Thread so you should not execute any heavy operation in there as it will block the dispatching to other workers!</strong></td>
  * </tr>
  * <tr>
  * <td>{@code "channelClosed"}</td>
@@ -81,6 +82,7 @@ import org.jboss.netty.channel.socket.ServerSocketChannel;
  * <td>{@code "channelBound"}</td>
  * <td>{@link ChannelStateEvent}<br/>(state = {@link ChannelState#BOUND BOUND}, value = {@link SocketAddress})</td>
  * <td>a {@link Channel} is open and bound to a local address, but not connected</td>
+ * <td><strong>Be aware that this event is fired from within the Boss-Thread so you should not execute any heavy operation in there as it will block the dispatching to other workers!</strong></td>
  * </tr>
  * <tr>
  * <td>{@code "channelUnbound"}</td>

src/main/java/org/jboss/netty/channel/SimpleChannelUpstreamHandler.java

@@ -151,7 +151,9 @@ public class SimpleChannelUpstreamHandler implements ChannelUpstreamHandler {
 
     /**
      * Invoked when a {@link Channel} is open, but not bound nor connected.
-     */
+     * <br/>
+     * <strong>Be aware that this event is fired from within the Boss-Thread so you should not execute any heavy operation in there as it will block the dispatching to other workers!</strong>    
+    */
     public void channelOpen(
             ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
         ctx.sendUpstream(e);
@@ -160,6 +162,8 @@ public class SimpleChannelUpstreamHandler implements ChannelUpstreamHandler {
     /**
      * Invoked when a {@link Channel} is open and bound to a local address,
      * but not connected.
+     * <br/>
+     * <strong>Be aware that this event is fired from within the Boss-Thread so you should not execute any heavy operation in there as it will block the dispatching to other workers!</strong>    
      */
     public void channelBound(
             ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {

src/main/java/org/jboss/netty/channel/socket/nio/NioAcceptedSocketChannel.java

@@ -48,6 +48,5 @@ final class NioAcceptedSocketChannel extends NioSocketChannel {
         setConnected();
         fireChannelOpen(this);
         fireChannelBound(this, getLocalAddress());
-        fireChannelConnected(this, getRemoteAddress());
     }
 }

src/main/java/org/jboss/netty/channel/socket/nio/NioWorker.java

@@ -782,6 +782,11 @@ class NioWorker implements Runnable {
                 }
                 fireChannelConnected(channel, remoteAddress);
             }
+            
+            // Handle the channelConnected in the worker thread
+            if (channel instanceof NioAcceptedSocketChannel) {
+                fireChannelConnected(channel, channel.getRemoteAddress());
+            }
         }
     }
 }

src/main/java/org/jboss/netty/channel/socket/nio/SocketSendBufferPool.java

@@ -306,7 +306,8 @@ final class SocketSendBufferPool {
 
         @Override
         public void release() {
-            // Unpooled.
+            // Make sure the FileRegion resource are released otherwise it may cause a FD leak or something similar
+            file.releaseExternalResources(); 
         }
     }
 

src/main/java/org/jboss/netty/channel/socket/oio/OioAcceptedSocketChannel.java

@@ -63,7 +63,6 @@ class OioAcceptedSocketChannel extends OioSocketChannel {
 
         fireChannelOpen(this);
         fireChannelBound(this, getLocalAddress());
-        fireChannelConnected(this, getRemoteAddress());
     }
 
     @Override

src/main/java/org/jboss/netty/channel/socket/oio/OioWorker.java

@@ -20,12 +20,15 @@ import static org.jboss.netty.channel.Channels.*;
 import java.io.OutputStream;
 import java.io.PushbackInputStream;
 import java.net.SocketException;
+import java.nio.channels.Channels;
 import java.nio.channels.ClosedChannelException;
+import java.nio.channels.WritableByteChannel;
 import java.util.regex.Pattern;
 
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.FileRegion;
 
 /**
  *
@@ -50,8 +53,13 @@ class OioWorker implements Runnable {
     public void run() {
         channel.workerThread = Thread.currentThread();
         final PushbackInputStream in = channel.getInputStream();
+        boolean fireOpen = channel instanceof OioAcceptedSocketChannel;
 
         while (channel.isOpen()) {
+            if (fireOpen) {
+                fireOpen = false;
+                fireChannelConnected(channel, channel.getRemoteAddress());
+            }
             synchronized (channel.interestOpsLock) {
                 while (!channel.isReadable()) {
                     try {
@@ -114,13 +122,39 @@ class OioWorker implements Runnable {
         }
 
         try {
-            ChannelBuffer a = (ChannelBuffer) message;
+            int length = 0;
-            int length = a.readableBytes();
+
-            synchronized (out) {
+            // Add support to write a FileRegion. This in fact will not give any performance gain but at least it not fail and 
-                a.getBytes(a.readerIndex(), out, length);
+            // we did the best to emulate it
+            if (message instanceof FileRegion) {
+                FileRegion fr = (FileRegion) message;
+                try {
+                    synchronized (out) {
+                        WritableByteChannel  bchannel = Channels.newChannel(out);
+                        
+                        long i = 0;
+                        while ((i = fr.transferTo(bchannel, length)) > 0) {
+                            length += i;
+                            if (length >= fr.getCount()) {
+                                break;
+                            }
+                        }
+                    }
+                } finally {
+                    fr.releaseExternalResources();
+
+                }
+            } else {
+                ChannelBuffer a = (ChannelBuffer) message;
+                length = a.readableBytes();
+                synchronized (out) {
+                    a.getBytes(a.readerIndex(), out, length);
+                }
             }
+
             fireWriteComplete(channel, length);
             future.setSuccess();
+ 
         } catch (Throwable t) {
             // Convert 'SocketException: Socket closed' to
             // ClosedChannelException.