diff --git a/src/modules/Elsa.Alterations.Core/Contracts/IAlterationPlanManager.cs b/src/modules/Elsa.Alterations.Core/Contracts/IAlterationPlanManager.cs new file mode 100644 index 000000000..79b72c6d9 --- /dev/null +++ b/src/modules/Elsa.Alterations.Core/Contracts/IAlterationPlanManager.cs @@ -0,0 +1,10 @@ +using Elsa.Alterations.Core.Entities; + +namespace Elsa.Alterations.Core.Contracts; + +public interface IAlterationPlanManager +{ + Task GetPlanAsync(string planId, CancellationToken cancellationToken = default); + Task GetIsAllJobsCompletedAsync(string planId, CancellationToken cancellationToken = default); + Task CompletePlanAsync(AlterationPlan plan, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.Core/Notifications/AlterationPlanCompleted.cs b/src/modules/Elsa.Alterations.Core/Notifications/AlterationPlanCompleted.cs new file mode 100644 index 000000000..70a13fb5b --- /dev/null +++ b/src/modules/Elsa.Alterations.Core/Notifications/AlterationPlanCompleted.cs @@ -0,0 +1,9 @@ +using Elsa.Alterations.Core.Entities; +using Elsa.Mediator.Contracts; + +namespace Elsa.Alterations.Core.Notifications; + +/// +/// A notification that is published when an alteration plan is completed. +/// +public record AlterationPlanCompleted(AlterationPlan Plan) : INotification; \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs b/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs index b4faa0832..27cbaf6b6 100644 --- a/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs +++ b/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs @@ -28,15 +28,15 @@ public class MassTransitAlterationsFeature : FeatureBase public override void Configure() { Module.Configure(feature => feature.AlterationJobDispatcherFactory = sp => sp.GetRequiredService()); - Module.AddMassTransitConsumer(); + var queueName = KebabCaseEndpointNameFormatter.Instance.Consumer(); + var queueAddress = new Uri($"queue:elsa-{queueName}"); + EndpointConvention.Map(queueAddress); + Module.AddMassTransitConsumer(queueName); } /// public override void Apply() { - var queueName = KebabCaseEndpointNameFormatter.Instance.Consumer(); - var queueAddress = new Uri($"queue:elsa-{queueName}"); - EndpointConvention.Map(queueAddress); Services.AddScoped(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Activities/CompleteAlterationPlan.cs b/src/modules/Elsa.Alterations/Activities/CompleteAlterationPlan.cs new file mode 100644 index 000000000..c4b664e2e --- /dev/null +++ b/src/modules/Elsa.Alterations/Activities/CompleteAlterationPlan.cs @@ -0,0 +1,48 @@ +using System.ComponentModel; +using System.Runtime.CompilerServices; +using Elsa.Alterations.Core.Contracts; +using Elsa.Workflows; +using Elsa.Workflows.Attributes; +using Elsa.Workflows.Exceptions; +using Elsa.Workflows.Memory; +using Elsa.Workflows.Models; + +namespace Elsa.Alterations.Activities; + +/// +/// Marks an alteration plan as completed. +/// +[Browsable(false)] +[Activity("Elsa", "Alterations", "Dispatches jobs for the specified Alteration Plan", Kind = ActivityKind.Job)] +public class CompleteAlterationPlan : CodeActivity +{ + /// + public CompleteAlterationPlan(Variable planId, [CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + { + PlanId = new Input(planId); + } + + /// + public CompleteAlterationPlan([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) + { + } + + /// + /// The ID of the alteration plan. + /// + public Input PlanId { get; set; } = default!; + + /// + protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) + { + var cancellationToken = context.CancellationToken; + var planId = context.Get(PlanId)!; + var manager = context.GetRequiredService(); + var plan = await manager.GetPlanAsync(planId, cancellationToken); + + if (plan == null) + throw new FaultException($"Alteration Plan with ID {planId} not found."); + + await manager.CompletePlanAsync(plan, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Activities/DispatchAlterationJobs.cs b/src/modules/Elsa.Alterations/Activities/DispatchAlterationJobs.cs index ac5df8353..495d7ea3d 100644 --- a/src/modules/Elsa.Alterations/Activities/DispatchAlterationJobs.cs +++ b/src/modules/Elsa.Alterations/Activities/DispatchAlterationJobs.cs @@ -1,20 +1,13 @@ using System.ComponentModel; using System.Runtime.CompilerServices; using Elsa.Alterations.Core.Contracts; -using Elsa.Alterations.Core.Entities; using Elsa.Alterations.Core.Enums; using Elsa.Alterations.Core.Filters; -using Elsa.Common.Contracts; using Elsa.Workflows; using Elsa.Workflows.Attributes; -using Elsa.Workflows.Contracts; using Elsa.Workflows.Exceptions; -using Elsa.Workflows.Management.Contracts; -using Elsa.Workflows.Management.Filters; using Elsa.Workflows.Memory; using Elsa.Workflows.Models; -using Elsa.Workflows.Runtime.Contracts; -using Elsa.Workflows.Runtime.Filters; namespace Elsa.Alterations.Activities; @@ -72,5 +65,9 @@ public class DispatchAlterationJobs : CodeActivity var alterationJobDispatcher = context.GetRequiredService(); foreach (var jobId in alterationJobIds) await alterationJobDispatcher.DispatchAsync(jobId, cancellationToken); + + // Update status. + plan.Status = AlterationPlanStatus.Running; + await alterationPlanStore.SaveAsync(plan, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Activities/GenerateAlterationJobs.cs b/src/modules/Elsa.Alterations/Activities/GenerateAlterationJobs.cs index 7aa4f7a65..6bf093267 100644 --- a/src/modules/Elsa.Alterations/Activities/GenerateAlterationJobs.cs +++ b/src/modules/Elsa.Alterations/Activities/GenerateAlterationJobs.cs @@ -6,6 +6,7 @@ using Elsa.Alterations.Core.Enums; using Elsa.Alterations.Core.Filters; using Elsa.Alterations.Core.Models; using Elsa.Common.Contracts; +using Elsa.Extensions; using Elsa.Workflows; using Elsa.Workflows.Attributes; using Elsa.Workflows.Contracts; @@ -24,7 +25,7 @@ namespace Elsa.Alterations.Activities; /// [Browsable(false)] [Activity("Elsa", "Alterations", "Generates jobs for the specified Alteration Plan", Kind = ActivityKind.Job)] -public class GenerateAlterationJobs : CodeActivity +public class GenerateAlterationJobs : CodeActivity { /// public GenerateAlterationJobs([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) @@ -47,10 +48,14 @@ public class GenerateAlterationJobs : CodeActivity { var plan = await GetPlanAsync(context); await UpdatePlanStatusAsync(context, plan); - var workflowInstanceIds = await FindMatchingWorkflowInstanceIdsAsync(context, plan.WorkflowInstanceFilter); - await GenerateJobsAsync(context, plan, workflowInstanceIds); + var workflowInstanceIds = (await FindMatchingWorkflowInstanceIdsAsync(context, plan.WorkflowInstanceFilter)).ToList(); + + if (workflowInstanceIds.Any()) + await GenerateJobsAsync(context, plan, workflowInstanceIds); + + context.SetResult(workflowInstanceIds.Count); } - + private async Task GetPlanAsync(ActivityExecutionContext context) { var cancellationToken = context.CancellationToken; @@ -64,7 +69,7 @@ public class GenerateAlterationJobs : CodeActivity if (plan == null) throw new FaultException($"Alteration Plan with ID {planId} not found."); - + return plan; } @@ -108,10 +113,10 @@ public class GenerateAlterationJobs : CodeActivity workflowInstanceIds.UnionWith(matchingWorkflowInstanceIds); } } - + return workflowInstanceIds; } - + private async Task GenerateJobsAsync(ActivityExecutionContext context, AlterationPlan plan, IEnumerable workflowInstanceIds) { var cancellationToken = context.CancellationToken; diff --git a/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs b/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs index d33062de7..a4588ee27 100644 --- a/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs +++ b/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs @@ -1,7 +1,6 @@ using Elsa.Alterations.Core.Contracts; using Elsa.Alterations.Core.Entities; using Elsa.Alterations.Core.Extensions; -using Elsa.Alterations.Core.Services; using Elsa.Alterations.Core.Stores; using Elsa.Alterations.Extensions; using Elsa.Alterations.Services; @@ -60,6 +59,7 @@ public class AlterationsFeature : FeatureBase /// public override void Apply() { + Services.AddScoped(); Services.AddAlterations(); Services.AddAlterationsCore(); Services.AddScoped(); diff --git a/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs b/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs index 37313e880..e9b2758e5 100644 --- a/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs +++ b/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs @@ -1,14 +1,6 @@ -using Elsa.Alterations.Activities; -using Elsa.Alterations.Bookmarks; using Elsa.Alterations.Core.Contracts; -using Elsa.Alterations.Core.Enums; -using Elsa.Alterations.Core.Filters; using Elsa.Alterations.Core.Notifications; -using Elsa.Common.Contracts; using Elsa.Mediator.Contracts; -using Elsa.Workflows.Helpers; -using Elsa.Workflows.Runtime.Contracts; -using Elsa.Workflows.Runtime.Requests; using JetBrains.Annotations; namespace Elsa.Alterations.Handlers; @@ -17,53 +9,19 @@ namespace Elsa.Alterations.Handlers; /// Handles notifications and updates the plan status if all jobs are completed. /// [UsedImplicitly] -public class AlterationJobCompletedHandler : INotificationHandler +public class AlterationJobCompletedHandler(IAlterationPlanManager manager) : INotificationHandler { - private readonly IWorkflowDispatcher _workflowDispatcher; - private readonly IAlterationPlanStore _alterationPlanStore; - private readonly IAlterationJobStore _alterationJobStore; - private readonly ISystemClock _systemClock; - - /// - /// Initializes a new instance of the class. - /// - public AlterationJobCompletedHandler(IWorkflowDispatcher workflowDispatcher, IAlterationPlanStore alterationPlanStore, IAlterationJobStore alterationJobStore, ISystemClock systemClock) - { - _workflowDispatcher = workflowDispatcher; - _alterationPlanStore = alterationPlanStore; - _alterationJobStore = alterationJobStore; - _systemClock = systemClock; - } - /// public async Task HandleAsync(AlterationJobCompleted notification, CancellationToken cancellationToken) { var job = notification.Job; var planId = job.PlanId; - var planFilter = new AlterationPlanFilter { Id = planId }; - var plan = (await _alterationPlanStore.FindAsync(planFilter, cancellationToken))!; - - // Check if all jobs are completed. - var jobFilter = new AlterationJobFilter - { - PlanId = planId, - Statuses = new[] { AlterationJobStatus.Pending, AlterationJobStatus.Running } - }; - - var allJobsCompleted = await _alterationJobStore.CountAsync(jobFilter, cancellationToken) == 0; + var plan = (await manager.GetPlanAsync(planId, cancellationToken))!; + var allJobsCompleted = await manager.GetIsAllJobsCompletedAsync(planId, cancellationToken); if(!allJobsCompleted) return; - // Update plan status. - plan.Status = AlterationPlanStatus.Completed; - plan.CompletedAt = _systemClock.UtcNow; - - await _alterationPlanStore.SaveAsync(plan, cancellationToken); - - // Trigger any workflow instances that are waiting for the plan to complete. - var bookmarkPayload = new AlterationPlanCompletedPayload(planId); - var triggerRequest = new DispatchTriggerWorkflowsRequest(ActivityTypeNameHelper.GenerateTypeName(), bookmarkPayload); - await _workflowDispatcher.DispatchAsync(triggerRequest, cancellationToken); + await manager.CompletePlanAsync(plan, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Handlers/AlterationPlanCompletedHandler.cs b/src/modules/Elsa.Alterations/Handlers/AlterationPlanCompletedHandler.cs new file mode 100644 index 000000000..2657cec3c --- /dev/null +++ b/src/modules/Elsa.Alterations/Handlers/AlterationPlanCompletedHandler.cs @@ -0,0 +1,26 @@ +using Elsa.Alterations.Bookmarks; +using Elsa.Alterations.Core.Notifications; +using Elsa.Mediator.Contracts; +using Elsa.Workflows.Helpers; +using Elsa.Workflows.Runtime.Contracts; +using Elsa.Workflows.Runtime.Requests; +using JetBrains.Annotations; + +namespace Elsa.Alterations.Handlers; + +/// +/// Handles notifications and triggers any workflows that are waiting for the plan to complete. +/// +[UsedImplicitly] +public class AlterationPlanCompletedHandler(IWorkflowDispatcher workflowDispatcher) : INotificationHandler +{ + /// + public async Task HandleAsync(AlterationPlanCompleted notification, CancellationToken cancellationToken) + { + // Trigger any workflow instances that are waiting for the plan to complete. + var planId = notification.Plan.Id; + var bookmarkPayload = new AlterationPlanCompletedPayload(planId); + var triggerRequest = new DispatchTriggerWorkflowsRequest(ActivityTypeNameHelper.GenerateTypeName(), bookmarkPayload); + await workflowDispatcher.DispatchAsync(triggerRequest, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Services/AlterationPlanManager.cs b/src/modules/Elsa.Alterations/Services/AlterationPlanManager.cs new file mode 100644 index 000000000..77d9ba6f5 --- /dev/null +++ b/src/modules/Elsa.Alterations/Services/AlterationPlanManager.cs @@ -0,0 +1,43 @@ +using Elsa.Alterations.Core.Contracts; +using Elsa.Alterations.Core.Entities; +using Elsa.Alterations.Core.Enums; +using Elsa.Alterations.Core.Filters; +using Elsa.Alterations.Core.Notifications; +using Elsa.Common.Contracts; +using Elsa.Mediator.Contracts; + +namespace Elsa.Alterations.Services; + +/// +public class AlterationPlanManager(IAlterationPlanStore planStore, IAlterationJobStore jobStore, ISystemClock systemClock, INotificationSender notificationSender) : IAlterationPlanManager +{ + /// + public async Task GetPlanAsync(string planId, CancellationToken cancellationToken = default) + { + var planFilter = new AlterationPlanFilter { Id = planId }; + return await planStore.FindAsync(planFilter, cancellationToken); + } + + /// + public async Task GetIsAllJobsCompletedAsync(string planId, CancellationToken cancellationToken = default) + { + // Check if all jobs are completed. + var jobFilter = new AlterationJobFilter + { + PlanId = planId, + Statuses = new[] { AlterationJobStatus.Pending, AlterationJobStatus.Running } + }; + + return await jobStore.CountAsync(jobFilter, cancellationToken) == 0; + } + + /// + public async Task CompletePlanAsync(AlterationPlan plan, CancellationToken cancellationToken = default) + { + plan.Status = AlterationPlanStatus.Completed; + plan.CompletedAt = systemClock.UtcNow; + + await planStore.SaveAsync(plan, cancellationToken); + await notificationSender.SendAsync(new AlterationPlanCompleted(plan), cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Services/DefaultAlterationJobRunner.cs b/src/modules/Elsa.Alterations/Services/DefaultAlterationJobRunner.cs index 36637c384..604baf9c0 100644 --- a/src/modules/Elsa.Alterations/Services/DefaultAlterationJobRunner.cs +++ b/src/modules/Elsa.Alterations/Services/DefaultAlterationJobRunner.cs @@ -6,7 +6,7 @@ using Elsa.Alterations.Core.Notifications; using Elsa.Common.Contracts; using Elsa.Mediator.Contracts; -namespace Elsa.Alterations.Core.Services; +namespace Elsa.Alterations.Services; /// public class DefaultAlterationJobRunner : IAlterationJobRunner diff --git a/src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs b/src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs index 56a5a269c..5cf2df277 100644 --- a/src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs +++ b/src/modules/Elsa.Alterations/Services/DefaultAlterationRunner.cs @@ -9,7 +9,7 @@ using Elsa.Workflows.Management.Contracts; using Elsa.Workflows.Runtime.Contracts; using Microsoft.Extensions.Logging; -namespace Elsa.Alterations.Core.Services; +namespace Elsa.Alterations.Services; /// public class DefaultAlterationRunner : IAlterationRunner diff --git a/src/modules/Elsa.Alterations/Workflows/ExecuteAlterationPlanWorkflow.cs b/src/modules/Elsa.Alterations/Workflows/ExecuteAlterationPlanWorkflow.cs index dd4b83c1c..cd74511a4 100644 --- a/src/modules/Elsa.Alterations/Workflows/ExecuteAlterationPlanWorkflow.cs +++ b/src/modules/Elsa.Alterations/Workflows/ExecuteAlterationPlanWorkflow.cs @@ -13,7 +13,7 @@ namespace Elsa.Alterations.Workflows; public class ExecuteAlterationPlanWorkflow : WorkflowBase { internal const string WorkflowDefinitionId = "Elsa.Alterations.ExecuteAlterationPlan"; - + /// protected override void Build(IWorkflowBuilder builder) { @@ -21,6 +21,7 @@ public class ExecuteAlterationPlanWorkflow : WorkflowBase builder.AsSystemWorkflow(); var plan = builder.WithInput("Plan", "The parameters for the new plan"); var planId = builder.WithVariable(); + var jobCount = builder.WithVariable(); builder.Root = new Sequence { @@ -32,8 +33,15 @@ public class ExecuteAlterationPlanWorkflow : WorkflowBase Result = new(planId) }, new Correlate(planId), - new GenerateAlterationJobs(planId), - new DispatchAlterationJobs(planId), + new GenerateAlterationJobs(planId) + { + Result = new(jobCount) + }, + new If(context => jobCount.Get(context) > 0) + { + Then = new DispatchAlterationJobs(planId), + Else = new CompleteAlterationPlan(planId) + }, new AlterationPlanCompleted(planId), } };