Merge pull request danielgerlag#845 from Granjow/feature/restart-fixes

Fix lifecycle events when restarting Workflow Core

Author: danielgerlag

WorkflowCore.sln

@@ -7,7 +7,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "src", "src", "{EF47161E-E39
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{F6AC9AEB-24EF-475A-B190-AA4D9E01270A}"
 	ProjectSection(SolutionItems) = preProject
-		readme.md = readme.md
+		README.md = README.md
 	EndProjectSection
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "samples", "samples", "{5080DB09-CBE8-4C45-9957-C3BB7651755E}"

src/WorkflowCore/Interface/IDistributedLockProvider.cs

@@ -9,6 +9,12 @@ namespace WorkflowCore.Interface
     /// </remarks>
     public interface IDistributedLockProvider
     {
+        /// <summary>
+        /// Acquire a lock on the specified resource.
+        /// </summary>
+        /// <param name="Id">Resource ID to lock.</param>
+        /// <param name="cancellationToken"></param>
+        /// <returns>`true`, if the lock was acquired.</returns>
         Task<bool> AcquireLock(string Id, CancellationToken cancellationToken);
 
         Task ReleaseLock(string Id);

src/WorkflowCore/Services/LifeCycleEventPublisher.cs

@@ -11,7 +11,7 @@ public class LifeCycleEventPublisher : ILifeCycleEventPublisher, IDisposable
     {
         private readonly ILifeCycleEventHub _eventHub;
         private readonly ILogger _logger;
-        private readonly BlockingCollection<LifeCycleEvent> _outbox;
+        private BlockingCollection<LifeCycleEvent> _outbox;
         private Task _dispatchTask;
 
         public LifeCycleEventPublisher(ILifeCycleEventHub eventHub, ILoggerFactory loggerFactory)
@@ -36,6 +36,11 @@ public void Start()
                 throw new InvalidOperationException();
             }
 
+            if (_outbox.IsAddingCompleted)
+            {
+                _outbox = new BlockingCollection<LifeCycleEvent>();
+            }
+
             _dispatchTask = new Task(Execute);
             _dispatchTask.Start();
         }

src/WorkflowCore/Services/WorkflowHost.cs

@@ -47,7 +47,6 @@ public WorkflowHost(IPersistenceProvider persistenceStore, IQueueProvider queueP
             _searchIndex = searchIndex;
             _activityController = activityController;
             _lifeCycleEventHub = lifeCycleEventHub;
-            _lifeCycleEventHub.Subscribe(HandleLifeCycleEvent);
         }
 
         public Task<string> StartWorkflow(string workflowId, object data = null, string reference=null)
@@ -91,6 +90,10 @@ public async Task StartAsync(CancellationToken cancellationToken)
             await _lifeCycleEventHub.Start();
             await _searchIndex.Start();
 
+            // Event subscriptions are removed when stopping the event hub.
+            // Add them when starting.
+            AddEventSubscriptions();
+
             Logger.LogInformation("Starting background tasks");
 
             foreach (var task in _backgroundTasks)
@@ -181,5 +184,10 @@ public Task SubmitActivityFailure(string token, object result)
         {
             return _activityController.SubmitActivityFailure(token, result);
         }
+
+        private void AddEventSubscriptions()
+        {
+            _lifeCycleEventHub.Subscribe(HandleLifeCycleEvent);
+        }
     }
 }

test/WorkflowCore.UnitTests/Services/LifeCycleEventPublisherTests.cs

@@ -0,0 +1,54 @@
+using System;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Moq;
+using WorkflowCore.Interface;
+using WorkflowCore.Models.LifeCycleEvents;
+using WorkflowCore.Services;
+using Xunit;
+
+namespace WorkflowCore.UnitTests.Services
+{
+    public class LifeCycleEventPublisherTests
+    {
+        [Fact(DisplayName = "Notifications should be published when the publisher is running")]
+        public async Task PublishNotification_WhenStarted_PublishesNotification()
+        {
+            // Arrange
+            var wasCalled = new TaskCompletionSource<bool>();
+            var eventHubMock = new Mock<ILifeCycleEventHub>();
+            eventHubMock
+                .Setup(hub => hub.PublishNotification(It.IsAny<StepCompleted>()))
+                .Callback(() => wasCalled.SetResult(true));
+            LifeCycleEventPublisher publisher = new LifeCycleEventPublisher(eventHubMock.Object, new LoggerFactory());
+
+            // Act
+            publisher.Start();
+            publisher.PublishNotification(new StepCompleted());
+
+            // Assert
+            await wasCalled.Task;
+        }
+
+        [Fact(DisplayName = "Notifications should be published when the publisher is running")]
+        public async Task PublishNotification_WhenRestarted_PublishesNotification()
+        {
+            // Arrange
+            var wasCalled = new TaskCompletionSource<bool>();
+            var eventHubMock = new Mock<ILifeCycleEventHub>();
+            eventHubMock
+                .Setup(hub => hub.PublishNotification(It.IsAny<StepCompleted>()))
+                .Callback(() => wasCalled.SetResult(true));
+            LifeCycleEventPublisher publisher = new LifeCycleEventPublisher(eventHubMock.Object, new LoggerFactory());
+
+            // Act
+            publisher.Start();
+            publisher.Stop();
+            publisher.Start();
+            publisher.PublishNotification(new StepCompleted());
+
+            // Assert
+            await wasCalled.Task;
+        }
+    }
+}