Fix issue #422 (#423)

* Fix issue #422

It's an regression for #280, the sharedBufferList may leak to async path.

* Do check for DefaultChannelConfiguration.WriteBufferHighWaterMark when set

* Also check whether IncompleteWrite finish in sync for AbstractSocketByteChannel.

Otherwise may not all flushed buffer will be writen before next flush

* Invalid nio buffer cache for partial writen

src/DotNetty.Transport/Channels/ChannelOutboundBuffer.cs

@@ -309,6 +309,10 @@ namespace DotNetty.Transport.Channels
                     // readableBytes > writtenBytes
                     if (writtenBytes != 0)
                     {
+                        //Invalid nio buffer cache for partial writen, see https://github.com/Azure/DotNetty/issues/422
+                        this.flushedEntry.Buffer = new ArraySegment<byte>();
+                        this.flushedEntry.Buffers = null;
+
                         buf.SetReaderIndex(readerIndex + (int)writtenBytes);
                         this.Progress(writtenBytes);
                     }

src/DotNetty.Transport/Channels/DefaultChannelConfiguration.cs

@@ -118,7 +118,7 @@ namespace DotNetty.Transport.Channels
             }
             else if (ChannelOption.WriteBufferHighWaterMark.Equals(option))
             {
-                this.writeBufferHighWaterMark = (int)(object)value;
+                this.WriteBufferHighWaterMark = (int)(object)value;
             }
             else if (ChannelOption.WriteBufferLowWaterMark.Equals(option))
             {

src/DotNetty.Transport/Channels/Sockets/AbstractSocketByteChannel.cs

@@ -231,9 +231,8 @@ namespace DotNetty.Transport.Channels.Sockets
                     {
                         input.Remove();
                     }
-                    else
+                    else if (this.IncompleteWrite(scheduleAsync, this.PrepareWriteOperation(buf.GetIoBuffer())))
                     {
-                        this.IncompleteWrite(scheduleAsync, this.PrepareWriteOperation(buf.GetIoBuffer()));
                         break;
                     }
                 } /*else if (msg is FileRegion) { todo: FileRegion support

src/DotNetty.Transport/Channels/Sockets/TcpSocketChannel.cs

@@ -316,7 +316,12 @@ namespace DotNetty.Transport.Channels.Sockets
 
                     if (!done)
                     {
-                        SocketChannelAsyncOperation asyncOperation = this.PrepareWriteOperation(bufferList);
+                        IList<ArraySegment<byte>> asyncBufferList = bufferList;
+                        if (object.ReferenceEquals(sharedBufferList, asyncBufferList))
+                        {
+                            asyncBufferList = sharedBufferList.ToArray(); // copying buffers to
+                        }
+                        SocketChannelAsyncOperation asyncOperation = this.PrepareWriteOperation(asyncBufferList);
 
                         // Did not write all buffers completely.
                         if (this.IncompleteWrite(true, asyncOperation))