#373 Replace ICurrentTimeProvider with TimeProvider

Signed-off-by: Richard Pringle <richardpringle@gmail.com>

Author: EtherZa

src/SlimMessageBus.Host.AmazonSQS/SqsMessageBus.cs

@@ -75,7 +75,7 @@ void AddConsumerFrom(string path, PathKind pathKind, IMessageProcessor<Message>
                 Settings.RequestResponse,
                 messageProvider: MessageProvider,
                 PendingRequestStore,
-                CurrentTimeProvider);
+                TimeProvider);
 
             AddConsumerFrom(
                 path,

src/SlimMessageBus.Host.AzureEventHub/Consumer/EhPartitionConsumerForResponses.cs

@@ -15,12 +15,12 @@ public EhPartitionConsumerForResponses(
         GroupPathPartitionId pathGroupPartition,
         MessageProvider<EventData> messageProvider,
         IPendingRequestStore pendingRequestStore,
-        ICurrentTimeProvider currentTimeProvider)
+        TimeProvider timeProvider)
         : base(messageBus, pathGroupPartition)
     {
         if (requestResponseSettings == null) throw new ArgumentNullException(nameof(requestResponseSettings));
 
-        MessageProcessor = new ResponseMessageProcessor<EventData>(MessageBus.LoggerFactory, requestResponseSettings, messageProvider, pendingRequestStore, currentTimeProvider);
+        MessageProcessor = new ResponseMessageProcessor<EventData>(MessageBus.LoggerFactory, requestResponseSettings, messageProvider, pendingRequestStore, timeProvider);
         CheckpointTrigger = new CheckpointTrigger(requestResponseSettings, MessageBus.LoggerFactory);
     }
 }

src/SlimMessageBus.Host.AzureEventHub/EventHubMessageBus.cs

@@ -72,9 +72,8 @@ protected override async Task CreateConsumers()
             object MessageProvider(Type messageType, EventData transportMessage) => messageSerializer.Deserialize(messageType, transportMessage.Body.ToArray());
 
             _logger.LogInformation("Creating response consumer for Path: {Path}, Group: {Group}", groupPath.Path, groupPath.Group);
-            AddConsumer(new EhGroupConsumer([Settings.RequestResponse], this, groupPath, groupPathPartition => new EhPartitionConsumerForResponses(this, Settings.RequestResponse, groupPathPartition, MessageProvider, PendingRequestStore, CurrentTimeProvider)));
+            AddConsumer(new EhGroupConsumer([Settings.RequestResponse], this, groupPath, groupPathPartition => new EhPartitionConsumerForResponses(this, Settings.RequestResponse, groupPathPartition, MessageProvider, PendingRequestStore, TimeProvider)));
         }
-
     }
 
     protected override async ValueTask DisposeAsyncCore()

src/SlimMessageBus.Host.AzureServiceBus/ServiceBusMessageBus.cs

@@ -120,7 +120,7 @@ void AddConsumerFrom(TopicSubscriptionParams topicSubscription, IMessageProcesso
                 Settings.RequestResponse,
                 MessageProvider,
                 PendingRequestStore,
-                CurrentTimeProvider);
+                TimeProvider);
 
             AddConsumerFrom(topicSubscription, messageProcessor, [Settings.RequestResponse]);
         }

src/SlimMessageBus.Host.Configuration/Settings/MessageBusSettings.cs

@@ -1,6 +1,6 @@
-namespace SlimMessageBus.Host;
+namespace SlimMessageBus.Host;
 
-public class MessageBusSettings : HasProviderExtensions
+public class MessageBusSettings : HasProviderExtensions
 {
     private IServiceProvider _serviceProvider;
     private readonly IList<MessageBusSettings> _children;
@@ -16,11 +16,11 @@ public IServiceProvider ServiceProvider
 
     /// <summary>
     /// The bus name.
-    /// </summary>
+    /// </summary>
     public string Name { get; set; }
-    public IList<ProducerSettings> Producers { get; }
-    public IList<ConsumerSettings> Consumers { get; }
-    public RequestResponseSettings RequestResponse { get; set; }
+    public IList<ProducerSettings> Producers { get; }
+    public IList<ConsumerSettings> Consumers { get; }
+    public RequestResponseSettings RequestResponse { get; set; }
     public Type SerializerType { get; set; }
     public Type MessageTypeResolverType { get; set; }
 
@@ -39,20 +39,20 @@ public IServiceProvider ServiceProvider
     /// </summary>
     public bool AutoStartConsumers { get; set; }
 
-    public MessageBusSettings(MessageBusSettings parent = null)
+    public MessageBusSettings(MessageBusSettings parent = null)
     {
         _children = [];
-        Producers = [];
-        Consumers = [];
+        Producers = [];
+        Consumers = [];
         SerializerType = typeof(IMessageSerializerProvider);
         AutoStartConsumers = true;
 
         if (parent != null)
         {
             Parent = parent;
             parent._children.Add(this);
-        }
-    }
+        }
+    }
 
     public virtual void MergeFrom(MessageBusSettings settings)
     {

src/SlimMessageBus.Host.Kafka/Consumer/KafkaPartitionConsumerForResponses.cs

@@ -16,7 +16,7 @@ public KafkaPartitionConsumerForResponses(
         IKafkaCommitController commitController,
         MessageProvider<ConsumeResult> messageProvider,
         IPendingRequestStore pendingRequestStore,
-        ICurrentTimeProvider currentTimeProvider,
+        TimeProvider timeProvider,
         IMessageSerializer headerSerializer)
         : base(
             loggerFactory,
@@ -30,7 +30,7 @@ public KafkaPartitionConsumerForResponses(
                 requestResponseSettings,
                 messageProvider,
                 pendingRequestStore,
-                currentTimeProvider))
+                timeProvider))
     {
     }
 }

src/SlimMessageBus.Host.Kafka/KafkaMessageBus.cs

@@ -76,7 +76,7 @@ IKafkaPartitionConsumer ResponseProcessorFactory(TopicPartition tp, IKafkaCommit
             var messageSerializer = SerializerProvider.GetSerializer(tp.Topic);
             object MessageProvider(Type messageType, ConsumeResult<Ignore, byte[]> transportMessage) => messageSerializer.Deserialize(messageType, transportMessage.Message.Value);
 
-            return new KafkaPartitionConsumerForResponses(LoggerFactory, Settings.RequestResponse, Settings.RequestResponse.GetGroup(), tp, cc, MessageProvider, PendingRequestStore, CurrentTimeProvider, headerSerializer);
+            return new KafkaPartitionConsumerForResponses(LoggerFactory, Settings.RequestResponse, Settings.RequestResponse.GetGroup(), tp, cc, MessageProvider, PendingRequestStore, TimeProvider, headerSerializer);
         }
 
         foreach (var consumersByGroup in Settings.Consumers.GroupBy(x => x.GetGroup()))

src/SlimMessageBus.Host.Mqtt/MqttMessageBus.cs

@@ -87,7 +87,7 @@ void AddTopicConsumer(IEnumerable<AbstractConsumerSettings> consumerSettings, st
                 Settings.RequestResponse,
                 messageProvider: MessageProvider,
                 PendingRequestStore,
-                CurrentTimeProvider);
+                TimeProvider);
 
             AddTopicConsumer([Settings.RequestResponse], path, processor);
         }

src/SlimMessageBus.Host.Nats/NatsMessageBus.cs

@@ -78,7 +78,7 @@ protected override async Task CreateConsumers()
             var messageSerializer = SerializerProvider.GetSerializer(subject);
             object MessageProvider(Type messageType, NatsMsg<byte[]> transportMessage) => messageSerializer.Deserialize(messageType, transportMessage.Data);
 
-            var processor = new ResponseMessageProcessor<NatsMsg<byte[]>>(LoggerFactory, Settings.RequestResponse, MessageProvider, PendingRequestStore, CurrentTimeProvider);
+            var processor = new ResponseMessageProcessor<NatsMsg<byte[]>>(LoggerFactory, Settings.RequestResponse, MessageProvider, PendingRequestStore, TimeProvider);
             AddSubjectConsumer([], subject, processor);
         }
     }

src/SlimMessageBus.Host.Outbox.Sql.DbContext/Configuration/MessageBusBuilderExtensions.cs

@@ -16,7 +16,7 @@ public static MessageBusBuilder AddOutboxUsingDbContext<TDbContext>(this Message
                         settings,
                         svp.GetRequiredService<SqlOutboxTemplate>(),
                         settings.IdGeneration.GuidGenerator ?? (IGuidGenerator)svp.GetRequiredService(settings.IdGeneration.GuidGeneratorType),
-                        svp.GetRequiredService<ICurrentTimeProvider>(),
+                        svp.GetRequiredService<TimeProvider>(),
                         svp.GetRequiredService<IInstanceIdProvider>(),
                         svp.GetRequiredService<TDbContext>(),
                         svp.GetRequiredService<ISqlTransactionService>()