Merged in feature/PAN-2194 (pull request #33)

DavidHogue · DavidHogue · commit 131871aa4406 · 2022-09-07T21:10:35.000Z
PAN-2194 Add async subscription handling

Approved-by: Beth Bennett
Approved-by: Thomas Mooney
Approved-by: Will Wolff-Myren
Approved-by: Dave Sanderling
diff --git a/src/Learning.EventStore.Common/Redis/IRedisClient.cs b/src/Learning.EventStore.Common/Redis/IRedisClient.cs
@@ -21,6 +21,7 @@ public interface IRedisClient
         Task PublishAsync(RedisChannel channel, RedisValue value);
         Task SubscribeAsync(RedisChannel channel, Action<RedisChannel, RedisValue> handler);
         void Subscribe(RedisChannel channel, Action<ChannelMessage> handler);
+        void Subscribe(RedisChannel channel, Func<ChannelMessage, Task> handler);
         Task<string> StringGetAsync(string key);
         Task<bool> StringSetAsync(string key, string value, TimeSpan? expiry = null);
         Task<bool> KeyExistsAsync(string key);
diff --git a/src/Learning.EventStore.Common/Redis/RedisClient.cs b/src/Learning.EventStore.Common/Redis/RedisClient.cs
@@ -140,6 +140,11 @@ public void Subscribe(RedisChannel channel, Action<ChannelMessage> handler)
             RetryPolicy.Execute(() => _redisConnection.GetSubscriber().Subscribe(channel).OnMessage(handler));
         }
 
+        public void Subscribe(RedisChannel channel, Func<ChannelMessage, Task> handler)
+        {
+            RetryPolicy.Execute(() => _redisConnection.GetSubscriber().Subscribe(channel).OnMessage(handler));
+        }
+
         public async Task<string> StringGetAsync(string key)
         {
             var result = await RetryPolicyAsync.ExecuteAsync(() => Database.StringGetAsync(key)).ConfigureAwait(false);
diff --git a/src/Learning.EventStore/Learning.EventStore.csproj b/src/Learning.EventStore/Learning.EventStore.csproj
@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <AssemblyTitle>Learning.EventStore</AssemblyTitle>
-    <VersionPrefix>11.1.2</VersionPrefix>
+    <VersionPrefix>11.2.0</VersionPrefix>
     <TargetFrameworks>netstandard2.0</TargetFrameworks>
     <AssemblyName>Learning.EventStore</AssemblyName>
     <PackageId>Learning.EventStore</PackageId>
diff --git a/src/Learning.MessageQueue/IEventSubscriber.cs b/src/Learning.MessageQueue/IEventSubscriber.cs
@@ -9,5 +9,9 @@ public interface IEventSubscriber
         Task SubscribeAsync<T>(Action<T> callBack) where T : IMessage;
         Task SubscribeAsync<T>(Action<T> callBack, bool enableLock) where T : IMessage;
         Task SubscribeAsync<T>(Action<T> callBack, bool enableLock, bool sequentialProcessing) where T : IMessage;
+
+        Task SubscribeAsync<T>(Func<T, Task> callBack) where T : IMessage;
+        Task SubscribeAsync<T>(Func<T, Task> callBack, bool enableLock) where T : IMessage;
+        Task SubscribeAsync<T>(Func<T, Task> callBack, bool enableLock, bool sequentialProcessing) where T : IMessage;
     }
 }
diff --git a/src/Learning.MessageQueue/Learning.MessageQueue.csproj b/src/Learning.MessageQueue/Learning.MessageQueue.csproj
@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <AssemblyTitle>Learning.MessageQueue</AssemblyTitle>
-    <VersionPrefix>4.1.1</VersionPrefix>
+    <VersionPrefix>4.2.0</VersionPrefix>
     <TargetFrameworks>netstandard2.0</TargetFrameworks>
     <AssemblyName>Learning.MessageQueue</AssemblyName>
     <PackageId>Learning.MessageQueue</PackageId>
diff --git a/src/Learning.MessageQueue/Messages/AsyncRedisSubscription.cs b/src/Learning.MessageQueue/Messages/AsyncRedisSubscription.cs
@@ -0,0 +1,41 @@
+﻿using System;
+using System.Threading.Tasks;
+using Learning.MessageQueue.Logging;
+
+namespace Learning.MessageQueue.Messages
+{
+    public abstract class AsyncRedisSubscription<T> : ISubscription where T : IMessage
+    {
+        private readonly IEventSubscriber _subscriber;
+        private readonly bool _useLock;
+        private readonly ILog _logger;
+        private readonly bool _sequentialProcessing;
+
+        protected AsyncRedisSubscription(IEventSubscriber subscriber)
+            : this(subscriber, false, false)
+        {
+        }
+
+        protected AsyncRedisSubscription(IEventSubscriber subscriber, bool useLock)
+            : this(subscriber, useLock, false)
+        {
+        }
+
+        protected AsyncRedisSubscription(IEventSubscriber subscriber, bool useLock, bool sequentialProcessing)
+        {
+            _subscriber = subscriber;
+            _useLock = useLock;
+            _logger = LogProvider.GetCurrentClassLogger();
+            _sequentialProcessing = sequentialProcessing;
+        }
+
+        public virtual async Task SubscribeAsync()
+        {
+            await _subscriber.SubscribeAsync((Func<T, Task>)CallBack, _useLock, _sequentialProcessing).ConfigureAwait(false);
+            var messageName = typeof(T).Name;
+            _logger.Info($"{messageName} subscription created");
+        }
+
+        protected abstract Task CallBack(T message);
+    }
+}
diff --git a/src/Learning.MessageQueue/Messages/RetryableAsyncRedisSubscription.cs b/src/Learning.MessageQueue/Messages/RetryableAsyncRedisSubscription.cs
@@ -0,0 +1,124 @@
+﻿using System;
+using System.Threading.Tasks;
+using Learning.MessageQueue.Logging;
+using Learning.MessageQueue.Repository;
+using Newtonsoft.Json;
+using StackExchange.Redis;
+
+namespace Learning.MessageQueue.Messages
+{
+    public abstract class RetryableAsyncRedisSubscription<T> : AsyncRedisSubscription<T>, IRetryableSubscription where T: IMessage
+    {
+        private readonly IMessageQueueRepository _messageQueueRepository;
+        private readonly ILog _logger;
+
+        protected RetryableAsyncRedisSubscription(IEventSubscriber subscriber, IMessageQueueRepository messageQueueRepository)
+            : this(subscriber, messageQueueRepository, false, false)
+        {
+        }
+
+        protected RetryableAsyncRedisSubscription(IEventSubscriber subscriber, IMessageQueueRepository messageQueueRepository, bool useLock)
+            : this(subscriber, messageQueueRepository, useLock, false)
+        {
+        }
+
+        protected RetryableAsyncRedisSubscription(IEventSubscriber subscriber, IMessageQueueRepository messageQueueRepository, bool useLock, bool sequentialProcessing)
+            : base(subscriber, useLock, sequentialProcessing)
+        {
+            _logger = LogProvider.GetCurrentClassLogger();
+            _messageQueueRepository = messageQueueRepository;
+        }
+        
+        public virtual int TimeToLiveHours { get; set; } = 168;
+
+        public virtual int RetryIntervalMinutes { get; set; } = 5;
+
+        public virtual int RetryIntervalMaxMinutes { get; set; } = 60;
+
+        public virtual async Task RetryAsync()
+        {
+            var eventType = typeof(T).Name;
+            var listLength = await _messageQueueRepository.GetDeadLetterListLength<T>().ConfigureAwait(false);
+            var eventsProcessed = 0;
+            var errors = 0;
+
+            if (listLength > 0)
+            {
+                _logger.Info($"Beginning retry of {listLength} {typeof(T).Name} events");
+            }
+
+            for (var i = 0; i < listLength; i++)
+            {
+                var indexToGet = i - eventsProcessed;
+                var eventData = await _messageQueueRepository.GetUnprocessedMessage<T>(indexToGet).ConfigureAwait(false);
+
+                if (string.IsNullOrEmpty(eventData))
+                {
+                    continue;
+                }
+
+                var @event = JsonConvert.DeserializeObject<T>(eventData);
+
+                try
+                {
+                    if (await ShouldRetry(@event, eventData).ConfigureAwait(false))
+                    {
+                        _logger.Info($"Beginning retry of processing for {eventType} event for Aggregate: {@event.Id}");
+
+                        await RetryCallBackAsync(@event).ConfigureAwait(false);
+                        await _messageQueueRepository.DeleteFromDeadLetterQueue<T>(eventData, @event).ConfigureAwait(false);
+                        eventsProcessed++;
+
+                        _logger.Info($"Completed retry of processing for {eventType} event for Aggregate: {@event.Id}");
+                    }
+                }
+                catch (Exception e)
+                {
+                    var message = $"{e.Message}{Environment.NewLine}{e.StackTrace}";
+                    await _messageQueueRepository.UpdateRetryData(@event, message).ConfigureAwait(false);
+                    _logger.WarnException($"Event processing retry failed for {eventType} with message: {e.Message}{Environment.NewLine}{e.StackTrace}", e);
+
+                    errors++;
+                }
+            }
+
+            _logger.Info($"Retry complete for {eventType}. Processed {eventsProcessed} events with {errors} errors.");
+        }
+
+        protected virtual async Task RetryCallBackAsync(T message)
+        {
+            await RetryCallBack(message).ConfigureAwait(false);
+        }
+
+        protected virtual async Task RetryCallBack(T message)
+        {
+            await CallBack(message).ConfigureAwait(false);
+        }
+
+        protected virtual async Task<bool> ShouldRetry(IMessage @event, RedisValue eventData)
+        {
+            var retryData = await _messageQueueRepository.GetRetryData(@event).ConfigureAwait(false);
+
+            // exponential backoff
+            var mpow = Math.Pow(2, retryData.RetryCount);
+            var interval = Math.Min(mpow * RetryIntervalMinutes, RetryIntervalMaxMinutes);
+            var intervalPassed = DateTimeOffset.UtcNow > retryData.LastRetryTime?.ToUniversalTime().AddMinutes(interval);
+
+            if (!intervalPassed && retryData.RetryCount > 0)
+            {
+                _logger.Debug($"Skipping retry for event with Aggregate Id {@event.Id}; Retry interval has not elapsed.");
+                return false;
+            }
+
+            if (TimeToLiveHours != default(int) &&
+                DateTimeOffset.UtcNow > @event.TimeStamp.ToUniversalTime().AddHours(TimeToLiveHours))
+            {
+                _logger.Debug($"Time to live of {TimeToLiveHours} hours exceeded for event with Aggregate Id {@event.Id}; Deleting from the dead letter queue.");
+                await _messageQueueRepository.DeleteFromDeadLetterQueue<T>(eventData, @event).ConfigureAwait(false);
+                return false;
+            }
+
+            return true;
+        }
+    }
+}
diff --git a/src/Learning.MessageQueue/RedisEventSubscriber.cs b/src/Learning.MessageQueue/RedisEventSubscriber.cs
@@ -50,7 +50,7 @@ public async Task SubscribeAsync<T>(Action<T> callBack) where T : IMessage
 
         public async Task SubscribeAsync<T>(Action<T> callBack, bool enableLock) where T : IMessage
         {
-            await SubscribeAsync(callBack, enableLock, false);
+            await SubscribeAsync(callBack, enableLock, false).ConfigureAwait(false);
         }
 
         public async Task SubscribeAsync<T>(Action<T> callBack, bool enableLock, bool sequentialProcessing) where T : IMessage
@@ -145,8 +145,7 @@ private void ExecuteCallbackWithLock<T>(Action<T> callBack, string eventKey) whe
                 _logger.ErrorException($"{e.Message}\n{e.StackTrace}", e);
                 throw;
             }
-        } 
-        
+        }
 
         private void ExecuteCallback<T>(Action<T> callBack) where T : IMessage
         {
@@ -186,5 +185,146 @@ Pop the event out of the queue and atomicaly push it into another 'processing' l
                 _redisClient.ListRemove(processingListKey, eventData);
             }
         }
+
+        public async Task SubscribeAsync<T>(Func<T, Task> callBack) where T : IMessage
+        {
+            await SubscribeAsync(callBack, false).ConfigureAwait(false);
+        }
+
+        public async Task SubscribeAsync<T>(Func<T, Task> callBack, bool enableLock) where T : IMessage
+        {
+            await SubscribeAsync(callBack, enableLock, false).ConfigureAwait(false);
+        }
+
+        public async Task SubscribeAsync<T>(Func<T, Task> callBack, bool enableLock, bool sequentialProcessing) where T : IMessage
+        {
+            if (enableLock && _distributedLockFactory == null)
+            {
+                throw new ArgumentNullException(nameof(_distributedLockFactory), "IDistributedLockFactory must be set in constructor if lock is enabled");
+            }
+
+            //Register subscriber
+            var eventType = typeof(T).Name;
+            var eventKey = $"{_environment}:{eventType}";
+            var subscriberSetKey = $"Subscribers:{{{eventKey}}}";
+            var publishedListKey = $"{_applicationName}:{{{eventKey}}}:PublishedEvents";
+            await _redisClient.SetAddAsync(subscriberSetKey, _applicationName).ConfigureAwait(false);
+
+            // Create async Task subscription callback
+            async Task TaskRedisCallback()
+            {
+                if (enableLock)
+                {
+                    await ExecuteCallbackWithLock(callBack, eventKey).ConfigureAwait(false);
+                }
+                else
+                {
+                    await ExecuteCallback(callBack).ConfigureAwait(false);
+                }
+            }
+
+            // Create async void subscription callback
+            async void VoidRedisCallback()
+            {
+                if (enableLock)
+                {
+                    await ExecuteCallbackWithLock(callBack, eventKey).ConfigureAwait(false);
+                }
+                else
+                {
+                    await ExecuteCallback(callBack).ConfigureAwait(false);
+                }
+            }
+
+            if (sequentialProcessing)
+            {
+                _redisClient.Subscribe(eventKey, message => TaskRedisCallback());
+            }
+            else
+            {
+                await _redisClient.SubscribeAsync(eventKey, (channel, data) => VoidRedisCallback()).ConfigureAwait(false);
+            }
+
+            //Grab any unprocessed events and process them
+            //Ensures that events that were fired before the application was started will be picked up
+            var awaitingEvents = await _redisClient.ListLengthAsync(publishedListKey).ConfigureAwait(false);
+            for (var i = 0; i < awaitingEvents; i++)
+            {
+                try
+                {
+                    await Task.Run(async () => await TaskRedisCallback().ConfigureAwait(false)).ConfigureAwait(false);
+                }
+                catch (Exception e)
+                {
+                    _logger.ErrorException(e.Message, e);
+                }
+            }
+        }
+
+        private async Task ExecuteCallbackWithLock<T>(Func<T, Task> callBack, string eventKey) where T : IMessage
+        {
+            try
+            {
+                _logger.Debug($"Distributed lock enabled. Attempting to get lock for message with ID {eventKey}...");
+                using (var distributedLock = await _distributedLockFactory.CreateLockAsync(
+                    eventKey,
+                    TimeSpan.FromSeconds(_lockSettings.ExpirySeconds),
+                    TimeSpan.FromSeconds(_lockSettings.WaitSeconds),
+                    TimeSpan.FromMilliseconds(_lockSettings.RetryMilliseconds))
+                    .ConfigureAwait(false))
+                {
+                    if (distributedLock.IsAcquired)
+                    {
+                        _logger.Debug($"Distributed lock acquired for message with ID {eventKey}; LockId: {distributedLock.LockId};");
+                        await ExecuteCallback(callBack).ConfigureAwait(false);
+                    }
+                    else
+                    {
+                        throw new DistributedLockException(distributedLock, _lockSettings);
+                    }
+                }
+            }
+            catch (Exception e)
+            {
+                _logger.ErrorException($"{e.Message}\n{e.StackTrace}", e);
+            }
+        }
+
+        private async Task ExecuteCallback<T>(Func<T, Task> callBack) where T : IMessage
+        {
+            var eventType = typeof(T).Name;
+            var eventKey = $"{_environment}:{eventType}";
+            var publishedListKey = $"{_applicationName}:{{{eventKey}}}:PublishedEvents";
+            var processingListKey = $"{_applicationName}:{{{eventKey}}}:ProcessingEvents";
+
+            /*
+            Pop the event out of the queue and atomicaly push it into another 'processing' list.
+            Creates a reliable queue where events can be retried if processing fails, see https://redis.io/commands/rpoplpush.
+            */
+            var eventData = _redisClient.ListRightPopLeftPush(publishedListKey, processingListKey);
+
+            // if the eventData is null, then the event has already been processed by another instance, skip further execution
+            if (!eventData.HasValue)
+            {
+                return;
+            }
+
+            //Deserialize the event data and invoke the handler
+            var message = JsonConvert.DeserializeObject<T>(eventData);
+            try
+            {
+                await callBack.Invoke(message).ConfigureAwait(false);
+            }
+            catch (Exception e)
+            {
+                _messageQueueRepository.AddToDeadLetterQueue<T>(eventData, message, e);
+                _logger.ErrorException($"{e.Message}\n{e.StackTrace}", e);
+            }
+            finally
+            {
+                //Remove the event from the 'processing' list.
+                await _redisClient.ListRemoveAsync(processingListKey, eventData).ConfigureAwait(false);
+            }
+        }
     }
 }
diff --git a/test/Learning.EventStore.Test/RedisEventSubscriber/CallbackErrorTest.cs b/test/Learning.EventStore.Test/RedisEventSubscriber/CallbackErrorTest.cs
diff --git a/test/Learning.EventStore.Test/RedisEventSubscriber/NoValueTest.cs b/test/Learning.EventStore.Test/RedisEventSubscriber/NoValueTest.cs
diff --git a/test/Learning.EventStore.Test/RedisEventSubscriber/SubscribeTestAsync.cs b/test/Learning.EventStore.Test/RedisEventSubscriber/SubscribeTestAsync.cs
diff --git a/test/Learning.EventStore.Test/RedisMessageQueue/RetryableAsyncSubscriptionTest.cs b/test/Learning.EventStore.Test/RedisMessageQueue/RetryableAsyncSubscriptionTest.cs

Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,11 @@ public void Subscribe(RedisChannel channel, Action<ChannelMessage> handler)`
`140`	`140`	`RetryPolicy.Execute(() => _redisConnection.GetSubscriber().Subscribe(channel).OnMessage(handler));`
`141`	`141`	`}`
`142`	`142`
	`143`	`+ public void Subscribe(RedisChannel channel, Func<ChannelMessage, Task> handler)`
	`144`	`+ {`
	`145`	`+ RetryPolicy.Execute(() => _redisConnection.GetSubscriber().Subscribe(channel).OnMessage(handler));`
	`146`	`+ }`
	`147`	`+`
`143`	`148`	`public async Task<string> StringGetAsync(string key)`
`144`	`149`	`{`
`145`	`150`	`var result = await RetryPolicyAsync.ExecuteAsync(() => Database.StringGetAsync(key)).ConfigureAwait(false);`
Original file line number	Diff line number	Diff line change
`@@ -9,5 +9,9 @@ public interface IEventSubscriber`
`9`	`9`	`Task SubscribeAsync<T>(Action<T> callBack) where T : IMessage;`
`10`	`10`	`Task SubscribeAsync<T>(Action<T> callBack, bool enableLock) where T : IMessage;`
`11`	`11`	`Task SubscribeAsync<T>(Action<T> callBack, bool enableLock, bool sequentialProcessing) where T : IMessage;`
	`12`	`+`
	`13`	`+ Task SubscribeAsync<T>(Func<T, Task> callBack) where T : IMessage;`
	`14`	`+ Task SubscribeAsync<T>(Func<T, Task> callBack, bool enableLock) where T : IMessage;`
	`15`	`+ Task SubscribeAsync<T>(Func<T, Task> callBack, bool enableLock, bool sequentialProcessing) where T : IMessage;`
`12`	`16`	`}`
`13`	`17`	`}`