#1757

Enhance disposal handling in Dispatcher & CapProcessingServer

Introduce `_disposed` flag in `Dispatcher` to track disposal state.
Update `Start` and `Dispose` methods to manage `CancellationTokenSource` lifecycle, allowing safe restart after disposal.
Modify `CapProcessingServer` to allow reassignment of `_cts` and implement similar disposal and recreation logic.
Ensure proper resource management and restart capability.

Author: yang-xiaodong

src/DotNetCore.CAP/Processor/IDispatcher.Default.cs

@@ -31,6 +31,7 @@ public class Dispatcher : IDispatcher
     private CancellationTokenSource? _tasksCts;
     private Channel<MediumMessage> _publishedChannel = default!;
     private Channel<(MediumMessage, ConsumerExecutorDescriptor?)> _receivedChannel = default!;
+    private bool _disposed;
 
     public Dispatcher(ILogger<Dispatcher> logger, IMessageSender sender, IOptions<CapOptions> options,
         ISubscribeExecutor executor, IDataStorage storage)
@@ -47,6 +48,14 @@ public Dispatcher(ILogger<Dispatcher> logger, IMessageSender sender, IOptions<Ca
 
     public async Task Start(CancellationToken stoppingToken)
     {
+        // If already disposed and restarting, recreate the CancellationTokenSource and reset state
+        if (_disposed || (_tasksCts != null && _tasksCts.IsCancellationRequested))
+        {
+            _tasksCts?.Dispose();
+            _tasksCts = null;
+            _disposed = false;
+        }
+
         stoppingToken.ThrowIfCancellationRequested();
         _tasksCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken, CancellationToken.None);
 
@@ -226,6 +235,8 @@ public async ValueTask EnqueueToExecute(MediumMessage message, ConsumerExecutorD
 
     public void Dispose()
     {
+        if (_disposed) return;
+        _disposed = true;
         _tasksCts?.Dispose();
         GC.SuppressFinalize(this);
     }

src/DotNetCore.CAP/Processor/IProcessingServer.Cap.cs

@@ -14,7 +14,7 @@ namespace DotNetCore.CAP.Processor;
 
 public class CapProcessingServer : IProcessingServer
 {
-    private readonly CancellationTokenSource _cts;
+    private CancellationTokenSource _cts;
     private readonly ILogger _logger;
     private readonly ILoggerFactory _loggerFactory;
     private readonly IServiceProvider _provider;
@@ -36,6 +36,14 @@ public CapProcessingServer(
 
     public Task Start(CancellationToken stoppingToken)
     {
+        // If already disposed and restarting, recreate the CancellationTokenSource
+        if (_disposed || _cts.IsCancellationRequested)
+        {
+            _cts?.Dispose();
+            _cts = new CancellationTokenSource();
+            _disposed = false;
+        }
+
         stoppingToken.Register(() => _cts.Cancel());
 
         _logger.ServerStarting();