diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultRegistriesPopulator.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultRegistriesPopulator.cs index 5987451e1..1ec647cb5 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultRegistriesPopulator.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultRegistriesPopulator.cs @@ -1,9 +1,15 @@ +using Elsa.Mediator.Contracts; using Elsa.Workflows.Management; +using Elsa.Workflows.Runtime.Models; +using Elsa.Workflows.Runtime.Notifications; namespace Elsa.Workflows.Runtime; /// -public class DefaultRegistriesPopulator(IWorkflowDefinitionStorePopulator workflowDefinitionStorePopulator, IActivityRegistryPopulator activityRegistryPopulator) : IRegistriesPopulator +public class DefaultRegistriesPopulator( + IWorkflowDefinitionStorePopulator workflowDefinitionStorePopulator, + IActivityRegistryPopulator activityRegistryPopulator, + INotificationSender notificationSender) : IRegistriesPopulator { /// public async Task PopulateAsync(CancellationToken cancellationToken = default) @@ -21,6 +27,11 @@ public class DefaultRegistriesPopulator(IWorkflowDefinitionStorePopulator workfl // Stage 4. Re-update the workflow definition store with the current set of activities. // Finally, we need to re-populate the workflow definition store to make sure that the workflow definitions are up-to-date. - await workflowDefinitionStorePopulator.PopulateStoreAsync(true, cancellationToken); + var workflowDefinitions = await workflowDefinitionStorePopulator.PopulateStoreAsync(true, cancellationToken); + + // Stage 5: Publish a notification that the workflow definitions have been reloaded. This ensures other replicated nodes can update their activity registry. + var reloadedWorkflowDefinitions = workflowDefinitions.Select(ReloadedWorkflowDefinition.FromDefinition).ToList(); + var notification = new WorkflowDefinitionsReloaded(reloadedWorkflowDefinitions); + await notificationSender.SendAsync(notification, cancellationToken); } } \ No newline at end of file diff --git a/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DefaultRegistriesPopulatorTests.cs b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DefaultRegistriesPopulatorTests.cs new file mode 100644 index 000000000..938c2e3c4 --- /dev/null +++ b/test/unit/Elsa.Workflows.Runtime.UnitTests/Services/DefaultRegistriesPopulatorTests.cs @@ -0,0 +1,59 @@ +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Management; +using Elsa.Workflows.Management.Entities; +using Elsa.Workflows.Runtime.Notifications; +using NSubstitute; + +namespace Elsa.Workflows.Runtime.UnitTests.Services; + +public class DefaultRegistriesPopulatorTests +{ + private readonly IWorkflowDefinitionStorePopulator _workflowDefinitionStorePopulatorMock; + private readonly INotificationSender _notificationSenderMock; + private readonly DefaultRegistriesPopulator _populator; + + public DefaultRegistriesPopulatorTests() + { + _workflowDefinitionStorePopulatorMock = Substitute.For(); + var activityRegistryPopulatorMock = Substitute.For(); + _notificationSenderMock = Substitute.For(); + + _populator = new( + _workflowDefinitionStorePopulatorMock, + activityRegistryPopulatorMock, + _notificationSenderMock); + } + + [Fact(DisplayName = "PopulateAsync publishes WorkflowDefinitionsReloaded notification")] + public async Task PopulateAsync_PublishesWorkflowDefinitionsReloadedNotification() + { + // Arrange + var workflowDefinitions = new List + { + new() + { + Id = "1", + DefinitionId = "def-1", + Version = 1 + }, + new() + { + Id = "2", + DefinitionId = "def-2", + Version = 1 + } + }; + + _workflowDefinitionStorePopulatorMock + .PopulateStoreAsync(true, Arg.Any()) + .Returns(workflowDefinitions); + + // Act + await _populator.PopulateAsync(); + + // Assert + await _notificationSenderMock.Received(1).SendAsync( + Arg.Is(n => n.ReloadedWorkflowDefinitions.Count == 2), + Arg.Any()); + } +} \ No newline at end of file