async activity results

Author: danielgerlag

src/WorkflowCore/Models/ActivityResult.cs

@@ -9,6 +9,7 @@ public class ActivityResult
     {
         public enum StatusType { Success, Fail }
         public StatusType Status { get; set; }
+        public string SubscriptionId { get; set; }
         public object Data { get; set; }
     }
 }

src/WorkflowCore/Services/ActivityController.cs

@@ -18,14 +18,16 @@ public class ActivityController : IActivityController
         private readonly IDistributedLockProvider _lockProvider;
         private readonly IDateTimeProvider _dateTimeProvider;
         private readonly IQueueProvider _queueProvider;
+        private readonly IWorkflowController _workflowController;
 
-        public ActivityController(ISubscriptionRepository subscriptionRepository, IWorkflowRepository persistenceStore, IDateTimeProvider dateTimeProvider, IDistributedLockProvider lockProvider, IQueueProvider queueProvider)
+        public ActivityController(ISubscriptionRepository subscriptionRepository, IWorkflowRepository persistenceStore, IWorkflowController workflowController, IDateTimeProvider dateTimeProvider, IDistributedLockProvider lockProvider, IQueueProvider queueProvider)
         {
             _persistenceStore = persistenceStore;
             _subscriptionRepository = subscriptionRepository;
             _dateTimeProvider = dateTimeProvider;
             _lockProvider = lockProvider;
             _queueProvider = queueProvider;
+            _workflowController = workflowController;
         }
 
         public async Task<PendingActivity> GetPendingActivity(string activityName, string workerId, TimeSpan? timeout = null)
@@ -93,7 +95,7 @@ public async Task SubmitActivityFailure(string token, object result)
             });
         }
 
-        private async Task SubmitActivityResult(string token, object result)
+        private async Task SubmitActivityResult(string token, ActivityResult result)
         {
             var tokenObj = Token.Decode(token);
             var sub = await _subscriptionRepository.GetSubscription(tokenObj.SubscriptionId);
@@ -102,28 +104,32 @@ private async Task SubmitActivityResult(string token, object result)
 
             if (sub.ExternalToken != token)
                 throw new NotFoundException("Token mismatch");
-
-            if (!await _lockProvider.AcquireLock(sub.WorkflowId, CancellationToken.None))
-                throw new WorkflowLockedException();
-
-            try
-            {
-                var workflow = await _persistenceStore.GetWorkflowInstance(sub.WorkflowId);
-                var pointer = workflow.ExecutionPointers.Single(p => p.Id == sub.ExecutionPointerId);
-
-                pointer.EventData = result;
-                pointer.EventPublished = true;
-                pointer.Active = true;
-
-                workflow.NextExecution = 0;
-                await _persistenceStore.PersistWorkflow(workflow);
-                await _subscriptionRepository.TerminateSubscription(sub.Id);
-            }
-            finally
-            {
-                await _lockProvider.ReleaseLock(sub.WorkflowId);
-                await _queueProvider.QueueWork(sub.WorkflowId, QueueType.Workflow);
-            }
+            
+            result.SubscriptionId = sub.Id;
+
+            await _workflowController.PublishEvent(sub.EventName, sub.EventKey, result);
+
+            //if (!await _lockProvider.AcquireLock(sub.WorkflowId, CancellationToken.None))
+            //    throw new WorkflowLockedException();
+
+            //try
+            //{
+            //    var workflow = await _persistenceStore.GetWorkflowInstance(sub.WorkflowId);
+            //    var pointer = workflow.ExecutionPointers.Single(p => p.Id == sub.ExecutionPointerId);
+
+            //    pointer.EventData = result;
+            //    pointer.EventPublished = true;
+            //    pointer.Active = true;
+
+            //    workflow.NextExecution = 0;
+            //    await _persistenceStore.PersistWorkflow(workflow);
+            //    await _subscriptionRepository.TerminateSubscription(sub.Id);
+            //}
+            //finally
+            //{
+            //    await _lockProvider.ReleaseLock(sub.WorkflowId);
+            //    await _queueProvider.QueueWork(sub.WorkflowId, QueueType.Workflow);
+            //}
         }
 
         class Token

src/WorkflowCore/Services/BackgroundTasks/EventConsumer.cs

@@ -43,7 +43,23 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
                 var evt = await _eventRepository.GetEvent(itemId);
                 if (evt.EventTime <= _datetimeProvider.UtcNow)
                 {
-                    var subs = await _subscriptionRepository.GetSubscriptions(evt.EventName, evt.EventKey, evt.EventTime);
+                    IEnumerable<EventSubscription> subs = null;
+                    if (evt.EventData is ActivityResult)
+                    {
+                        var activity = await _subscriptionRepository.GetSubscription((evt.EventData as ActivityResult).SubscriptionId);
+                        if (activity == null)
+                        {
+                            Logger.LogWarning($"Activity already processed - {(evt.EventData as ActivityResult).SubscriptionId}");
+                            await _eventRepository.MarkEventProcessed(itemId);
+                            return;
+                        }
+                        subs = new List<EventSubscription>() { activity };
+                    }
+                    else
+                    {
+                        subs = await _subscriptionRepository.GetSubscriptions(evt.EventName, evt.EventKey, evt.EventTime);
+                    }
+
                     var toQueue = new List<string>();
                     var complete = true;
 
@@ -90,7 +106,13 @@ private async Task<bool> SeedSubscription(Event evt, EventSubscription sub, List
             try
             {
                 var workflow = await _workflowRepository.GetWorkflowInstance(sub.WorkflowId);
-                var pointers = workflow.ExecutionPointers.Where(p => p.EventName == sub.EventName && p.EventKey == sub.EventKey && !p.EventPublished && p.EndTime == null);
+                IEnumerable<ExecutionPointer> pointers = null;
+                
+                if (!string.IsNullOrEmpty(sub.ExecutionPointerId))
+                    pointers = workflow.ExecutionPointers.Where(p => p.Id == sub.ExecutionPointerId && !p.EventPublished && p.EndTime == null);
+                else
+                    pointers = workflow.ExecutionPointers.Where(p => p.EventName == sub.EventName && p.EventKey == sub.EventKey && !p.EventPublished && p.EndTime == null);
+
                 foreach (var p in pointers)
                 {
                     p.EventData = evt.EventData;