Zero contentLength msgbody doesn't need state

Author: benaadams

src/Microsoft.AspNetCore.Server.Kestrel/Internal/Http/MessageBody.cs

@@ -8,12 +8,14 @@
 using Microsoft.AspNetCore.Http;
 using Microsoft.AspNetCore.Server.Kestrel.Internal.Infrastructure;
 using Microsoft.Extensions.Internal;
-using Microsoft.Extensions.Primitives;
 
 namespace Microsoft.AspNetCore.Server.Kestrel.Internal.Http
 {
     public abstract class MessageBody
     {
+        private static readonly MessageBody _zeroContentLengthClose = new ForZeroContentLength(keepAlive: false);
+        private static readonly MessageBody _zeroContentLengthKeepAlive = new ForZeroContentLength(keepAlive: true);
+
         private readonly Frame _context;
         private bool _send100Continue = true;
 
@@ -268,7 +270,13 @@ public static MessageBody For(
 
             if (headers.ContentLength.HasValue)
             {
-                return new ForContentLength(keepAlive, headers.ContentLength.Value, context);
+                var contentLength = headers.ContentLength.Value;
+                if (contentLength == 0)
+                {
+                    return keepAlive ? _zeroContentLengthKeepAlive : _zeroContentLengthClose;
+                }
+
+                return new ForContentLength(keepAlive, contentLength, context);
             }
 
             // Avoid slowing down most common case
@@ -283,7 +291,7 @@ public static MessageBody For(
                 }
             }
 
-            return new ForContentLength(keepAlive, 0, context);
+            return keepAlive ? _zeroContentLengthKeepAlive : _zeroContentLengthClose;
         }
 
         private class ForRemainingData : MessageBody
@@ -300,6 +308,28 @@ protected override ValueTask<ArraySegment<byte>> PeekAsync(CancellationToken can
             }
         }
 
+        private class ForZeroContentLength : MessageBody
+        {
+            public ForZeroContentLength(bool keepAlive)
+                : base(null)
+            {
+                RequestKeepAlive = keepAlive;
+            }
+
+            protected override ValueTask<ArraySegment<byte>> PeekAsync(CancellationToken cancellationToken)
+            {
+                return new ValueTask<ArraySegment<byte>>();
+            }
+
+            protected override void OnConsumedBytes(int count)
+            {
+                if (count > 0)
+                {
+                    throw new InvalidDataException("Consuming non-existant data");
+                }
+            }
+        }
+
         private class ForContentLength : MessageBody
         {
             private readonly long _contentLength;