diff --git a/Elsa.sln b/Elsa.sln index bbce11dd9..94aa78944 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -247,8 +247,6 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Alterations.Core", "sr EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Alterations", "src\modules\Elsa.Alterations\Elsa.Alterations.csproj", "{CC7DE119-736D-4FC5-85DC-70A6770749F6}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Alterations.BackgroundRunner", "src\modules\Elsa.Alterations.BackgroundRunner\Elsa.Alterations.BackgroundRunner.csproj", "{444760FF-1D1A-4B7B-94AA-EA5FCC81BE6B}" -EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.EntityFrameworkCore.Common", "src\modules\Elsa.EntityFrameworkCore.Common\Elsa.EntityFrameworkCore.Common.csproj", "{6F30A941-6EB6-4C9B-BF89-9BBAB757B491}" EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.EntityFrameworkCore.Sqlite", "src\modules\Elsa.EntityFrameworkCore.Sqlite\Elsa.EntityFrameworkCore.Sqlite.csproj", "{EFE4A814-F58D-4C19-8F7F-B95F4EE343F0}" @@ -259,6 +257,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.EntityFrameworkCore.My EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.EntityFrameworkCore.PostgreSql", "src\modules\Elsa.EntityFrameworkCore.PostgreSql\Elsa.EntityFrameworkCore.PostgreSql.csproj", "{84AB9951-3DBF-43A4-8F5D-AEED1102CE88}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Alterations.MassTransit", "src\modules\Elsa.Alterations.MassTransit\Elsa.Alterations.MassTransit.csproj", "{1F621382-08FB-45EA-B94C-96A7FECB9489}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -625,10 +625,6 @@ Global {CC7DE119-736D-4FC5-85DC-70A6770749F6}.Debug|Any CPU.Build.0 = Debug|Any CPU {CC7DE119-736D-4FC5-85DC-70A6770749F6}.Release|Any CPU.ActiveCfg = Release|Any CPU {CC7DE119-736D-4FC5-85DC-70A6770749F6}.Release|Any CPU.Build.0 = Release|Any CPU - {444760FF-1D1A-4B7B-94AA-EA5FCC81BE6B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {444760FF-1D1A-4B7B-94AA-EA5FCC81BE6B}.Debug|Any CPU.Build.0 = Debug|Any CPU - {444760FF-1D1A-4B7B-94AA-EA5FCC81BE6B}.Release|Any CPU.ActiveCfg = Release|Any CPU - {444760FF-1D1A-4B7B-94AA-EA5FCC81BE6B}.Release|Any CPU.Build.0 = Release|Any CPU {6F30A941-6EB6-4C9B-BF89-9BBAB757B491}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {6F30A941-6EB6-4C9B-BF89-9BBAB757B491}.Debug|Any CPU.Build.0 = Debug|Any CPU {6F30A941-6EB6-4C9B-BF89-9BBAB757B491}.Release|Any CPU.ActiveCfg = Release|Any CPU @@ -649,6 +645,10 @@ Global {84AB9951-3DBF-43A4-8F5D-AEED1102CE88}.Debug|Any CPU.Build.0 = Debug|Any CPU {84AB9951-3DBF-43A4-8F5D-AEED1102CE88}.Release|Any CPU.ActiveCfg = Release|Any CPU {84AB9951-3DBF-43A4-8F5D-AEED1102CE88}.Release|Any CPU.Build.0 = Release|Any CPU + {1F621382-08FB-45EA-B94C-96A7FECB9489}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {1F621382-08FB-45EA-B94C-96A7FECB9489}.Debug|Any CPU.Build.0 = Debug|Any CPU + {1F621382-08FB-45EA-B94C-96A7FECB9489}.Release|Any CPU.ActiveCfg = Release|Any CPU + {1F621382-08FB-45EA-B94C-96A7FECB9489}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(NestedProjects) = preSolution {155227F0-A33B-40AA-A4B4-06F813EB921B} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F} @@ -760,11 +760,11 @@ Global {CF3DA5F1-F98D-49C9-A980-27E570B38B8D} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79} {8DBEC8F3-908D-46B3-95CA-47EA138AB1A4} = {CF3DA5F1-F98D-49C9-A980-27E570B38B8D} {CC7DE119-736D-4FC5-85DC-70A6770749F6} = {CF3DA5F1-F98D-49C9-A980-27E570B38B8D} - {444760FF-1D1A-4B7B-94AA-EA5FCC81BE6B} = {CF3DA5F1-F98D-49C9-A980-27E570B38B8D} {6F30A941-6EB6-4C9B-BF89-9BBAB757B491} = {9B4F139F-7D26-435C-A561-89E65A67A8E5} {EFE4A814-F58D-4C19-8F7F-B95F4EE343F0} = {9B4F139F-7D26-435C-A561-89E65A67A8E5} {1C84408D-A0A7-4513-AD9A-1BA137ABE62C} = {9B4F139F-7D26-435C-A561-89E65A67A8E5} {E4300D97-1D53-4029-9435-483FB7EC9621} = {9B4F139F-7D26-435C-A561-89E65A67A8E5} {84AB9951-3DBF-43A4-8F5D-AEED1102CE88} = {9B4F139F-7D26-435C-A561-89E65A67A8E5} + {1F621382-08FB-45EA-B94C-96A7FECB9489} = {CF3DA5F1-F98D-49C9-A980-27E570B38B8D} EndGlobalSection EndGlobal diff --git a/src/bundles/Elsa.WorkflowServer.Web/Elsa.WorkflowServer.Web.csproj b/src/bundles/Elsa.WorkflowServer.Web/Elsa.WorkflowServer.Web.csproj index 8c833d716..bfc68cac8 100644 --- a/src/bundles/Elsa.WorkflowServer.Web/Elsa.WorkflowServer.Web.csproj +++ b/src/bundles/Elsa.WorkflowServer.Web/Elsa.WorkflowServer.Web.csproj @@ -10,6 +10,7 @@ + diff --git a/src/bundles/Elsa.WorkflowServer.Web/Program.cs b/src/bundles/Elsa.WorkflowServer.Web/Program.cs index 70eafe9fa..3726ff1cc 100644 --- a/src/bundles/Elsa.WorkflowServer.Web/Program.cs +++ b/src/bundles/Elsa.WorkflowServer.Web/Program.cs @@ -1,5 +1,6 @@ using System.Text.Encodings.Web; using Elsa.Alterations.Extensions; +using Elsa.Alterations.MassTransit.Extensions; using Elsa.Dapper.Extensions; using Elsa.Dapper.Services; using Elsa.DropIns.Extensions; @@ -156,6 +157,8 @@ services else ef.UseSqlite(sqliteConnectionString); }); + + alterations.UseMassTransitDispatcher(); }); // Initialize drop-ins. diff --git a/src/bundles/Elsa.WorkflowServer.Web/appsettings.json b/src/bundles/Elsa.WorkflowServer.Web/appsettings.json index 27fa35940..f4c10c172 100644 --- a/src/bundles/Elsa.WorkflowServer.Web/appsettings.json +++ b/src/bundles/Elsa.WorkflowServer.Web/appsettings.json @@ -15,7 +15,7 @@ }, "AllowedHosts": "*", "ConnectionStrings": { - "Sqlite": "Data Source=elsa.sqlite.db;Cache=Shared;", + "Sqlite": "Data Source=App_Data/elsa.sqlite.db;Cache=Shared;", "MongoDb": "mongodb://localhost:27017/elsa-workflows" }, "Smtp": { diff --git a/src/modules/Elsa.Alterations.BackgroundRunner/Features/AlterationsBackgroundRunnerFeature.cs b/src/modules/Elsa.Alterations.BackgroundRunner/Features/AlterationsBackgroundRunnerFeature.cs deleted file mode 100644 index 80169847b..000000000 --- a/src/modules/Elsa.Alterations.BackgroundRunner/Features/AlterationsBackgroundRunnerFeature.cs +++ /dev/null @@ -1,17 +0,0 @@ -using Elsa.Features.Abstractions; -using Elsa.Features.Services; -using Microsoft.Extensions.DependencyInjection; - -namespace Elsa.Alterations.BackgroundRunner.Features; - -public class AlterationsBackgroundRunnerFeature : FeatureBase -{ - public AlterationsBackgroundRunnerFeature(IModule module) : base(module) - { - } - - public override void Apply() - { - Services.AddNotificationHandlersFrom(typeof(AlterationsBackgroundRunnerFeature)); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobDispatcher.cs b/src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobDispatcher.cs new file mode 100644 index 000000000..e3191dd96 --- /dev/null +++ b/src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobDispatcher.cs @@ -0,0 +1,14 @@ +namespace Elsa.Alterations.Core.Contracts; + +/// +/// Dispatches an alteration job for execution. +/// +public interface IAlterationJobDispatcher +{ + /// + /// Dispatches an alteration job for execution. + /// + /// The ID of the job to dispatch. + /// An optional cancellation token. + ValueTask DispatchAsync(string jobId, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobStore.cs b/src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobStore.cs index ea85c5a03..6d2dc59bc 100644 --- a/src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobStore.cs +++ b/src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobStore.cs @@ -28,6 +28,12 @@ public interface IAlterationJobStore /// Task> FindManyAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default); + /// + /// Finds the IDs of all alteration jobs matching the specified filter. + /// + /// + Task> FindManyIdsAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default); + /// /// Returns the number of alteration jobs matching the specified filter. /// diff --git a/src/modules/Elsa.Alterations.Core/Services/DefaultAlterationJobRunner.cs b/src/modules/Elsa.Alterations.Core/Services/DefaultAlterationJobRunner.cs index f6db16436..36637c384 100644 --- a/src/modules/Elsa.Alterations.Core/Services/DefaultAlterationJobRunner.cs +++ b/src/modules/Elsa.Alterations.Core/Services/DefaultAlterationJobRunner.cs @@ -40,6 +40,11 @@ public class DefaultAlterationJobRunner : IAlterationJobRunner var job = (await _alterationJobStore.FindAsync(new AlterationJobFilter { Id = jobId }, cancellationToken))!; var plan = (await _alterationPlanStore.FindAsync(new AlterationPlanFilter { Id = job.PlanId }, cancellationToken))!; var workflowInstanceId = job.WorkflowInstanceId; + + job.Status = AlterationJobStatus.Running; + job.StartedAt = _systemClock.UtcNow; + await _alterationJobStore.SaveAsync(job, cancellationToken); + var result = await _alterationRunner.RunAsync(workflowInstanceId, plan.Alterations, cancellationToken); job.Status = result.IsSuccessful ? AlterationJobStatus.Completed : AlterationJobStatus.Failed; diff --git a/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs b/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs index 0da4e32f1..1af9f76f0 100644 --- a/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs +++ b/src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs @@ -48,6 +48,13 @@ public class MemoryAlterationJobStore : IAlterationJobStore return Task.FromResult(entities); } + /// + public Task> FindManyIdsAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default) + { + var ids = _store.Query(query => Filter(query, filter)).Select(x => x.Id).ToList().AsEnumerable(); + return Task.FromResult(ids); + } + /// public Task CountAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default) { diff --git a/src/modules/Elsa.Alterations.MassTransit/Consumers/RunAlterationJobConsumer.cs b/src/modules/Elsa.Alterations.MassTransit/Consumers/RunAlterationJobConsumer.cs new file mode 100644 index 000000000..8bc2fe7cf --- /dev/null +++ b/src/modules/Elsa.Alterations.MassTransit/Consumers/RunAlterationJobConsumer.cs @@ -0,0 +1,27 @@ +using Elsa.Alterations.Core.Contracts; +using Elsa.Alterations.MassTransit.Messages; +using MassTransit; + +namespace Elsa.Alterations.MassTransit.Consumers; + +/// +/// Consumes messages. +/// +public class RunAlterationJobConsumer : IConsumer +{ + private readonly IAlterationJobRunner _alterationJobRunner; + + /// + /// Initializes a new instance of the class. + /// + public RunAlterationJobConsumer(IAlterationJobRunner alterationJobRunner) + { + _alterationJobRunner = alterationJobRunner; + } + + /// + public async Task Consume(ConsumeContext context) + { + await _alterationJobRunner.RunAsync(context.Message.JobId, context.CancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.BackgroundRunner/Elsa.Alterations.BackgroundRunner.csproj b/src/modules/Elsa.Alterations.MassTransit/Elsa.Alterations.MassTransit.csproj similarity index 51% rename from src/modules/Elsa.Alterations.BackgroundRunner/Elsa.Alterations.BackgroundRunner.csproj rename to src/modules/Elsa.Alterations.MassTransit/Elsa.Alterations.MassTransit.csproj index 6acbe49f1..e228ff551 100644 --- a/src/modules/Elsa.Alterations.BackgroundRunner/Elsa.Alterations.BackgroundRunner.csproj +++ b/src/modules/Elsa.Alterations.MassTransit/Elsa.Alterations.MassTransit.csproj @@ -6,13 +6,17 @@ net6.0;net7.0 - Provides alteration background job runners. + Provides alteration background job runners using MassTransit. - elsa module alterations + elsa module alterations masstransit - + + + + + diff --git a/src/modules/Elsa.Alterations.MassTransit/Extensions/ModuleExtensions.cs b/src/modules/Elsa.Alterations.MassTransit/Extensions/ModuleExtensions.cs new file mode 100644 index 000000000..f5fef366b --- /dev/null +++ b/src/modules/Elsa.Alterations.MassTransit/Extensions/ModuleExtensions.cs @@ -0,0 +1,20 @@ +using Elsa.Alterations.Features; +using Elsa.Alterations.MassTransit.Features; +using Elsa.Extensions; +using Elsa.Features.Services; + +namespace Elsa.Alterations.MassTransit.Extensions; + +/// +/// Adds the . +/// +public static class ModuleExtensions +{ + /// + /// Adds the . + /// + public static IModule UseMassTransitDispatcher(this AlterationsFeature alterations, Action? configure = default) + { + return alterations.Module.Use(configure); + } +} \ 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 new file mode 100644 index 000000000..c2a80a0c3 --- /dev/null +++ b/src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs @@ -0,0 +1,37 @@ +using Elsa.Alterations.Features; +using Elsa.Alterations.MassTransit.Consumers; +using Elsa.Alterations.MassTransit.Services; +using Elsa.Extensions; +using Elsa.Features.Abstractions; +using Elsa.Features.Attributes; +using Elsa.Features.Services; +using Elsa.MassTransit.Features; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Alterations.MassTransit.Features; + +/// +/// A feature for enabling the Alterations MassTransit feature. +/// +[DependsOn(typeof(MassTransitFeature))] +[DependsOn(typeof(AlterationsFeature))] +public class MassTransitAlterationsFeature : FeatureBase +{ + /// + public MassTransitAlterationsFeature(IModule module) : base(module) + { + } + + /// + public override void Configure() + { + Module.Configure(feature => feature.AlterationJobDispatcherFactory = sp => sp.GetRequiredService()); + Module.AddMassTransitConsumer(); + } + + /// + public override void Apply() + { + Services.AddSingleton(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.BackgroundRunner/FodyWeavers.xml b/src/modules/Elsa.Alterations.MassTransit/FodyWeavers.xml similarity index 100% rename from src/modules/Elsa.Alterations.BackgroundRunner/FodyWeavers.xml rename to src/modules/Elsa.Alterations.MassTransit/FodyWeavers.xml diff --git a/src/modules/Elsa.Alterations.MassTransit/Messages/RunAlterationJob.cs b/src/modules/Elsa.Alterations.MassTransit/Messages/RunAlterationJob.cs new file mode 100644 index 000000000..b96e5f932 --- /dev/null +++ b/src/modules/Elsa.Alterations.MassTransit/Messages/RunAlterationJob.cs @@ -0,0 +1,7 @@ +namespace Elsa.Alterations.MassTransit.Messages; + +/// +/// Represents a request to run an alteration job. +/// +/// The job ID. +public record RunAlterationJob(string JobId); \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.MassTransit/Readme.md b/src/modules/Elsa.Alterations.MassTransit/Readme.md new file mode 100644 index 000000000..9e45a99f3 --- /dev/null +++ b/src/modules/Elsa.Alterations.MassTransit/Readme.md @@ -0,0 +1,4 @@ +# MassTransit Alterations Background Runner + +This module provides an implementation that processes alteration plans in the background using a MassTransit dispatcher and consumer. +This implementation is more resilient than the in-memory queue implementation provided by **Elsa.Alterations.BackgroundRunner** because it uses a message broker to persist messages. \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.MassTransit/Services/MassTransitAlterationJobDispatcher.cs b/src/modules/Elsa.Alterations.MassTransit/Services/MassTransitAlterationJobDispatcher.cs new file mode 100644 index 000000000..edf79941b --- /dev/null +++ b/src/modules/Elsa.Alterations.MassTransit/Services/MassTransitAlterationJobDispatcher.cs @@ -0,0 +1,28 @@ +using Elsa.Alterations.Core.Contracts; +using Elsa.Alterations.MassTransit.Messages; +using MassTransit; + +namespace Elsa.Alterations.MassTransit.Services; + +/// +/// Dispatches an alteration job for execution using MassTransit. +/// +public class MassTransitAlterationJobDispatcher : IAlterationJobDispatcher +{ + private readonly IBus _bus; + + /// + /// Initializes a new instance of the class. + /// + public MassTransitAlterationJobDispatcher(IBus bus) + { + _bus = bus; + } + + /// + public async ValueTask DispatchAsync(string jobId, CancellationToken cancellationToken = default) + { + var message = new RunAlterationJob(jobId); + await _bus.Publish(message, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs b/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs index de41b7040..85a71164f 100644 --- a/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs +++ b/src/modules/Elsa.Alterations/Features/AlterationsFeature.cs @@ -1,13 +1,12 @@ -using Elsa.Alterations.BackgroundRunner.Features; using Elsa.Alterations.Core.Contracts; using Elsa.Alterations.Core.Entities; using Elsa.Alterations.Core.Extensions; using Elsa.Alterations.Core.Stores; using Elsa.Alterations.Extensions; +using Elsa.Alterations.Services; using Elsa.Common.Services; using Elsa.Extensions; using Elsa.Features.Abstractions; -using Elsa.Features.Attributes; using Elsa.Features.Services; using Microsoft.Extensions.DependencyInjection; @@ -16,7 +15,6 @@ namespace Elsa.Alterations.Features; /// /// Adds the Elsa alterations services. /// -[DependsOn(typeof(AlterationsBackgroundRunnerFeature))] public class AlterationsFeature : FeatureBase { /// @@ -33,6 +31,11 @@ public class AlterationsFeature : FeatureBase /// Gets or sets the factory for the alteration job store. /// public Func AlterationJobStoreFactory { get; set; } = sp => sp.GetRequiredService(); + + /// + /// Gets or sets the factory for the alteration job dispatcher. + /// + public Func AlterationJobDispatcherFactory { get; set; } = sp => sp.GetRequiredService(); /// /// Adds an alteration and its handler. @@ -56,11 +59,13 @@ public class AlterationsFeature : FeatureBase { Services.AddAlterations(); Services.AddAlterationsCore(); + Services.AddSingleton(); Services.AddSingleton(); Services.AddSingleton(); Services.AddSingleton(new MemoryStore()); Services.AddSingleton(new MemoryStore()); Services.AddSingleton(AlterationPlanStoreFactory); Services.AddSingleton(AlterationJobStoreFactory); + Services.AddSingleton(AlterationJobDispatcherFactory); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations.BackgroundRunner/Handlers/AlterationJobCompletedHandler.cs b/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs similarity index 96% rename from src/modules/Elsa.Alterations.BackgroundRunner/Handlers/AlterationJobCompletedHandler.cs rename to src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs index 3036fd962..b0f95d33d 100644 --- a/src/modules/Elsa.Alterations.BackgroundRunner/Handlers/AlterationJobCompletedHandler.cs +++ b/src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs @@ -4,12 +4,14 @@ using Elsa.Alterations.Core.Filters; using Elsa.Alterations.Core.Notifications; using Elsa.Common.Contracts; using Elsa.Mediator.Contracts; +using JetBrains.Annotations; -namespace Elsa.Alterations.BackgroundRunner.Handlers; +namespace Elsa.Alterations.Handlers; /// /// Handles notifications and updates the plan status if all jobs are completed. /// +[UsedImplicitly] public class AlterationJobCompletedHandler : INotificationHandler { private readonly IAlterationPlanStore _alterationPlanStore; diff --git a/src/modules/Elsa.Alterations.BackgroundRunner/Handlers/AlterationPlanSubmittedHandler.cs b/src/modules/Elsa.Alterations/Handlers/AlterationPlanSubmittedHandler.cs similarity index 67% rename from src/modules/Elsa.Alterations.BackgroundRunner/Handlers/AlterationPlanSubmittedHandler.cs rename to src/modules/Elsa.Alterations/Handlers/AlterationPlanSubmittedHandler.cs index bd5fad06a..bee86155a 100644 --- a/src/modules/Elsa.Alterations.BackgroundRunner/Handlers/AlterationPlanSubmittedHandler.cs +++ b/src/modules/Elsa.Alterations/Handlers/AlterationPlanSubmittedHandler.cs @@ -4,28 +4,34 @@ using Elsa.Alterations.Core.Enums; using Elsa.Alterations.Core.Filters; using Elsa.Alterations.Core.Notifications; using Elsa.Mediator.Contracts; +using JetBrains.Annotations; -namespace Elsa.Alterations.BackgroundRunner.Handlers; +namespace Elsa.Alterations.Handlers; /// /// Handles notifications and runs the plan in the background. /// +[UsedImplicitly] public class AlterationPlanSubmittedHandler : INotificationHandler { - private readonly IJobQueue _jobQueue; private readonly IAlterationPlanStore _alterationPlanStore; private readonly IAlterationJobStore _alterationJobStore; - private readonly IAlterationJobRunner _alterationJobRunner; + private readonly IAlterationJobDispatcher _alterationJobDispatcher; + private readonly IJobQueue _jobQueue; /// /// Initializes a new instance of the class. /// - public AlterationPlanSubmittedHandler(IJobQueue jobQueue, IAlterationPlanStore alterationPlanStore, IAlterationJobStore alterationJobStore, IAlterationJobRunner alterationJobRunner) + public AlterationPlanSubmittedHandler( + IAlterationPlanStore alterationPlanStore, + IAlterationJobStore alterationJobStore, + IAlterationJobDispatcher alterationJobDispatcher, + IJobQueue jobQueue) { - _jobQueue = jobQueue; _alterationPlanStore = alterationPlanStore; _alterationJobStore = alterationJobStore; - _alterationJobRunner = alterationJobRunner; + _alterationJobDispatcher = alterationJobDispatcher; + _jobQueue = jobQueue; } /// @@ -47,17 +53,10 @@ public class AlterationPlanSubmittedHandler : INotificationHandler ExecuteJobAsync(alterationJob.Id, ct)); - } - } - - private async Task ExecuteJobAsync(string alterationJobId, CancellationToken cancellationToken) - { - await _alterationJobRunner.RunAsync(alterationJobId, cancellationToken); + foreach (var jobId in alterationJobIds) + await _alterationJobDispatcher.DispatchAsync(jobId, cancellationToken); } } \ No newline at end of file diff --git a/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs b/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs index de0e43b7b..22eba245a 100644 --- a/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs +++ b/src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs @@ -3,12 +3,14 @@ using Elsa.Alterations.Core.Notifications; using Elsa.Mediator.Contracts; using Elsa.Workflows.Runtime.Contracts; using Elsa.Workflows.Runtime.Requests; +using JetBrains.Annotations; namespace Elsa.Alterations.Handlers; /// /// Resumes a workflow instance when an alteration job has been completed and the workflow contains scheduled work. /// +[UsedImplicitly] public class ResumeWorkflowInstance : INotificationHandler { private readonly IWorkflowDispatcher _workflowDispatcher; diff --git a/src/modules/Elsa.Alterations/Services/BackgroundAlterationJobDispatcher.cs b/src/modules/Elsa.Alterations/Services/BackgroundAlterationJobDispatcher.cs new file mode 100644 index 000000000..65e3b9ace --- /dev/null +++ b/src/modules/Elsa.Alterations/Services/BackgroundAlterationJobDispatcher.cs @@ -0,0 +1,34 @@ +using Elsa.Alterations.Core.Contracts; +using Elsa.Mediator.Contracts; + +namespace Elsa.Alterations.Services; + +/// +/// Dispatches an alteration job for execution using an in-memory channel. +/// +public class BackgroundAlterationJobDispatcher : IAlterationJobDispatcher +{ + private readonly IJobQueue _jobQueue; + private readonly IAlterationJobRunner _alterationJobRunner; + + /// + /// Initializes a new instance of the class. + /// + public BackgroundAlterationJobDispatcher(IJobQueue jobQueue, IAlterationJobRunner alterationJobRunner) + { + _jobQueue = jobQueue; + _alterationJobRunner = alterationJobRunner; + } + + /// + public ValueTask DispatchAsync(string jobId, CancellationToken cancellationToken = default) + { + _jobQueue.Enqueue(ct => ExecuteJobAsync(jobId, ct)); + return default; + } + + private async Task ExecuteJobAsync(string alterationJobId, CancellationToken cancellationToken) + { + await _alterationJobRunner.RunAsync(alterationJobId, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/AlterationJobStore.cs b/src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/AlterationJobStore.cs index 7fe74e1ea..66964c4b2 100644 --- a/src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/AlterationJobStore.cs +++ b/src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/AlterationJobStore.cs @@ -47,6 +47,12 @@ public class EFCoreAlterationJobStore : IAlterationJobStore return await _store.QueryAsync(filter.Apply, OnLoadAsync, cancellationToken).ToList(); } + /// + public async Task> FindManyIdsAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default) + { + return await _store.QueryAsync(filter.Apply, x => x.Id, cancellationToken); + } + /// public async Task CountAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default) { diff --git a/src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/Configurations.cs b/src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/Configurations.cs index 8339b7ff5..12d3cf638 100644 --- a/src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/Configurations.cs +++ b/src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/Configurations.cs @@ -16,6 +16,7 @@ public class Configurations : IEntityTypeConfiguration, IEntityT builder.Ignore(x => x.WorkflowInstanceIds); builder.Property("SerializedAlterations"); builder.Property("SerializedWorkflowInstanceIds"); + builder.Property(x => x.Status).HasConversion(); builder.HasIndex(x => x.Status).HasDatabaseName($"IX_{nameof(AlterationPlan)}_{nameof(AlterationPlan.Status)}"); builder.HasIndex(x => x.CreatedAt).HasDatabaseName($"IX_{nameof(AlterationPlan)}_{nameof(AlterationPlan.CreatedAt)}"); builder.HasIndex(x => x.StartedAt).HasDatabaseName($"IX_{nameof(AlterationPlan)}_{nameof(AlterationPlan.StartedAt)}"); @@ -27,6 +28,7 @@ public class Configurations : IEntityTypeConfiguration, IEntityT { builder.Ignore(x => x.Log); builder.Property("SerializedLog"); + builder.Property(x => x.Status).HasConversion(); builder.HasIndex(x => x.PlanId).HasDatabaseName($"IX_{nameof(AlterationJob)}_{nameof(AlterationJob.PlanId)}"); builder.HasIndex(x => x.WorkflowInstanceId).HasDatabaseName($"IX_{nameof(AlterationJob)}_{nameof(AlterationJob.WorkflowInstanceId)}"); builder.HasIndex(x => x.Status).HasDatabaseName($"IX_{nameof(AlterationJob)}_{nameof(AlterationJob.Status)}"); diff --git a/src/modules/Elsa.MassTransit/Extensions/ModuleExtensions.cs b/src/modules/Elsa.MassTransit/Extensions/ModuleExtensions.cs index f521f4e1a..49db276a5 100644 --- a/src/modules/Elsa.MassTransit/Extensions/ModuleExtensions.cs +++ b/src/modules/Elsa.MassTransit/Extensions/ModuleExtensions.cs @@ -1,6 +1,7 @@ using Elsa.Features.Services; using Elsa.MassTransit.Features; using Elsa.MassTransit.Options; +using MassTransit; // ReSharper disable once CheckNamespace namespace Elsa.Extensions; @@ -15,6 +16,15 @@ public static class ModuleExtensions /// public static IModule UseMassTransit(this IModule module, Action? configure = default) => module.Use(configure); + /// + /// Registers the specified consumer with MassTransit. + /// + public static IModule AddMassTransitConsumer(this IModule module) where T : IConsumer + { + module.Configure(massTransit => massTransit.AddConsumer()); + return module; + } + /// /// Enable and configure the RabbitMQ broker for MassTransit. /// diff --git a/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs b/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs index f9a2761d0..6a1b73375 100644 --- a/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs +++ b/src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs @@ -25,7 +25,7 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase /// public override void Configure() { - Module.Configure(massTransit => massTransit.AddConsumer()); + Module.AddMassTransitConsumer(); Module.Configure(f => f.WorkflowDispatcher = sp => sp.GetRequiredService()); }