SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

MassTransit Alterations Dispatcher (#4547)

Browse source 
* Add MassTransit dispatcher for alteration jobs
This commit is contained in:
Sipke Schoorstra 2023-10-17 08:51:45 +02:00 committed by GitHub
parent 8c76360f5d
commit 903486ed5a
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
26 changed files with 255 additions and 49 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
Elsa.sln
View file

@ -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

1
src/bundles/Elsa.WorkflowServer.Web/Elsa.WorkflowServer.Web.csproj
View file

@ -10,6 +10,7 @@
<ItemGroup>
<ProjectReference Include="..\..\common\Elsa.DropIns\Elsa.DropIns.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Alterations.MassTransit\Elsa.Alterations.MassTransit.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Alterations\Elsa.Alterations.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.Sqlite\Elsa.EntityFrameworkCore.Sqlite.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.SqlServer\Elsa.EntityFrameworkCore.SqlServer.csproj" />

3
src/bundles/Elsa.WorkflowServer.Web/Program.cs
View file

@ -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.

2
src/bundles/Elsa.WorkflowServer.Web/appsettings.json
View file

@ -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": {

17
src/modules/Elsa.Alterations.BackgroundRunner/Features/AlterationsBackgroundRunnerFeature.cs
View file

@ -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));
}
}

14
src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobDispatcher.cs Normal file
View file

@ -0,0 +1,14 @@
namespace Elsa.Alterations.Core.Contracts;
/// <summary>
/// Dispatches an alteration job for execution.
/// </summary>
public interface IAlterationJobDispatcher
{
/// <summary>
/// Dispatches an alteration job for execution.
/// </summary>
/// <param name="jobId">The ID of the job to dispatch.</param>
/// <param name="cancellationToken">An optional cancellation token.</param>
ValueTask DispatchAsync(string jobId, CancellationToken cancellationToken = default);
}

6
src/modules/Elsa.Alterations.Core/Contracts/IAlterationJobStore.cs
View file

@ -28,6 +28,12 @@ public interface IAlterationJobStore
/// </summary>
Task<IEnumerable<AlterationJob>> FindManyAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default);
/// <summary>
/// Finds the IDs of all alteration jobs matching the specified filter.
/// </summary>
/// <returns></returns>
Task<IEnumerable<string>> FindManyIdsAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default);
/// <summary>
/// Returns the number of alteration jobs matching the specified filter.
/// </summary>

5
src/modules/Elsa.Alterations.Core/Services/DefaultAlterationJobRunner.cs
View file

@ -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;

7
src/modules/Elsa.Alterations.Core/Stores/MemoryAlterationJobStore.cs
View file

@ -48,6 +48,13 @@ public class MemoryAlterationJobStore : IAlterationJobStore
return Task.FromResult(entities);
}
/// <inheritdoc />
public Task<IEnumerable<string>> FindManyIdsAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default)
{
var ids = _store.Query(query => Filter(query, filter)).Select(x => x.Id).ToList().AsEnumerable();
return Task.FromResult(ids);
}
/// <inheritdoc />
public Task<long> CountAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default)
{

27
src/modules/Elsa.Alterations.MassTransit/Consumers/RunAlterationJobConsumer.cs Normal file
View file

@ -0,0 +1,27 @@
using Elsa.Alterations.Core.Contracts;
using Elsa.Alterations.MassTransit.Messages;
using MassTransit;
namespace Elsa.Alterations.MassTransit.Consumers;
/// <summary>
/// Consumes <see cref="RunAlterationJob"/> messages.
/// </summary>
public class RunAlterationJobConsumer : IConsumer<RunAlterationJob>
{
private readonly IAlterationJobRunner _alterationJobRunner;
/// <summary>
/// Initializes a new instance of the <see cref="RunAlterationJobConsumer"/> class.
/// </summary>
public RunAlterationJobConsumer(IAlterationJobRunner alterationJobRunner)
{
_alterationJobRunner = alterationJobRunner;
}
/// <inheritdoc />
public async Task Consume(ConsumeContext<RunAlterationJob> context)
{
await _alterationJobRunner.RunAsync(context.Message.JobId, context.CancellationToken);
}
}

10
src/modules/Elsa.Alterations.BackgroundRunner/Elsa.Alterations.BackgroundRunner.csproj → src/modules/Elsa.Alterations.MassTransit/Elsa.Alterations.MassTransit.csproj
View file

@ -6,13 +6,17 @@
<PropertyGroup>
<TargetFrameworks>net6.0;net7.0</TargetFrameworks>
<Description>
Provides alteration background job runners.
Provides alteration background job runners using MassTransit.
</Description>
<PackageTags>elsa module alterations</PackageTags>
<PackageTags>elsa module alterations masstransit</PackageTags>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\Elsa.Alterations.Core\Elsa.Alterations.Core.csproj" />
<ProjectReference Include="..\Elsa.Alterations\Elsa.Alterations.csproj" />
<ProjectReference Include="..\Elsa.MassTransit\Elsa.MassTransit.csproj" />
</ItemGroup>
</Project>

20
src/modules/Elsa.Alterations.MassTransit/Extensions/ModuleExtensions.cs Normal file
View file

@ -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;
/// <summary>
/// Adds the <see cref="AlterationsFeature"/>.
/// </summary>
public static class ModuleExtensions
{
/// <summary>
/// Adds the <see cref="MassTransitAlterationsFeature"/>.
/// </summary>
public static IModule UseMassTransitDispatcher(this AlterationsFeature alterations, Action<MassTransitAlterationsFeature>? configure = default)
{
return alterations.Module.Use(configure);
}
}

37
src/modules/Elsa.Alterations.MassTransit/Features/AlterationsMassTransitFeature.cs Normal file
View file

@ -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;
/// <summary>
/// A feature for enabling the Alterations MassTransit feature.
/// </summary>
[DependsOn(typeof(MassTransitFeature))]
[DependsOn(typeof(AlterationsFeature))]
public class MassTransitAlterationsFeature : FeatureBase
{
/// <inheritdoc />
public MassTransitAlterationsFeature(IModule module) : base(module)
{
}
/// <inheritdoc />
public override void Configure()
{
Module.Configure<AlterationsFeature>(feature => feature.AlterationJobDispatcherFactory = sp => sp.GetRequiredService<MassTransitAlterationJobDispatcher>());
Module.AddMassTransitConsumer<RunAlterationJobConsumer>();
}
/// <inheritdoc />
public override void Apply()
{
Services.AddSingleton<MassTransitAlterationJobDispatcher>();
}
}

0
src/modules/Elsa.Alterations.BackgroundRunner/FodyWeavers.xml → src/modules/Elsa.Alterations.MassTransit/FodyWeavers.xml
View file

7
src/modules/Elsa.Alterations.MassTransit/Messages/RunAlterationJob.cs Normal file
View file

@ -0,0 +1,7 @@
namespace Elsa.Alterations.MassTransit.Messages;
/// <summary>
/// Represents a request to run an alteration job.
/// </summary>
/// <param name="JobId">The job ID.</param>
public record RunAlterationJob(string JobId);

4
src/modules/Elsa.Alterations.MassTransit/Readme.md Normal file
View file

@ -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.

28
src/modules/Elsa.Alterations.MassTransit/Services/MassTransitAlterationJobDispatcher.cs Normal file
View file

@ -0,0 +1,28 @@
using Elsa.Alterations.Core.Contracts;
using Elsa.Alterations.MassTransit.Messages;
using MassTransit;
namespace Elsa.Alterations.MassTransit.Services;
/// <summary>
/// Dispatches an alteration job for execution using MassTransit.
/// </summary>
public class MassTransitAlterationJobDispatcher : IAlterationJobDispatcher
{
private readonly IBus _bus;
/// <summary>
/// Initializes a new instance of the <see cref="MassTransitAlterationJobDispatcher"/> class.
/// </summary>
public MassTransitAlterationJobDispatcher(IBus bus)
{
_bus = bus;
}
/// <inheritdoc />
public async ValueTask DispatchAsync(string jobId, CancellationToken cancellationToken = default)
{
var message = new RunAlterationJob(jobId);
await _bus.Publish(message, cancellationToken);
}
}

11
src/modules/Elsa.Alterations/Features/AlterationsFeature.cs
View file

@ -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;
/// <summary>
/// Adds the Elsa alterations services.
/// </summary>
[DependsOn(typeof(AlterationsBackgroundRunnerFeature))]
public class AlterationsFeature : FeatureBase
{
/// <inheritdoc />
@ -33,6 +31,11 @@ public class AlterationsFeature : FeatureBase
/// Gets or sets the factory for the alteration job store.
/// </summary>
public Func<IServiceProvider, IAlterationJobStore> AlterationJobStoreFactory { get; set; } = sp => sp.GetRequiredService<MemoryAlterationJobStore>();
/// <summary>
/// Gets or sets the factory for the alteration job dispatcher.
/// </summary>
public Func<IServiceProvider, IAlterationJobDispatcher> AlterationJobDispatcherFactory { get; set; } = sp => sp.GetRequiredService<BackgroundAlterationJobDispatcher>();
/// <summary>
/// Adds an alteration and its handler.
@ -56,11 +59,13 @@ public class AlterationsFeature : FeatureBase
{
Services.AddAlterations();
Services.AddAlterationsCore();
Services.AddSingleton<BackgroundAlterationJobDispatcher>();
Services.AddSingleton<MemoryAlterationPlanStore>();
Services.AddSingleton<MemoryAlterationJobStore>();
Services.AddSingleton(new MemoryStore<AlterationPlan>());
Services.AddSingleton(new MemoryStore<AlterationJob>());
Services.AddSingleton(AlterationPlanStoreFactory);
Services.AddSingleton(AlterationJobStoreFactory);
Services.AddSingleton(AlterationJobDispatcherFactory);
}
}

4
src/modules/Elsa.Alterations.BackgroundRunner/Handlers/AlterationJobCompletedHandler.cs → src/modules/Elsa.Alterations/Handlers/AlterationJobCompletedHandler.cs
View file

@ -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;
/// <summary>
/// Handles <see cref="AlterationJobCompleted"/> notifications and updates the plan status if all jobs are completed.
/// </summary>
[UsedImplicitly]
public class AlterationJobCompletedHandler : INotificationHandler<AlterationJobCompleted>
{
private readonly IAlterationPlanStore _alterationPlanStore;

31
src/modules/Elsa.Alterations.BackgroundRunner/Handlers/AlterationPlanSubmittedHandler.cs → src/modules/Elsa.Alterations/Handlers/AlterationPlanSubmittedHandler.cs
View file

@ -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;
/// <summary>
/// Handles <see cref="AlterationPlanSubmitted"/> notifications and runs the plan in the background.
/// </summary>
[UsedImplicitly]
public class AlterationPlanSubmittedHandler : INotificationHandler<AlterationPlanSubmitted>
{
private readonly IJobQueue _jobQueue;
private readonly IAlterationPlanStore _alterationPlanStore;
private readonly IAlterationJobStore _alterationJobStore;
private readonly IAlterationJobRunner _alterationJobRunner;
private readonly IAlterationJobDispatcher _alterationJobDispatcher;
private readonly IJobQueue _jobQueue;
/// <summary>
/// Initializes a new instance of the <see cref="AlterationPlanSubmittedHandler"/> class.
/// </summary>
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;
}
/// <inheritdoc />
@ -47,17 +53,10 @@ public class AlterationPlanSubmittedHandler : INotificationHandler<AlterationPla
// Create a job for each job in the plan.
var filter = new AlterationJobFilter { PlanId = plan.Id };
var alterationJobs = await _alterationJobStore.FindManyAsync(filter, cancellationToken);
var alterationJobIds = await _alterationJobStore.FindManyIdsAsync(filter, cancellationToken);
// Enqueue each job.
foreach (var alterationJob in alterationJobs)
{
_jobQueue.Enqueue(ct => 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);
}
}

2
src/modules/Elsa.Alterations/Handlers/ResumeWorkflowInstance.cs
View file

@ -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;
/// <summary>
/// Resumes a workflow instance when an alteration job has been completed and the workflow contains scheduled work.
/// </summary>
[UsedImplicitly]
public class ResumeWorkflowInstance : INotificationHandler<AlterationJobCompleted>
{
private readonly IWorkflowDispatcher _workflowDispatcher;

34
src/modules/Elsa.Alterations/Services/BackgroundAlterationJobDispatcher.cs Normal file
View file

@ -0,0 +1,34 @@
using Elsa.Alterations.Core.Contracts;
using Elsa.Mediator.Contracts;
namespace Elsa.Alterations.Services;
/// <summary>
/// Dispatches an alteration job for execution using an in-memory channel.
/// </summary>
public class BackgroundAlterationJobDispatcher : IAlterationJobDispatcher
{
private readonly IJobQueue _jobQueue;
private readonly IAlterationJobRunner _alterationJobRunner;
/// <summary>
/// Initializes a new instance of the <see cref="BackgroundAlterationJobDispatcher"/> class.
/// </summary>
public BackgroundAlterationJobDispatcher(IJobQueue jobQueue, IAlterationJobRunner alterationJobRunner)
{
_jobQueue = jobQueue;
_alterationJobRunner = alterationJobRunner;
}
/// <inheritdoc />
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);
}
}

6
src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/AlterationJobStore.cs
View file

@ -47,6 +47,12 @@ public class EFCoreAlterationJobStore : IAlterationJobStore
return await _store.QueryAsync(filter.Apply, OnLoadAsync, cancellationToken).ToList();
}
/// <inheritdoc />
public async Task<IEnumerable<string>> FindManyIdsAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default)
{
return await _store.QueryAsync(filter.Apply, x => x.Id, cancellationToken);
}
/// <inheritdoc />
public async Task<long> CountAsync(AlterationJobFilter filter, CancellationToken cancellationToken = default)
{

2
src/modules/Elsa.EntityFrameworkCore/Modules/Alterations/Configurations.cs
View file

@ -16,6 +16,7 @@ public class Configurations : IEntityTypeConfiguration<AlterationPlan>, IEntityT
builder.Ignore(x => x.WorkflowInstanceIds);
builder.Property<string>("SerializedAlterations");
builder.Property<string>("SerializedWorkflowInstanceIds");
builder.Property(x => x.Status).HasConversion<string>();
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<AlterationPlan>, IEntityT
{
builder.Ignore(x => x.Log);
builder.Property<string>("SerializedLog");
builder.Property(x => x.Status).HasConversion<string>();
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)}");

10
src/modules/Elsa.MassTransit/Extensions/ModuleExtensions.cs
View file

@ -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
/// </summary>
public static IModule UseMassTransit(this IModule module, Action<MassTransitFeature>? configure = default) => module.Use(configure);
/// <summary>
/// Registers the specified consumer with MassTransit.
/// </summary>
public static IModule AddMassTransitConsumer<T>(this IModule module) where T : IConsumer
{
module.Configure<MassTransitFeature>(massTransit => massTransit.AddConsumer<T>());
return module;
}
/// <summary>
/// Enable and configure the RabbitMQ broker for MassTransit.
/// </summary>

2
src/modules/Elsa.MassTransit/Features/MassTransitWorkflowDispatcherFeature.cs
View file

@ -25,7 +25,7 @@ public class MassTransitWorkflowDispatcherFeature : FeatureBase
/// <inheritdoc />
public override void Configure()
{
Module.Configure<MassTransitFeature>(massTransit => massTransit.AddConsumer<DispatchWorkflowRequestConsumer>());
Module.AddMassTransitConsumer<DispatchWorkflowRequestConsumer>();
Module.Configure<WorkflowRuntimeFeature>(f => f.WorkflowDispatcher = sp => sp.GetRequiredService<MassTransitWorkflowDispatcher>());
}