add connectionToken property from IConnection

src/SuperSocket.Connection/ConnectionBase.cs

@@ -9,13 +9,13 @@
 namespace SuperSocket.Connection
 {
     public abstract class ConnectionBase : IConnection
-    {        
+    {
         public abstract IAsyncEnumerable<TPackageInfo> RunAsync<TPackageInfo>(IPipelineFilter<TPackageInfo> pipelineFilter);
 
         public abstract ValueTask SendAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default);
 
         public abstract ValueTask SendAsync<TPackage>(IPackageEncoder<TPackage> packageEncoder, TPackage package, CancellationToken cancellationToken = default);
-        
+
         public abstract ValueTask SendAsync(Action<PipeWriter> write, CancellationToken cancellationToken = default);
 
         public bool IsClosed { get; private set; }
@@ -28,6 +28,8 @@ public abstract class ConnectionBase : IConnection
 
         public DateTimeOffset LastActiveTime { get; protected set; } = DateTimeOffset.Now;
 
+        public CancellationToken ConnectionToken { get; protected set; }
+
         protected virtual void OnClosed()
         {
             IsClosed = true;

src/SuperSocket.Connection/IConnection.cs

@@ -33,5 +33,7 @@ public interface IConnection
         ValueTask DetachAsync();
 
         CloseReason? CloseReason { get; }
+
+        CancellationToken ConnectionToken { get; }
     }
 }

src/SuperSocket.Connection/PipeConnectionBase.cs

@@ -32,7 +32,7 @@ PipeReader IPipeConnection.InputReader
         {
             get { return InputReader; }
         }
-        
+
         IPipelineFilter IPipeConnection.PipelineFilter
         {
             get { return _pipelineFilter; }
@@ -52,6 +52,7 @@ protected PipeConnectionBase(PipeReader inputReader, PipeWriter outputWriter, Co
             Logger = options.Logger;
             InputReader = inputReader;
             OutputWriter = outputWriter;
+            ConnectionToken = _cts.Token;
         }
 
         protected virtual Task StartTask<TPackageInfo>(IObjectPipe<TPackageInfo> packagePipe)
@@ -72,7 +73,7 @@ public async override IAsyncEnumerable<TPackageInfo> RunAsync<TPackageInfo>(IPip
 
             _packagePipe = packagePipe;
             _pipelineFilter = pipelineFilter;
-            
+
             _pipeTask = StartTask(packagePipe);
 
             _ = HandleClosing();
@@ -118,7 +119,7 @@ private async ValueTask HandleClosing()
                     {
                         if (!IsIgnorableException(exc))
                             OnError("Unhandled exception in the method PipeChannel.Close.", exc);
-                    }                    
+                    }
                 }
             }
         }
@@ -172,7 +173,7 @@ public override async ValueTask SendAsync(ReadOnlyMemory<byte> buffer, Cancellat
             finally
             {
                 SendLock.Release();
-            }            
+            }
         }
 
         private void WriteBuffer(PipeWriter writer, ReadOnlyMemory<byte> buffer)
@@ -236,7 +237,7 @@ protected async Task ReadPipeAsync<TPackageInfo>(PipeReader reader, IObjectPipe<
                 {
                     if (!IsIgnorableException(e) && !(e is OperationCanceledException))
                         OnError("Failed to read from the pipe", e);
-                    
+
                     break;
                 }
 
@@ -267,7 +268,7 @@ protected async Task ReadPipeAsync<TPackageInfo>(PipeReader reader, IObjectPipe<
                         {
                             completed = true;
                             break;
-                        }                        
+                        }
                     }
 
                     if (completed)
@@ -344,7 +345,7 @@ private bool ReaderBuffer<TPackageInfo>(ref ReadOnlySequence<byte> buffer, IPipe
                     Close();
                     return false;
                 }
-                
+
                 if (packageInfo == null)
                 {
                     // the current pipeline filter needs more data to process
@@ -370,12 +371,12 @@ private bool ReaderBuffer<TPackageInfo>(ref ReadOnlySequence<byte> buffer, IPipe
                     examined = consumed = buffer.End;
                     return true;
                 }
-                
+
                 if (bytesConsumed > 0)
                     seqReader = new SequenceReader<byte>(seqReader.Sequence.Slice(bytesConsumed));
             }
         }
-    
+
         public override async ValueTask DetachAsync()
         {
             _isDetaching = true;

src/SuperSocket.Kestrel/KestrelPipeConnection.cs

@@ -16,7 +16,7 @@ public KestrelPipeConnection(ConnectionContext context, ConnectionOptions option
         : base(context.Transport.Input, context.Transport.Output, options)
     {
         _context = context;
-        context.ConnectionClosed.Register(() => OnClosed());
+        context.ConnectionClosed.Register(() => OnConnectionClosed());
         LocalEndPoint = context.LocalEndPoint;
         RemoteEndPoint = context.RemoteEndPoint;
     }
@@ -39,14 +39,6 @@ protected override async void Close()
         }
     }
 
-    protected override void OnClosed()
-    {
-        if (!CloseReason.HasValue)
-            CloseReason = Connection.CloseReason.RemoteClosing;
-
-        base.OnClosed();
-    }
-
     protected override void OnInputPipeRead(ReadResult result)
     {
         if (!result.IsCanceled && !result.IsCompleted)
@@ -72,4 +64,12 @@ public override async ValueTask SendAsync<TPackage>(IPackageEncoder<TPackage> pa
         await base.SendAsync(packageEncoder, package, cancellationToken);
         UpdateLastActiveTime();
     }
+
+    private void OnConnectionClosed()
+    {
+        if (!CloseReason.HasValue)
+            CloseReason = Connection.CloseReason.RemoteClosing;
+
+        Cancel();
+    }
 }

src/SuperSocket.Server/SuperSocketService.cs

@@ -314,11 +314,11 @@ protected virtual ValueTask OnSessionClosedAsync(IAppSession session, CloseEvent
             if (closedHandler != null)
                 return closedHandler.Invoke(session, e);
 
-            #if NETSTANDARD2_1
-                return GetCompletedTask();
-            #else
-                return ValueTask.CompletedTask;
-            #endif
+#if NETSTANDARD2_1
+            return GetCompletedTask();
+#else
+            return ValueTask.CompletedTask;
+#endif
         }
 
         protected virtual async ValueTask FireSessionConnectedEvent(AppSession session)
@@ -350,7 +350,7 @@ protected virtual async ValueTask FireSessionClosedEvent(AppSession session, Clo
                 if (!handshakeSession.Handshaked)
                     return;
             }
-            
+
             await UnRegisterSessionFromMiddlewares(session);
 
             _logger.LogInformation($"The session disconnected: {session.SessionID} ({reason})");
@@ -396,18 +396,18 @@ private async ValueTask HandleSession(AppSession session, IConnection connection
                 var packageHandlingScheduler = _packageHandlingScheduler;
 
 #if NET6_0_OR_GREATER
-                using var cancellationTokenSource = GetPackageHandlingCancellationTokenSource(CancellationToken.None);
+                using var cancellationTokenSource = GetPackageHandlingCancellationTokenSource(connection.ConnectionToken);
 #endif
 
                 await foreach (var p in packageStream)
                 {
-                    if(_packageHandlingContextAccessor != null)
+                    if (_packageHandlingContextAccessor != null)
                     {
                         _packageHandlingContextAccessor.PackageHandlingContext = new PackageHandlingContext<IAppSession, TReceivePackageInfo>(session, p);
                     }
 
 #if !NET6_0_OR_GREATER
-                    using var cancellationTokenSource = GetPackageHandlingCancellationTokenSource(CancellationToken.None);
+                    using var cancellationTokenSource = GetPackageHandlingCancellationTokenSource(connection.ConnectionToken);
 #endif
                     await packageHandlingScheduler.HandlePackage(session, p, cancellationTokenSource.Token);
 
@@ -424,13 +424,9 @@ private async ValueTask HandleSession(AppSession session, IConnection connection
 
         protected virtual CancellationTokenSource GetPackageHandlingCancellationTokenSource(CancellationToken cancellationToken)
         {
-#if NET6_0_OR_GREATER
-            return CancellationTokenSourcePool.Shared.Rent(TimeSpan.FromSeconds(Options.PackageHandlingTimeOut));
-#else
             var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
             cancellationTokenSource.CancelAfter(TimeSpan.FromSeconds(Options.PackageHandlingTimeOut));
             return cancellationTokenSource;
-#endif            
         }
 
         protected virtual ValueTask<bool> OnSessionErrorAsync(IAppSession session, PackageHandlingException<TReceivePackageInfo> exception)
@@ -471,20 +467,20 @@ public async Task StartAsync(CancellationToken cancellationToken)
 
         protected virtual ValueTask OnStartedAsync()
         {
-            #if NETSTANDARD2_1
-                return GetCompletedTask();
-            #else
-                return ValueTask.CompletedTask;
-            #endif
+#if NETSTANDARD2_1
+            return GetCompletedTask();
+#else
+            return ValueTask.CompletedTask;
+#endif
         }
 
         protected virtual ValueTask OnStopAsync()
         {
-            #if NETSTANDARD2_1
-                return GetCompletedTask();
-            #else
-                return ValueTask.CompletedTask;
-            #endif
+#if NETSTANDARD2_1
+            return GetCompletedTask();
+#else
+            return ValueTask.CompletedTask;
+#endif
         }
 
         private async Task StopListener(IConnectionListener listener)