Merge pull request #68 from wemogy/feature/renew-sesssion-lock

Introduce renew session lock interval

Author: SebastianKuesters

src/extensions/azureServiceBus/Wemogy.CQRS.Extensions.AzureServiceBus/Processors/AzureServiceBusCommandSessionProcessor`1.cs

@@ -26,9 +26,13 @@ public class AzureServiceBusCommandSessionProcessor<TCommand> : IAzureServiceBus
         /// </summary>
         public bool IsAlive => _isStarted && !_serviceBusSessionProcessor.IsClosed;
 
+        private readonly string _handleMessageActivityName;
+        private readonly TimeSpan _renewSessionLockInterval;
+
         public AzureServiceBusCommandSessionProcessor(
             ServiceBusSessionProcessor serviceBusSessionProcessor,
-            IServiceCollection serviceCollection)
+            IServiceCollection serviceCollection,
+            TimeSpan renewSessionLockInterval)
         {
             _serviceBusSessionProcessor = serviceBusSessionProcessor;
             _serviceCollection = serviceCollection;
@@ -41,10 +45,13 @@ public AzureServiceBusCommandSessionProcessor(
             _scheduledCommandDependencies = serviceCollection
                 .BuildServiceProvider()
                 .GetRequiredService<ScheduledCommandDependencies>();
+            _handleMessageActivityName = $"HandleMessageOf{typeof(TCommand).Name}";
+            _renewSessionLockInterval = renewSessionLockInterval;
         }
 
         public async Task HandleMessageAsync(ProcessSessionMessageEventArgs arg)
         {
+            using var activity = Observability.DefaultActivities.StartActivity(_handleMessageActivityName);
             var services = new ServiceCollection();
             foreach (var serviceDescriptor in _serviceCollection)
             {
@@ -80,18 +87,44 @@ public async Task HandleMessageAsync(ProcessSessionMessageEventArgs arg)
             var scopeFactory = services.BuildServiceProvider().GetRequiredService<IServiceScopeFactory>();
             var scope = scopeFactory.CreateScope();
 
+            var renewSessionLockCancellationTokenSource = new CancellationTokenSource();
+            var renewSessionLockCancellationToken = renewSessionLockCancellationTokenSource.Token;
+            Task? renewSessionLockTask = null;
             try
             {
-                var scheduledCommandRunner = scope.ServiceProvider.GetRequiredService<IScheduledCommandRunner<TCommand>>();
+                var scheduledCommandRunner =
+                    scope.ServiceProvider.GetRequiredService<IScheduledCommandRunner<TCommand>>();
+                var scheduledCommandRunnerTask = scheduledCommandRunner.RunAsync(scheduledCommand);
 
-                await scheduledCommandRunner.RunAsync(scheduledCommand);
+                // renew session lock every 30 seconds
+                renewSessionLockTask = Task.Run(
+                    async () =>
+                    {
+                        while (!renewSessionLockCancellationToken.IsCancellationRequested &&
+                                !scheduledCommandRunnerTask.IsCompleted)
+                        {
+                            await Task.Delay(_renewSessionLockInterval, renewSessionLockCancellationToken);
+                            Console.WriteLine($"Renewing session lock for session{arg.SessionId}...");
+                            await arg.RenewSessionLockAsync(arg.CancellationToken);
+                            Console.WriteLine($"Renewed session lock for session {arg.SessionId}");
+                        }
+                    },
+                    renewSessionLockCancellationTokenSource.Token);
+
+                await scheduledCommandRunnerTask;
+                renewSessionLockCancellationTokenSource.Cancel();
             }
             catch (Exception e)
             {
                 // ToDo: Dead letter message ==> Maybe remove try/catch let AutoComplete manage this
                 Console.WriteLine(e);
                 throw;
             }
+            finally
+            {
+                renewSessionLockCancellationTokenSource.Dispose();
+                renewSessionLockTask?.Dispose();
+            }
         }
 
         public async Task StartAsync(CancellationToken cancellationToken)

src/extensions/azureServiceBus/Wemogy.CQRS.Extensions.AzureServiceBus/Setup/AzureServiceBusSetupEnvironment.cs

@@ -53,10 +53,15 @@ public AzureServiceBusSetupEnvironment AddDelayedProcessor<TCommand>(
         /// <summary>
         /// Creates a ServiceBusProcessor and subscribes to messages of type <typeparamref name="TCommand"/>
         /// </summary>
+        /// <param name="maxConcurrentSessions">The maximum number of concurrent sessions (default 1)</param>
+        /// <param name="maxConcurrentCallsPerSession">The maximum number of concurrent calls per session (default 1)</param>
+        /// <param name="configureSessionProcessorOptions">Optional custom configuration of the ServiceBusSessionProcessorOptions</param>
+        /// <param name="renewSessionLockInterval">The interval to renew the session lock (default 1 minute)</param>
         public AzureServiceBusSetupEnvironment AddDelayedSessionProcessor<TCommand>(
             int maxConcurrentSessions = 1,
             int maxConcurrentCallsPerSession = 1,
-            Action<ServiceBusSessionProcessorOptions>? configureSessionProcessorOptions = null)
+            Action<ServiceBusSessionProcessorOptions>? configureSessionProcessorOptions = null,
+            TimeSpan? renewSessionLockInterval = null)
             where TCommand : ICommandBase
         {
             var queueName = GetQueueName<TCommand>();
@@ -84,7 +89,8 @@ public AzureServiceBusSetupEnvironment AddDelayedSessionProcessor<TCommand>(
                     serviceBusSessionProcessorOptions);
                 var processor = new AzureServiceBusCommandSessionProcessor<TCommand>(
                     serviceBusSessionProcessor,
-                    _serviceCollection);
+                    _serviceCollection,
+                    renewSessionLockInterval ?? TimeSpan.FromMinutes(1));
 
                 return processor;
             });