post rebase fixes

examples/Telnet.Server/TelnetServerHandler.cs

@@ -6,6 +6,7 @@ namespace Telnet.Server
     using System;
     using System.Net;
     using System.Threading.Tasks;
+    using DotNetty.Codecs;
     using DotNetty.Common.Concurrency;
     using DotNetty.Transport.Channels;
 
@@ -17,7 +18,7 @@ public override void ChannelActive(IChannelHandlerContext contex)
             contex.WriteAndFlushAsync(string.Format("It is {0} now !\r\n", DateTime.Now));
         }
 
-        protected override async void ChannelRead0(IChannelHandlerContext contex, string msg)
+        protected override void ChannelRead0(IChannelHandlerContext context, string msg)
         {
             // Generate and write a response.
             string response;
@@ -36,11 +37,10 @@ protected override async void ChannelRead0(IChannelHandlerContext contex, string
                 response = "Did you say '" + msg + "'?\r\n";
             }
 
-            ValueTask waitClose = contex.WriteAndFlushAsync(response);
+            Task waitClose = context.WriteAndFlushAsync(response);
             if (close)
             {
-                await waitClose;
-                contex.CloseAsync();
+                waitClose.CloseOnComplete(context);
             }
         }
 

src/DotNetty.Codecs.Http/Cors/CorsHandler.cs

@@ -167,7 +167,7 @@ void SetExposeHeaders(IHttpResponse response)
 
         void SetMaxAge(IHttpResponse response) => response.Headers.Set(HttpHeaderNames.AccessControlMaxAge, this.config.MaxAge);
 
-        public override Task WriteAsync(IChannelHandlerContext context, object message)
+        public override ValueTask WriteAsync(IChannelHandlerContext context, object message)
         {
             if (this.config.IsCorsSupportEnabled && message is IHttpResponse response)
             {
@@ -177,7 +177,7 @@ public override Task WriteAsync(IChannelHandlerContext context, object message)
                     this.SetExposeHeaders(response);
                 }
             }
-            return context.WriteAndFlushAsync(message);
+            return context.WriteAndFlushAsync(message, true);
         }
 
         static void Forbidden(IChannelHandlerContext ctx, IHttpRequest request)
@@ -197,15 +197,8 @@ static void Respond(IChannelHandlerContext ctx, IHttpRequest request, IHttpRespo
             Task task = ctx.WriteAndFlushAsync(response);
             if (!keepAlive)
             {
-                task.ContinueWith(CloseOnComplete, ctx, 
-                    TaskContinuationOptions.ExecuteSynchronously);
+                task.CloseOnComplete(ctx);
             }
         }
-
-        static void CloseOnComplete(Task task, object state)
-        {
-            var ctx = (IChannelHandlerContext)state;
-            ctx.CloseAsync();
-        }
     }
 }

src/DotNetty.Codecs.Http/DotNetty.Codecs.Http.csproj

@@ -29,7 +29,7 @@
     <Compile Include="..\shared\SharedAssemblyInfo.cs" Link="Properties\SharedAssemblyInfo.cs" />
   </ItemGroup>
   <ItemGroup>
-    <PackageReference Include="System.Runtime.CompilerServices.Unsafe" Version="4.4.0" />
+    <PackageReference Include="System.Runtime.CompilerServices.Unsafe" Version="4.5.0-preview2-26406-04" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\DotNetty.Common\DotNetty.Common.csproj" />

src/DotNetty.Codecs.Http/HttpClientUpgradeHandler.cs

@@ -72,7 +72,7 @@ public HttpClientUpgradeHandler(ISourceCodec sourceCodec, IUpgradeCodec upgradeC
             this.upgradeCodec = upgradeCodec;
         }
 
-        public override Task WriteAsync(IChannelHandlerContext context, object message)
+        public override ValueTask WriteAsync(IChannelHandlerContext context, object message)
         {
             if (!(message is IHttpRequest))
             {
@@ -81,14 +81,14 @@ public override Task WriteAsync(IChannelHandlerContext context, object message)
 
             if (this.upgradeRequested)
             {
-                return TaskEx.FromException(new InvalidOperationException("Attempting to write HTTP request with upgrade in progress"));
+                return new ValueTask(TaskEx.FromException(new InvalidOperationException("Attempting to write HTTP request with upgrade in progress")));
             }
 
             this.upgradeRequested = true;
             this.SetUpgradeRequestHeaders(context, (IHttpRequest)message);
 
             // Continue writing the request.
-            Task task = context.WriteAsync(message);
+            ValueTask task = context.WriteAsync(message);
 
             // Notify that the upgrade request was issued.
             context.FireUserEventTriggered(UpgradeEvent.UpgradeIssued);

src/DotNetty.Codecs.Http/HttpServerExpectContinueHandler.cs

@@ -39,27 +39,15 @@ public override void ChannelRead(IChannelHandlerContext context, object message)
                         // the expectation failed so we refuse the request.
                         IHttpResponse rejection = this.RejectResponse(req);
                         ReferenceCountUtil.Release(message);
-                        context.WriteAndFlushAsync(rejection)
-                            .ContinueWith(CloseOnFailure, context, TaskContinuationOptions.ExecuteSynchronously);
+                        context.WriteAndFlushAsync(rejection).CloseOnFailure(context);
                         return;
                     }
 
-                    context.WriteAndFlushAsync(accept)
-                        .ContinueWith(CloseOnFailure, context, TaskContinuationOptions.ExecuteSynchronously);
+                    context.WriteAndFlushAsync(accept).CloseOnFailure(context);
                     req.Headers.Remove(HttpHeaderNames.Expect);
                 }
                 base.ChannelRead(context, message);
             }
         }
-
-        static Task CloseOnFailure(Task task, object state)
-        {
-            if (task.IsFaulted)
-            {
-                var context = (IChannelHandlerContext)state;
-                return context.CloseAsync();
-            }
-            return TaskEx.Completed;
-        }
     }
 }

src/DotNetty.Codecs.Http/HttpServerKeepAliveHandler.cs

@@ -31,7 +31,7 @@ public override void ChannelRead(IChannelHandlerContext context, object message)
             base.ChannelRead(context, message);
         }
 
-        public override Task WriteAsync(IChannelHandlerContext context, object message)
+        public override ValueTask WriteAsync(IChannelHandlerContext context, object message)
         {
             // modify message on way out to add headers if needed
             if (message is IHttpResponse response)
@@ -52,18 +52,11 @@ public override Task WriteAsync(IChannelHandlerContext context, object message)
             }
             if (message is ILastHttpContent && !this.ShouldKeepAlive())
             {
-                return base.WriteAsync(context, message)
-                    .ContinueWith(CloseOnComplete, context, TaskContinuationOptions.ExecuteSynchronously);
+                return new ValueTask(base.WriteAsync(context, message).CloseOnComplete(context));
             }
             return base.WriteAsync(context, message);
         }
 
-        static Task CloseOnComplete(Task task, object state)
-        {
-            var context = (IChannelHandlerContext)state;
-            return context.CloseAsync();
-        }
-
         void TrackResponse(IHttpResponse response)
         {
             if (!IsInformational(response))

src/DotNetty.Codecs/Compression/JZlibEncoder.cs

@@ -240,8 +240,7 @@ Task FinishEncode(IChannelHandlerContext context)
                 this.z.next_out = null;
             }
 
-            return context.WriteAndFlushAsync(footer)
-                .ContinueWith(_ => context.CloseAsync());
+            return context.WriteAndFlushAsync(footer).CloseOnComplete(context);
         }
 
         public override void HandlerAdded(IChannelHandlerContext context) => this.ctx = context;

src/DotNetty.Codecs/MessageAggregator.cs

@@ -130,13 +130,10 @@ protected internal override void Decode(IChannelHandlerContext context, TMessage
                     bool closeAfterWrite = this.CloseAfterContinueResponse(continueResponse);
                     this.handlingOversizedMessage = this.IgnoreContentAfterContinueResponse(continueResponse);
 
-                    Task task = context
-                        .WriteAndFlushAsync(continueResponse)
-                        .ContinueWith(ContinueResponseWriteAction, context, TaskContinuationOptions.ExecuteSynchronously);
+                    WriteContinueResponse(context, continueResponse, closeAfterWrite);
 
                     if (closeAfterWrite)
                     {
-                        task.ContinueWith(CloseAfterWriteAction, context, TaskContinuationOptions.ExecuteSynchronously);
                         return;
                     }
 
@@ -245,19 +242,21 @@ protected internal override void Decode(IChannelHandlerContext context, TMessage
                 throw new MessageAggregationException("Unknown aggregation state.");
             }
         }
-
-        static void CloseAfterWriteAction(Task task, object state)
+        
+        static async void WriteContinueResponse(IChannelHandlerContext ctx, object message, bool closeAfterWrite)
         {
-            var ctx = (IChannelHandlerContext)state;
-            ctx.Channel.CloseAsync();
-        }
-
-        static void ContinueResponseWriteAction(Task task, object state)
-        {
-            if (task.IsFaulted)
+            try
+            {
+                await ctx.WriteAndFlushAsync(message);
+            }
+            catch (Exception ex)
+            {
+                ctx.FireExceptionCaught(ex);
+            }
+
+            if (closeAfterWrite)
             {
-                var ctx = (IChannelHandlerContext)state;
-                ctx.FireExceptionCaught(task.Exception);
+                ctx.Channel.CloseAsync();
             }
         }
 

src/DotNetty.Codecs/MessageToMessageCodec.cs

@@ -50,7 +50,7 @@ protected MessageToMessageCodec()
         public sealed override void ChannelRead(IChannelHandlerContext context, object message) => 
             this.decoder.ChannelRead(context, message);
 
-        public sealed override Task WriteAsync(IChannelHandlerContext context, object message) => 
+        public sealed override ValueTask WriteAsync(IChannelHandlerContext context, object message) => 
             this.encoder.WriteAsync(context, message);
 
         public virtual bool AcceptInboundMessage(object msg) => msg is TInbound;

src/DotNetty.Codecs/TaskExtensions.cs

@@ -0,0 +1,59 @@
+// Copyright (c) Microsoft. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for full license information.
+
+namespace DotNetty.Codecs
+{
+    using System;
+    using System.Threading.Tasks;
+    using DotNetty.Common.Utilities;
+    using DotNetty.Transport.Channels;
+
+    public static class TaskExtensions
+    {
+        public static async Task CloseOnComplete(this ValueTask task, IChannelHandlerContext ctx)
+        {
+            try
+            {
+                await task;
+            }
+            finally
+            {
+                await ctx.CloseAsync();
+            }
+        }
+
+        static readonly Func<Task, object, Task> CloseOnCompleteContinuation = Close;
+        static readonly Func<Task, object, Task> CloseOnFailureContinuation = CloseOnFailure; 
+
+        public static Task CloseOnComplete(this Task task, IChannelHandlerContext ctx) 
+            => task.ContinueWith(CloseOnCompleteContinuation, ctx, TaskContinuationOptions.ExecuteSynchronously);
+
+        public static Task CloseOnComplete(this Task task, IChannel channel) 
+            => task.ContinueWith(CloseOnCompleteContinuation, channel, TaskContinuationOptions.ExecuteSynchronously);
+
+        public static Task CloseOnFailure(this Task task, IChannelHandlerContext ctx) 
+            => task.ContinueWith(CloseOnFailureContinuation, ctx, TaskContinuationOptions.ExecuteSynchronously);
+
+        static Task Close(Task task, object state)
+        {
+            switch (state)
+            {
+                case IChannelHandlerContext ctx:
+                    return ctx.CloseAsync();
+                case IChannel ch:
+                    return ch.CloseAsync();
+                default:
+                    throw new InvalidOperationException("must never get here");
+            }
+        }
+
+        static Task CloseOnFailure(Task task, object state)
+        {
+            if (task.IsFaulted)
+            {
+                return Close(task, state);
+            }
+            return TaskEx.Completed;
+        }
+    }
+}

src/DotNetty.Common/Concurrency/AbstractPromise.cs

@@ -22,7 +22,7 @@ public abstract class AbstractPromise : IPromise, IValueTaskSource
         static readonly Action<object> TaskSchedulerCallback = Execute;
         static readonly Action<object> TaskScheduleCallbackWithExecutionContext = ExecuteWithExecutionContext;
 
-        static readonly Exception CompletedSentinel = new Exception();
+        protected static readonly Exception CompletedSentinel = new Exception();
 
         short currentId;
         protected Exception exception;