Adjust the timing for verifying whether CAP is started. After CAP stops, failed messages will be stored in the storage to be retried later upon restart. (#1238)

Author: yang-xiaodong

src/DotNetCore.CAP/ICapTransaction.Base.cs

@@ -59,9 +59,7 @@ protected virtual async Task FlushAsync()
                 var isDelayMessage = message.Origin.Headers.ContainsKey(Headers.DelayTime);
                 if (isDelayMessage)
                 {
-
                     await _dispatcher.EnqueueToScheduler(message, DateTime.Parse(message.Origin.Headers[Headers.SentTime]!, CultureInfo.InvariantCulture)).ConfigureAwait(false);
-
                 }
                 else
                 {

src/DotNetCore.CAP/Internal/ICapPublisher.Default.cs

@@ -26,14 +26,12 @@ internal class CapPublisher : ICapPublisher
     private readonly ISnowflakeId _snowflakeId;
     private readonly IDispatcher _dispatcher;
     private readonly IDataStorage _storage;
-    private readonly IBootstrapper _bootstrapper;
 
     private readonly AsyncLocal<CapTransactionHolder> _asyncLocal;
 
     public CapPublisher(IServiceProvider service)
     {
         ServiceProvider = service;
-        _bootstrapper = service.GetRequiredService<IBootstrapper>();
         _dispatcher = service.GetRequiredService<IDispatcher>();
         _storage = service.GetRequiredService<IDataStorage>();
         _capOptions = service.GetRequiredService<IOptions<CapOptions>>().Value;
@@ -43,8 +41,8 @@ public CapPublisher(IServiceProvider service)
 
     public IServiceProvider ServiceProvider { get; }
 
-    public ICapTransaction? Transaction {
-
+    public ICapTransaction? Transaction
+    {
         get => _asyncLocal.Value?.Transaction;
         set
         {
@@ -114,11 +112,6 @@ public void PublishDelay<T>(TimeSpan delayTime, string name, T? value, string? c
     private async Task PublishInternalAsync<T>(string name, T? value, IDictionary<string, string?> headers, TimeSpan? delayTime = null,
         CancellationToken cancellationToken = default)
     {
-        if (!_bootstrapper.IsStarted)
-        {
-            throw new InvalidOperationException("CAP has not been started!");
-        }
-
         if (string.IsNullOrEmpty(name)) throw new ArgumentNullException(nameof(name));
 
         if (!string.IsNullOrEmpty(_capOptions.TopicNamePrefix)) name = $"{_capOptions.TopicNamePrefix}.{name}";
@@ -176,8 +169,7 @@ private async Task PublishInternalAsync<T>(string name, T? value, IDictionary<st
             {
                 var transaction = (CapTransactionBase)Transaction;
 
-                var mediumMessage = await _storage.StoreMessageAsync(name, message, transaction.DbTransaction)
-                    .ConfigureAwait(false);
+                var mediumMessage = await _storage.StoreMessageAsync(name, message, transaction.DbTransaction).ConfigureAwait(false);
 
                 TracingAfter(tracingTimestamp, message);
 

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

@@ -112,10 +112,10 @@ await Task.WhenAll(Enumerable.Range(0, _options.SubscriberParallelExecuteThreadC
                 }
                 catch (Exception ex)
                 {
-                    _logger.LogWarning(ex, 
+                    _logger.LogWarning(ex,
                         "Scheduled message publishing failed unexpectedly, which will stop future scheduled " +
                         "messages from publishing. See more details here: https://github.com/dotnetcore/CAP/issues/1637. " +
-                        "Exception: {Message}", 
+                        "Exception: {Message}",
                         ex.Message);
                     throw;
                 }
@@ -145,7 +145,11 @@ public async ValueTask EnqueueToPublish(MediumMessage message)
     {
         try
         {
-            if (_tasksCts!.IsCancellationRequested) return;
+            if (_tasksCts!.IsCancellationRequested)
+            {
+                _logger.LogWarning("The message has been persisted, but CAP is currently stopped. It will be attempted to be sent once CAP becomes available.");
+                return;
+            }
 
             if (_enableParallelSend && message.Retries == 0)
             {