Try making SuperSocket.Kestrel support Detach

Author: wj8400684

src/SuperSocket.Connection/IObjectPipe.cs

@@ -20,10 +20,10 @@ public interface IObjectPipe<T> : IObjectPipe
         ValueTask<T> ReadAsync();
     }
 
-    interface ISupplyController
+    public interface ISupplyController
     {
         ValueTask SupplyRequired();
 
         void SupplyEnd();
     }
-}
+}

src/SuperSocket.Connection/PipeConnectionBase.cs

@@ -47,8 +47,6 @@ IPipelineFilter IPipeConnection.PipelineFilter
 
         private bool _isDetaching = false;
 
-        private ISupplyController _supplyController;
-
         protected PipeConnectionBase(PipeReader inputReader, PipeWriter outputWriter, ConnectionOptions options)
         {
             Options = options;
@@ -70,17 +68,9 @@ protected void UpdateLastActiveTime()
 
         public async override IAsyncEnumerable<TPackageInfo> RunAsync<TPackageInfo>(IPipelineFilter<TPackageInfo> pipelineFilter)
         {
-            IObjectPipe<TPackageInfo> packagePipe;
-            if (Options.ReadAsDemand)
-            {
-                var defaultObjectPipe = new DefaultObjectPipeWithSupplyControl<TPackageInfo>();
-                _supplyController = defaultObjectPipe;
-                packagePipe = defaultObjectPipe;
-            }
-            else
-            {
-                packagePipe = new DefaultObjectPipe<TPackageInfo>();
-            }
+            var packagePipe = !Options.ReadAsDemand
+                ? new DefaultObjectPipe<TPackageInfo>()
+                : new DefaultObjectPipeWithSupplyControl<TPackageInfo>();
 
             _packagePipe = packagePipe;
             _pipelineFilter = pipelineFilter;
@@ -423,16 +413,5 @@ protected void OnError(string message, Exception e = null)
             else
                 Logger?.LogError(message);
         }
-
-        protected void SupplyEnd()
-        {
-            _supplyController?.SupplyEnd();
-        }
-
-        protected async Task SupplyRequiredAsync()
-        {
-            if (_supplyController != null)
-                await _supplyController.SupplyRequired().ConfigureAwait(false);
-        }
     }
 }

src/SuperSocket.Kestrel/KestrelPipeConnection.cs

@@ -13,6 +13,7 @@
 public class KestrelPipeConnection : PipeConnectionBase
 {
     private ConnectionContext _context;
+    private ISupplyController _supplyController;
 
     public KestrelPipeConnection(ConnectionContext context, ConnectionOptions options)
         : base(context.Transport.Input, context.Transport.Output, options)
@@ -49,7 +50,8 @@ protected override async Task OnInputPipeReadAsync(ReadResult result)
         if (result is { IsCanceled: false, IsCompleted: false })
             UpdateLastActiveTime();
 
-        await SupplyRequiredAsync();
+        if (_supplyController != null)
+            await _supplyController.SupplyRequired().ConfigureAwait(false);
     }
 
     public override async ValueTask SendAsync(Action<PipeWriter> write, CancellationToken cancellationToken)
@@ -64,7 +66,8 @@ public override async ValueTask SendAsync(ReadOnlyMemory<byte> buffer, Cancellat
         UpdateLastActiveTime();
     }
 
-    public override async ValueTask SendAsync<TPackage>(IPackageEncoder<TPackage> packageEncoder, TPackage package, CancellationToken cancellationToken)
+    public override async ValueTask SendAsync<TPackage>(IPackageEncoder<TPackage> packageEncoder, TPackage package,
+        CancellationToken cancellationToken)
     {
         await base.SendAsync(packageEncoder, package, cancellationToken);
         UpdateLastActiveTime();
@@ -93,7 +96,11 @@ private void OnConnectionClosed()
     protected override Task StartInputPipeTask<TPackageInfo>(IObjectPipe<TPackageInfo> packagePipe,
         CancellationToken cancellationToken)
     {
-        cancellationToken.Register(SupplyEnd);
+        _supplyController = packagePipe as ISupplyController;
+        
+        if (_supplyController != null)
+            cancellationToken.Register(() => _supplyController.SupplyEnd());
+
         return base.StartInputPipeTask(packagePipe, cancellationToken);
     }
-}
+}