Add support to scheduler message

docs/adr/0024-Scheduling.md

@@ -0,0 +1,89 @@
+# 24. Scoping dependencies inline with lifetime scope
+
+Date: 2025-01-20
+
+## Status
+
+Proposed
+
+## Context
+
+Adding the ability to schedule message (by providing `TimeSpan` or `DateTimeOffset`) give to user flexibility to `Send`, `Publis` and `Post`.
+
+
+
+## Decision
+
+Giving support to schedule message, it's necessary breaking on `IAmACommandProcessor` by adding these methods:
+
+```c#
+public interface IAmACommandProcessor
+{
+    string SchedulerSend<TRequest>(TimeSpan delay, TRequest request) where TRequest : class, IRequest;
+    string SchedulerSend<TRequest>(DateTimeOffset delay, TRequest request) where TRequest : class, IRequest;
+    Task<string> SchedulerSendAsync<TRequest>(TimeSpan delay, TRequest request, bool continueOnCapturedContext = true, CancellationToken cancellationToken = default) where TRequest : class, IRequest;
+    Task<string> SchedulerSendAsync<TRequest>(DateTimeOffset delay, TRequest request, bool continueOnCapturedContext = true, CancellationToken cancellationToken = default) where TRequest : class, IRequest;
+
+    string SchedulerPublish<TRequest>(TimeSpan delay, TRequest request) where TRequest : class, IRequest;
+    string SchedulerPublish<TRequest>(DateTimeOffset delay, TRequest request) where TRequest : class, IRequest;
+    Task<string> SchedulerPublishAsync<TRequest>(TimeSpan delay, TRequest request, bool continueOnCapturedContext = true, CancellationToken cancellationToken = default) where TRequest : class, IRequest;
+    Task<string> SchedulerPublishsync<TRequest>(DateTimeOffset delay, TRequest request, bool continueOnCapturedContext = true, CancellationToken cancellationToken = default) where TRequest : class, IRequest;
+
+    string SchedulerPost<TRequest>(TimeSpan delay, TRequest request) where TRequest : class, IRequest;
+    string SchedulerPost<TRequest>(DateTimeOffset delay, TRequest request) where TRequest : class, IRequest;
+    Task<string> SchedulerPostAsync<TRequest>(TimeSpan delay, TRequest request, bool continueOnCapturedContext = true, CancellationToken cancellationToken = default) where TRequest : class, IRequest;
+    Task<string> SchedulerPostAsync<TRequest>(DateTimeOffset delay, TRequest request, bool continueOnCapturedContext = true, CancellationToken cancellationToken = default) where TRequest : class, IRequest;
+}
+```
+
+Scheduling can be break into 2 part (Producer & Consumer):
+- Producer -> Producing a message we are going to have a new interface:
+
+```c#
+public interface IAmAMessageScheduler
+{
+}
+
+
+public interface IAmAMessageSchedulerAsync : IAmAMessageScheduler, IDisposable
+{
+    Task<string> ScheduleAsync<TRequest>(DateTimeOffset at, SchedulerFireType fireType, TRequest request, CancellationToken cancellationToken = default) where TRequest : class, IRequest;
+    Task<string> ScheduleAsync<TRequest>(TimeSpan delay, SchedulerFireType fireType, TRequest request, CancellationToken cancellationToken = default) where TRequest : class, IRequest;
+    Task CancelSchedulerAsync(string id, CancellationToken cancellationToken = default);
+}
+
+
+public interface IAmAMessageSchedulerSync : IAmAMessageScheduler, IDisposable
+{
+    string Schedule<TRequest>(DateTimeOffset at, SchedulerFireType fireType, TRequest request) where TRequest : class, IRequest;
+    string Schedule<TRequest>(TimeSpan delay, SchedulerFireType fireType, TRequest request) where TRequest : class, IRequest;
+    void CancelScheduler(string id);
+}
+```
+
+- Consumer -> To avoid duplication code we are going to introduce a new message and have a handler for that:
+
+```c#
+public class SchedulerMessageFired : Event
+{
+    .....
+}
+
+
+public class SchedulerMessageFiredHandlerAsync(IAmACommandProcessor processor) : RequestHandlerAsync<SchedulerMessageFired>
+{
+    ....
+}
+```
+
+So on Scheduler implementation we need to send the SchedulerMessageFired
+
+```c#
+public class JobExecute(IAmACommandProcessor processor)
+{
+    public async Task ExecuteAsync(Arg arg)
+    {
+        await processor.SendAsync(new SchedulerMessageFired{ ... });
+    }
+}
+```

samples/TaskQueue/AWSTaskQueue/GreetingsSender/Program.cs

@@ -75,7 +75,9 @@ static void Main(string[] args)
                     }
                 ).Create();
 
-                serviceCollection.AddBrighter()
+                serviceCollection
+                    .AddSingleton<IAmAMessageSchedulerFactory>(new InMemoryMessageSchedulerFactory())
+                    .AddBrighter()
                     .UseExternalBus((configure) =>
                     {
                         configure.ProducerRegistry = producerRegistry;
@@ -84,9 +86,23 @@ static void Main(string[] args)
 
                 var serviceProvider = serviceCollection.BuildServiceProvider();
 
-                var commandProcessor = serviceProvider.GetService<IAmACommandProcessor>();
+                var commandProcessor = serviceProvider.GetRequiredService<IAmACommandProcessor>();
 
                 commandProcessor.Post(new GreetingEvent("Ian says: Hi there!"));
+
+                // TODO Remove this code:
+                while (true)
+                {
+                    Console.WriteLine("Enter a name to greet (Q to quit):");
+                    var name = Console.ReadLine();
+                    if (name is "Q" or "q")
+                    {
+                        break;
+                    }
+
+                    commandProcessor.SchedulerPost(TimeSpan.FromSeconds(10), new GreetingEvent($"Ian says: Hi {name}"));
+                }
+
                 commandProcessor.Post(new FarewellEvent("Ian says: See you later!"));
             }
         }

samples/TaskQueue/RMQTaskQueue/GreetingsSender/Program.cs

@@ -74,7 +74,9 @@ static void Main(string[] args)
                     }
                 }).Create();
 
-            serviceCollection.AddBrighter()
+            serviceCollection
+                .AddSingleton<IAmAMessageSchedulerFactory>(new InMemoryMessageSchedulerFactory())
+                .AddBrighter()
                 .UseExternalBus((configure) =>
                 {
                     configure.ProducerRegistry = producerRegistry;
@@ -88,6 +90,20 @@ static void Main(string[] args)
             var commandProcessor = serviceProvider.GetService<IAmACommandProcessor>();
 
             commandProcessor.Post(new GreetingEvent("Ian says: Hi there!"));
+
+            // TODO Remove this code:
+            while (true)
+            {
+                Console.WriteLine("Enter a name to greet (Q to quit):");
+                var name = Console.ReadLine();
+                if (name is "Q" or "q")
+                {
+                    break;
+                }
+
+                commandProcessor.SchedulerPost(TimeSpan.FromSeconds(60), new GreetingEvent($"Ian says: Hi {name}"));
+            }
+
             commandProcessor.Post(new FarewellEvent("Ian says: See you later!"));
         }
     }

src/Paramore.Brighter.Extensions.DependencyInjection/ServiceCollectionExtensions.cs

@@ -315,6 +315,10 @@ private static object BuildCommandProcessor(IServiceProvider provider)
             var requestContextFactory = provider.GetService<IAmARequestContextFactory>();
 
             var builder = contextBuilder.RequestContextFactory(requestContextFactory);
+
+            var schedulerMessageFactory = provider.GetService<IAmAMessageSchedulerFactory>();
+
+            builder.MessageSchedulerFactory(schedulerMessageFactory);
 
             var commandProcessor = builder.Build();
 

src/Paramore.Brighter.MessagingGateway.AWSSQS/SnsMessagePublisher.cs

@@ -50,6 +50,11 @@ public SnsMessagePublisher(string topicArn, AmazonSimpleNotificationServiceClien
         var messageString = message.Body.Value;
         var publishRequest = new PublishRequest(_topicArn, messageString, message.Header.Subject);
 
+        if (string.IsNullOrEmpty(message.Header.CorrelationId))
+        {
+            message.Header.CorrelationId = Guid.NewGuid().ToString();
+        }
+
         var messageAttributes = new Dictionary<string, MessageAttributeValue>
         {
             [HeaderNames.Id] =

src/Paramore.Brighter.ServiceActivator/ControlBus/ControlBusReceiverBuilder.cs

@@ -26,6 +26,8 @@ THE SOFTWARE. */
 using System.Collections.Generic;
 using System.Transactions;
 using Paramore.Brighter.Observability;
+using Paramore.Brighter.Scheduler.Events;
+using Paramore.Brighter.Scheduler.Handlers;
 using Paramore.Brighter.ServiceActivator.Ports;
 using Paramore.Brighter.ServiceActivator.Ports.Commands;
 using Paramore.Brighter.ServiceActivator.Ports.Handlers;
@@ -139,6 +141,7 @@ public Dispatcher Build(string hostName)
             var subscriberRegistry = new SubscriberRegistry();
             subscriberRegistry.Register<ConfigurationCommand, ConfigurationCommandHandler>();
             subscriberRegistry.Register<HeartbeatRequest, HeartbeatRequestCommandHandler>();
+            subscriberRegistry.RegisterAsync<SchedulerMessageFired, SchedulerMessageFiredHandlerAsync>();
 
             var incomingMessageMapperRegistry = new MessageMapperRegistry(
             new ControlBusMessageMapperFactory(), null

src/Paramore.Brighter/CommandProcessor.cs

@@ -37,6 +37,8 @@ THE SOFTWARE. */
 using Paramore.Brighter.FeatureSwitch;
 using Paramore.Brighter.Logging;
 using Paramore.Brighter.Observability;
+using Paramore.Brighter.Scheduler.Events;
+using Paramore.Brighter.Tasks;
 using Polly;
 using Polly.Registry;
 using Exception = System.Exception;
@@ -61,6 +63,7 @@ public class CommandProcessor : IAmACommandProcessor
         private readonly IAmAFeatureSwitchRegistry? _featureSwitchRegistry;
         private readonly IEnumerable<Subscription>? _replySubscriptions;
         private readonly IAmABrighterTracer? _tracer;
+        private readonly IAmAMessageSchedulerFactory? _messageSchedulerFactory;
 
         //Uses -1 to indicate no outbox and will thus force a throw on a failed publish
 
@@ -117,6 +120,7 @@ public class CommandProcessor : IAmACommandProcessor
         /// <param name="inboxConfiguration">Do we want to insert an inbox handler into pipelines without the attribute. Null (default = no), yes = how to configure</param>
         /// <param name="tracer">What is the tracer we will use for telemetry</param>
         /// <param name="instrumentationOptions">When creating a span for <see cref="CommandProcessor"/> operations how noisy should the attributes be</param>
+        /// <param name="messageSchedulerFactory">TODO: ADD description </param>
         public CommandProcessor(
             IAmASubscriberRegistry subscriberRegistry,
             IAmAHandlerFactory handlerFactory,
@@ -125,7 +129,8 @@ public CommandProcessor(
             IAmAFeatureSwitchRegistry? featureSwitchRegistry = null,
             InboxConfiguration? inboxConfiguration = null,
             IAmABrighterTracer? tracer = null,
-            InstrumentationOptions instrumentationOptions = InstrumentationOptions.All)
+            InstrumentationOptions instrumentationOptions = InstrumentationOptions.All,
+            IAmAMessageSchedulerFactory? messageSchedulerFactory = null)
         {
             _subscriberRegistry = subscriberRegistry;
 
@@ -144,6 +149,7 @@ public CommandProcessor(
             _inboxConfiguration = inboxConfiguration;
             _tracer = tracer;
             _instrumentationOptions = instrumentationOptions;
+            _messageSchedulerFactory = messageSchedulerFactory;
         }
 
         /// <summary>
@@ -162,6 +168,7 @@ public CommandProcessor(
         /// <param name="responseChannelFactory">If we are expecting a response, then we need a channel to listen on</param>
         /// <param name="tracer">What is the tracer we will use for telemetry</param>
         /// <param name="instrumentationOptions">When creating a span for <see cref="CommandProcessor"/> operations how noisy should the attributes be</param>
+        /// <param name="messageSchedulerFactory">TODO: ADD description </param>
         public CommandProcessor(
             IAmASubscriberRegistry subscriberRegistry,
             IAmAHandlerFactory handlerFactory,
@@ -173,14 +180,15 @@ public CommandProcessor(
             IEnumerable<Subscription>? replySubscriptions = null,
             IAmAChannelFactory? responseChannelFactory = null,
             IAmABrighterTracer? tracer = null,
-            InstrumentationOptions instrumentationOptions = InstrumentationOptions.All)
-            : this(subscriberRegistry, handlerFactory, requestContextFactory, policyRegistry, featureSwitchRegistry, inboxConfiguration)
+            InstrumentationOptions instrumentationOptions = InstrumentationOptions.All,
+            IAmAMessageSchedulerFactory? messageSchedulerFactory = null)
+            : this(subscriberRegistry, handlerFactory, requestContextFactory, policyRegistry, featureSwitchRegistry, inboxConfiguration, messageSchedulerFactory: messageSchedulerFactory)
         {
             _responseChannelFactory = responseChannelFactory;
             _tracer = tracer;
             _instrumentationOptions = instrumentationOptions;
             _replySubscriptions = replySubscriptions;
-
+            
             InitExtServiceBus(bus); 
         }
 
@@ -217,6 +225,102 @@ public CommandProcessor(
             InitExtServiceBus(mediator); 
         }
 
+        /// <inheritdoc />
+        public void SchedulerPost<TRequest>(TimeSpan delay, TRequest request, RequestContext? requestContext = null) 
+            where TRequest : class, IRequest
+        {
+           if (_messageSchedulerFactory == null)
+           {
+               throw new InvalidOperationException("No message scheduler factory set.");
+           }
+
+           s_logger.LogInformation("Scheduling request: {RequestType} {Id}", request.GetType(), request.Id);
+           var scheduler = _messageSchedulerFactory.Create(this);
+           if (scheduler is IAmAMessageSchedulerSync sync)
+           {
+               sync.Schedule(delay, SchedulerFireType.Post, request);
+           }
+           else if (scheduler is IAmAMessageSchedulerAsync asyncScheduler)
+           {
+               BrighterAsyncContext.Run(async () => await asyncScheduler.ScheduleAsync(delay, SchedulerFireType.Post, request));
+           }
+        }
+
+        /// <inheritdoc />
+        public void SchedulerPost<TRequest>(DateTimeOffset at,
+            TRequest request,
+            RequestContext? requestContext = null)
+            where TRequest : class, IRequest
+        {
+            if (_messageSchedulerFactory == null) 
+            {
+                throw new InvalidOperationException("No message scheduler factory set.");
+            }
+
+            s_logger.LogInformation("Scheduling request: {RequestType} {Id}", request.GetType(), request.Id);
+            var scheduler = _messageSchedulerFactory.Create(this);
+            if (scheduler is IAmAMessageSchedulerSync sync)
+            {
+                sync.Schedule(at, SchedulerFireType.Post, request);
+            }
+            else if (scheduler is IAmAMessageSchedulerAsync asyncScheduler)
+            {
+                BrighterAsyncContext.Run(async () => await asyncScheduler.ScheduleAsync(at, SchedulerFireType.Post, request));
+            }
+        }
+
+
+        /// <inheritdoc />
+        public async Task SchedulerAsync<TRequest>(TimeSpan delay,
+            TRequest request,
+            RequestContext? requestContext = null,
+            bool continueOnCapturedContext = true,
+            CancellationToken cancellationToken = default) 
+            where TRequest : class, IRequest
+        { 
+            if (_messageSchedulerFactory == null)
+            {
+                throw new InvalidOperationException("No message scheduler factory set.");
+            }
+
+            s_logger.LogInformation("Scheduling request: {RequestType} {Id}", request.GetType(), request.Id);
+            var scheduler = _messageSchedulerFactory.Create(this);
+            if (scheduler is IAmAMessageSchedulerAsync asyncScheduler)
+            {
+                await asyncScheduler.ScheduleAsync(delay, SchedulerFireType.Post, request, cancellationToken);
+            }
+            else if (scheduler is IAmAMessageSchedulerSync sync)
+            {
+                sync.Schedule(delay, SchedulerFireType.Post, request);
+            }
+        }
+
+        /// <inheritdoc />
+        public async Task SchedulerAsync<TRequest>(DateTimeOffset at,
+            TRequest request, 
+            RequestContext? requestContext = null,
+            bool continueOnCapturedContext = true, 
+            CancellationToken cancellationToken = default) 
+            where TRequest : class, IRequest
+        {
+            if (_messageSchedulerFactory == null)
+            {
+                throw new InvalidOperationException("No message scheduler factory set.");
+            }
+
+            s_logger.LogInformation("Scheduling request: {RequestType} {Id}", request.GetType(), request.Id);
+            var scheduler = _messageSchedulerFactory.Create(this);
+            if (scheduler is IAmAMessageSchedulerAsync asyncScheduler)
+            {
+                await asyncScheduler.ScheduleAsync(at, SchedulerFireType.Post, request, cancellationToken);
+            }
+            else if (scheduler is IAmAMessageSchedulerSync sync)
+            {
+                sync.Schedule(at, SchedulerFireType.Post, request);
+            }
+        }
+
+
         /// <summary>
         /// Sends the specified command. We expect only one handler. The command is handled synchronously.
         /// </summary>