Imlemented progress notification for NIO writes

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

@@ -394,7 +394,9 @@ public class DefaultChannelFuture implements ChannelFuture {
                 return false;
             }
 
-            if (progressListeners.isEmpty()) {
+            Collection<ChannelFutureProgressListener> progressListeners =
+                this.progressListeners;
+            if (progressListeners == null || progressListeners.isEmpty()) {
                 // Nothing to notify - no need to create an empty array.
                 return true;
             }

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

@@ -466,7 +466,9 @@ class NioWorker implements Runnable {
                     bb = buf.buffer;
                 }
 
+                ChannelFuture future = evt.getFuture();
                 try {
+                    int oldWrittenBytes = writtenBytes;
                     for (int i = writeSpinCount; i > 0; i --) {
                         int localWrittenBytes = ch.write(bb);
                         if (localWrittenBytes != 0) {
@@ -478,7 +480,6 @@ class NioWorker implements Runnable {
                     if (!bb.hasRemaining()) {
                         // Successful write - proceed to the next message.
                         buf.release();
-                        ChannelFuture future = evt.getFuture();
                         channel.currentWriteEvent = null;
                         channel.currentWriteBuffer = null;
                         evt = null;
@@ -489,13 +490,18 @@ class NioWorker implements Runnable {
                         // Not written fully - perhaps the kernel buffer is full.
                         addOpWrite = true;
                         channel.writeSuspended = true;
+
+                        // Notify progress listeners if necessary.
+                        future.setProgress(
+                                writtenBytes - oldWrittenBytes,
+                                bb.position() - buf.initialPos,
+                                bb.limit() - buf.initialPos);
                         break;
                     }
                 } catch (AsynchronousCloseException e) {
                     // Doesn't need a user attention - ignore.
                 } catch (Throwable t) {
                     buf.release();
-                    ChannelFuture future = evt.getFuture();
                     channel.currentWriteEvent = null;
                     channel.currentWriteBuffer = null;
                     buf = null;

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

@@ -145,10 +145,12 @@ final class SocketSendBufferPool {
     final class SendBuffer {
         private final Preallocation parent;
         final ByteBuffer buffer;
+        final int initialPos;
 
         SendBuffer(Preallocation parent, ByteBuffer buffer) {
             this.parent = parent;
             this.buffer = buffer;
+            initialPos = buffer.position();
         }
 
         void release() {