Implement workflow inbox cleanup service (#4327)
This commit is contained in:
Sipke Schoorstra
GitHub
parent
af34276d60
commit
f4ae684edd
|
|
@ -14,7 +14,7 @@ using Proto.Persistence.Sqlite;
|
|||
|
||||
const bool useMongoDb = false;
|
||||
const bool useProtoActor = false;
|
||||
const bool useHangfire = true;
|
||||
const bool useHangfire = false;
|
||||
|
||||
var builder = WebApplication.CreateBuilder(args);
|
||||
var services = builder.Services;
|
||||
|
|
@ -90,6 +90,8 @@ services
|
|||
e.UseEntityFrameworkCore();
|
||||
});
|
||||
runtime.UseMassTransitDispatcher();
|
||||
|
||||
runtime.WorkflowInboxCleanupOptions = options => configuration.GetSection("Runtime:WorkflowInboxCleanup").Bind(options);
|
||||
})
|
||||
.UseEnvironments(environments => environments.EnvironmentsOptions = options => configuration.GetSection("Environments").Bind(options))
|
||||
.UseScheduling(scheduling =>
|
||||
|
|
@ -99,13 +101,12 @@ services
|
|||
})
|
||||
.UseWorkflowsApi(api => api.AddFastEndpointsAssembly<Program>())
|
||||
.UseRealTimeWorkflows()
|
||||
.UseJavaScript()
|
||||
.UseJavaScript(js => js.JintOptions = options => options.AllowClrAccess = true)
|
||||
.UseLiquid()
|
||||
.UseHttp(http => http.HttpEndpointAuthorizationHandler = sp => sp.GetRequiredService<AllowAnonymousHttpEndpointAuthorizationHandler>())
|
||||
.UseEmail(email => email.ConfigureOptions = options => configuration.GetSection("Smtp").Bind(options));
|
||||
});
|
||||
|
||||
services.Configure<JintOptions>(options => options.AllowClrAccess = true);
|
||||
services.AddHealthChecks();
|
||||
services.AddCors(cors => cors.AddDefaultPolicy(policy => policy.AllowAnyHeader().AllowAnyMethod().AllowAnyOrigin().WithExposedHeaders("*")));
|
||||
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
"LogLevel": {
|
||||
"Default": "Warning",
|
||||
"Elsa.Mediator": "Warning",
|
||||
"Elsa.Workflows.Runtime.HostedServices": "Information",
|
||||
"MassTransit": "Warning",
|
||||
"Microsoft.Extensions.Http": "Warning",
|
||||
"Microsoft.Hosting.Lifetime": "Information",
|
||||
|
|
@ -78,5 +79,11 @@
|
|||
"ServerUrl": "https://production.acme.com/elsa/api"
|
||||
}
|
||||
]
|
||||
},
|
||||
"Runtime": {
|
||||
"WorkflowInboxCleanup": {
|
||||
"SweepInterval": "00:00:10:00",
|
||||
"BatchSize": 1000
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -53,7 +53,7 @@ public static class QueryableExtensions
|
|||
/// <param name="pageArgs">The pagination arguments.</param>
|
||||
/// <typeparam name="T">The type of the queryable.</typeparam>
|
||||
/// <returns>The paginated queryable.</returns>
|
||||
public static IQueryable<T> Paginate<T>(this IQueryable<T> queryable, PageArgs pageArgs)
|
||||
public static IQueryable<T> Paginate<T>(this IQueryable<T> queryable, PageArgs? pageArgs)
|
||||
{
|
||||
if (pageArgs?.Offset != null) queryable = queryable.Skip(pageArgs.Offset.Value);
|
||||
if (pageArgs?.Limit != null) queryable = queryable.Take(pageArgs.Limit.Value);
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ public class DapperWorkflowRuntimePersistenceFeature : FeatureBase
|
|||
{
|
||||
feature.TriggerStore = sp => sp.GetRequiredService<DapperTriggerStore>();
|
||||
feature.BookmarkStore = sp => sp.GetRequiredService<DapperBookmarkStore>();
|
||||
feature.WorkflowInboxStore = sp => sp.GetRequiredService<DapperWorkflowInboxStore>();
|
||||
feature.WorkflowInboxStore = sp => sp.GetRequiredService<DapperWorkflowInboxMessageStore>();
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -39,6 +39,6 @@ public class DapperWorkflowRuntimePersistenceFeature : FeatureBase
|
|||
|
||||
Services.AddSingleton<DapperTriggerStore>();
|
||||
Services.AddSingleton<DapperBookmarkStore>();
|
||||
Services.AddSingleton<DapperWorkflowInboxStore>();
|
||||
Services.AddSingleton<DapperWorkflowInboxMessageStore>();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,4 +1,7 @@
|
|||
using System.Text.Json;
|
||||
using Elsa.Common.Contracts;
|
||||
using Elsa.Common.Entities;
|
||||
using Elsa.Common.Models;
|
||||
using Elsa.Dapper.Contracts;
|
||||
using Elsa.Dapper.Extensions;
|
||||
using Elsa.Dapper.Models;
|
||||
|
|
@ -14,19 +17,21 @@ namespace Elsa.Dapper.Modules.Runtime.Stores;
|
|||
/// <summary>
|
||||
/// A Dapper-based <see cref="IBookmarkStore"/> implementation.
|
||||
/// </summary>
|
||||
public class DapperWorkflowInboxStore : IWorkflowInboxStore
|
||||
public class DapperWorkflowInboxMessageStore : IWorkflowInboxMessageStore
|
||||
{
|
||||
private readonly IPayloadSerializer _payloadSerializer;
|
||||
private readonly ISystemClock _systemClock;
|
||||
private const string TableName = "WorkflowInboxMessages";
|
||||
private const string PrimaryKeyName = "Id";
|
||||
private readonly Store<WorkflowInboxMessageRecord> _store;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="DapperWorkflowInboxStore"/> class.
|
||||
/// Initializes a new instance of the <see cref="DapperWorkflowInboxMessageStore"/> class.
|
||||
/// </summary>
|
||||
public DapperWorkflowInboxStore(IDbConnectionProvider dbConnectionProvider, IPayloadSerializer payloadSerializer)
|
||||
public DapperWorkflowInboxMessageStore(IDbConnectionProvider dbConnectionProvider, IPayloadSerializer payloadSerializer, ISystemClock systemClock)
|
||||
{
|
||||
_payloadSerializer = payloadSerializer;
|
||||
_systemClock = systemClock;
|
||||
_store = new Store<WorkflowInboxMessageRecord>(dbConnectionProvider, TableName, PrimaryKeyName);
|
||||
}
|
||||
|
||||
|
|
@ -52,9 +57,12 @@ public class DapperWorkflowInboxStore : IWorkflowInboxStore
|
|||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<long> DeleteAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default)
|
||||
public async ValueTask<long> DeleteManyAsync(WorkflowInboxMessageFilter filter, PageArgs? pageArgs = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return await _store.DeleteAsync(q => ApplyFilter(q, filter), cancellationToken);
|
||||
if (pageArgs == null)
|
||||
return await _store.DeleteAsync(q => ApplyFilter(q, filter), cancellationToken);
|
||||
|
||||
return await _store.DeleteAsync(q => ApplyFilter(q, filter), pageArgs, new[] { new OrderField(nameof(WorkflowInboxMessage.CreatedAt), OrderDirection.Ascending) }, cancellationToken);
|
||||
}
|
||||
|
||||
private void ApplyFilter(ParameterizedQuery query, params WorkflowInboxMessageFilter[] filters)
|
||||
|
|
@ -231,7 +231,7 @@ public class Store<T> where T : notnull
|
|||
/// <param name="filter">The conditions to apply to the query.</param>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
/// <returns>The number of records deleted.</returns>
|
||||
public async Task<int> DeleteAsync(Action<ParameterizedQuery> filter, CancellationToken cancellationToken = default)
|
||||
public async Task<long> DeleteAsync(Action<ParameterizedQuery> filter, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var query = _dbConnectionProvider.CreateQuery().Delete(TableName);
|
||||
filter(query);
|
||||
|
|
@ -239,6 +239,23 @@ public class Store<T> where T : notnull
|
|||
return await query.ExecuteAsync(connection);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deletes all records matching the specified query.
|
||||
/// </summary>
|
||||
/// <param name="filter">The conditions to apply to the query.</param>
|
||||
/// <param name="pageArgs">The page arguments.</param>
|
||||
/// <param name="orderFields">The fields by which to order the results.</param>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
/// <returns>The number of records deleted.</returns>
|
||||
public async Task<long> DeleteAsync(Action<ParameterizedQuery> filter, PageArgs pageArgs, IEnumerable<OrderField> orderFields, CancellationToken cancellationToken = default)
|
||||
{
|
||||
using var connection = _dbConnectionProvider.GetConnection();
|
||||
var query = _dbConnectionProvider.CreateQuery().Delete(TableName);
|
||||
filter(query);
|
||||
query = query.OrderBy(orderFields.ToArray()).Page(pageArgs);
|
||||
return await query.ExecuteAsync(connection);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns <c>true</c> if any records match the specified query.
|
||||
/// </summary>
|
||||
|
|
|
|||
|
|
@ -1,4 +1,6 @@
|
|||
using System.Text.Json;
|
||||
using Elsa.Common.Contracts;
|
||||
using Elsa.Common.Models;
|
||||
using Elsa.EntityFrameworkCore.Common;
|
||||
using Elsa.Workflows.Core.Contracts;
|
||||
using Elsa.Workflows.Runtime.Contracts;
|
||||
|
|
@ -10,39 +12,47 @@ namespace Elsa.EntityFrameworkCore.Modules.Runtime;
|
|||
/// <summary>
|
||||
/// An EF Core implementation of <see cref="IBookmarkStore"/>.
|
||||
/// </summary>
|
||||
public class EFCoreWorkflowInboxStore : IWorkflowInboxStore
|
||||
public class EFCoreWorkflowInboxMessageStore : IWorkflowInboxMessageStore
|
||||
{
|
||||
private readonly EntityStore<RuntimeElsaDbContext, WorkflowInboxMessage> _store;
|
||||
private readonly IPayloadSerializer _payloadSerializer;
|
||||
private readonly ISystemClock _systemClock;
|
||||
|
||||
/// <summary>
|
||||
/// Constructor.
|
||||
/// </summary>
|
||||
public EFCoreWorkflowInboxStore(EntityStore<RuntimeElsaDbContext, WorkflowInboxMessage> store, IPayloadSerializer payloadSerializer)
|
||||
public EFCoreWorkflowInboxMessageStore(EntityStore<RuntimeElsaDbContext, WorkflowInboxMessage> store, IPayloadSerializer payloadSerializer, ISystemClock systemClock)
|
||||
{
|
||||
_store = store;
|
||||
_payloadSerializer = payloadSerializer;
|
||||
_systemClock = systemClock;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask SaveAsync(WorkflowInboxMessage record, CancellationToken cancellationToken = default) => await _store.SaveAsync(record, OnSaveAsync, cancellationToken);
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default) => await _store.QueryAsync(filter.Apply, OnLoadAsync, cancellationToken);
|
||||
public async ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return await _store.QueryAsync(q => filter.Apply(q, _systemClock.UtcNow), OnLoadAsync, cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(IEnumerable<WorkflowInboxMessageFilter> filters, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return await _store.QueryAsync(query =>
|
||||
{
|
||||
foreach (var filter in filters) filter.Apply(query);
|
||||
foreach (var filter in filters) filter.Apply(query, _systemClock.UtcNow);
|
||||
return query;
|
||||
}, OnLoadAsync, cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<long> DeleteAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default) => await _store.DeleteWhereAsync(filter.Apply, cancellationToken);
|
||||
|
||||
public async ValueTask<long> DeleteManyAsync(WorkflowInboxMessageFilter filter, PageArgs? pageArgs = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return await _store.DeleteWhereAsync(q => Paginate(filter.Apply(q, _systemClock.UtcNow), pageArgs), cancellationToken);
|
||||
}
|
||||
|
||||
private ValueTask OnSaveAsync(RuntimeElsaDbContext dbContext, WorkflowInboxMessage entity, CancellationToken cancellationToken)
|
||||
{
|
||||
dbContext.Entry(entity).Property("SerializedBookmarkPayload").CurrentValue = _payloadSerializer.Serialize(entity.BookmarkPayload);
|
||||
|
|
@ -59,11 +69,18 @@ public class EFCoreWorkflowInboxStore : IWorkflowInboxStore
|
|||
var bookmarkPayloadJson = dbContext.Entry(entity).Property<string>("SerializedBookmarkPayload").CurrentValue;
|
||||
var inputJson = dbContext.Entry(entity).Property<string>("SerializedInput").CurrentValue;
|
||||
var affectedWorkflowInstancesIdsJson = dbContext.Entry(entity).Property<string>("SerializedAffectedWorkflowInstancesIds").CurrentValue;
|
||||
|
||||
|
||||
entity.BookmarkPayload = _payloadSerializer.Deserialize(bookmarkPayloadJson);
|
||||
entity.Input = !string.IsNullOrEmpty(inputJson) ? _payloadSerializer.Deserialize<Dictionary<string, object>>(inputJson) : null;
|
||||
entity.AffectedWorkflowInstancesIds = !string.IsNullOrEmpty(affectedWorkflowInstancesIdsJson) ? JsonSerializer.Deserialize<List<string>>(affectedWorkflowInstancesIdsJson) : null;
|
||||
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
private static IQueryable<WorkflowInboxMessage> Paginate(IQueryable<WorkflowInboxMessage> queryable, PageArgs? pageArgs)
|
||||
{
|
||||
if (pageArgs?.Offset != null) queryable = queryable.Skip(pageArgs.Offset.Value);
|
||||
if (pageArgs?.Limit != null) queryable = queryable.Take(pageArgs.Limit.Value);
|
||||
return queryable;
|
||||
}
|
||||
}
|
||||
|
|
@ -25,7 +25,7 @@ public class EFCoreWorkflowRuntimePersistenceFeature : PersistenceFeatureBase<Ru
|
|||
{
|
||||
feature.TriggerStore = sp => sp.GetRequiredService<EFCoreTriggerStore>();
|
||||
feature.BookmarkStore = sp => sp.GetRequiredService<EFCoreBookmarkStore>();
|
||||
feature.WorkflowInboxStore = sp => sp.GetRequiredService<EFCoreWorkflowInboxStore>();
|
||||
feature.WorkflowInboxStore = sp => sp.GetRequiredService<EFCoreWorkflowInboxMessageStore>();
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -36,6 +36,6 @@ public class EFCoreWorkflowRuntimePersistenceFeature : PersistenceFeatureBase<Ru
|
|||
|
||||
AddEntityStore<StoredTrigger, EFCoreTriggerStore>();
|
||||
AddStore<StoredBookmark, EFCoreBookmarkStore>();
|
||||
AddEntityStore<WorkflowInboxMessage, EFCoreWorkflowInboxStore>();
|
||||
AddEntityStore<WorkflowInboxMessage, EFCoreWorkflowInboxMessageStore>();
|
||||
}
|
||||
}
|
||||
|
|
@ -8,6 +8,7 @@ using Elsa.Features.Services;
|
|||
using Elsa.JavaScript.Contracts;
|
||||
using Elsa.JavaScript.Expressions;
|
||||
using Elsa.JavaScript.Extensions;
|
||||
using Elsa.JavaScript.Options;
|
||||
using Elsa.JavaScript.Providers;
|
||||
using Elsa.JavaScript.Services;
|
||||
using Elsa.JavaScript.TypeDefinitions.Contracts;
|
||||
|
|
@ -28,10 +29,17 @@ public class JavaScriptFeature : FeatureBase
|
|||
public JavaScriptFeature(IModule module) : base(module)
|
||||
{
|
||||
}
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// Configures the Jint options.
|
||||
/// </summary>
|
||||
public Action<JintOptions> JintOptions { get; set; } = _ => { };
|
||||
|
||||
/// <inheritdoc />
|
||||
public override void Apply()
|
||||
{
|
||||
Services.Configure(JintOptions);
|
||||
|
||||
// JavaScript services.
|
||||
Services
|
||||
.AddSingleton<IExpressionSyntaxProvider, JavaScriptExpressionSyntaxProvider>()
|
||||
|
|
|
|||
|
|
@ -1,3 +1,6 @@
|
|||
using Elsa.Common.Contracts;
|
||||
using Elsa.Common.Models;
|
||||
using Elsa.Extensions;
|
||||
using Elsa.MongoDb.Common;
|
||||
using Elsa.Workflows.Runtime.Contracts;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
|
|
@ -9,16 +12,18 @@ namespace Elsa.MongoDb.Modules.Runtime;
|
|||
/// <summary>
|
||||
/// A MongoDb implementation of <see cref="IBookmarkStore"/>.
|
||||
/// </summary>
|
||||
public class MongoWorkflowInboxStore : IWorkflowInboxStore
|
||||
public class MongoWorkflowInboxMessageStore : IWorkflowInboxMessageStore
|
||||
{
|
||||
private readonly MongoDbStore<WorkflowInboxMessage> _mongoDbStore;
|
||||
private readonly ISystemClock _systemClock;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="MongoWorkflowInboxStore"/> class.
|
||||
/// Initializes a new instance of the <see cref="MongoWorkflowInboxMessageStore"/> class.
|
||||
/// </summary>
|
||||
public MongoWorkflowInboxStore(MongoDbStore<WorkflowInboxMessage> mongoDbStore)
|
||||
public MongoWorkflowInboxMessageStore(MongoDbStore<WorkflowInboxMessage> mongoDbStore, ISystemClock systemClock)
|
||||
{
|
||||
_mongoDbStore = mongoDbStore;
|
||||
_systemClock = systemClock;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
|
@ -38,12 +43,14 @@ public class MongoWorkflowInboxStore : IWorkflowInboxStore
|
|||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<long> DeleteAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default) =>
|
||||
await _mongoDbStore.DeleteWhereAsync<string>(query => Filter(query, filter), x => x.Id, cancellationToken);
|
||||
public async ValueTask<long> DeleteManyAsync(WorkflowInboxMessageFilter filter, PageArgs? pageArgs = default, CancellationToken cancellationToken = default) =>
|
||||
await _mongoDbStore.DeleteWhereAsync<string>(query => Paginate(Filter(query, filter), pageArgs), x => x.Id, cancellationToken);
|
||||
|
||||
private static IMongoQueryable<WorkflowInboxMessage> Filter(IMongoQueryable<WorkflowInboxMessage> queryable, params WorkflowInboxMessageFilter[] filters)
|
||||
private IMongoQueryable<WorkflowInboxMessage> Filter(IMongoQueryable<WorkflowInboxMessage> queryable, params WorkflowInboxMessageFilter[] filters)
|
||||
{
|
||||
foreach (var filter in filters) filter.Apply(queryable);
|
||||
foreach (var filter in filters) filter.Apply(queryable, _systemClock.UtcNow);
|
||||
return queryable;
|
||||
}
|
||||
|
||||
private IMongoQueryable<WorkflowInboxMessage> Paginate(IMongoQueryable<WorkflowInboxMessage> queryable, PageArgs? pageArgs) => (queryable.Paginate(pageArgs) as IMongoQueryable<WorkflowInboxMessage>)!;
|
||||
}
|
||||
|
|
@ -26,7 +26,7 @@ public class MongoWorkflowRuntimePersistenceFeature : PersistenceFeatureBase
|
|||
feature.TriggerStore = sp => sp.GetRequiredService<MongoTriggerStore>();
|
||||
feature.BookmarkStore = sp => sp.GetRequiredService<MongoBookmarkStore>();
|
||||
feature.ActivityExecutionLogStore = sp => sp.GetRequiredService<MongoActivityExecutionLogStore>();
|
||||
feature.WorkflowInboxStore = sp => sp.GetRequiredService<MongoWorkflowInboxStore>();
|
||||
feature.WorkflowInboxStore = sp => sp.GetRequiredService<MongoWorkflowInboxMessageStore>();
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -43,7 +43,7 @@ public class MongoWorkflowRuntimePersistenceFeature : PersistenceFeatureBase
|
|||
AddStore<StoredTrigger, MongoTriggerStore>();
|
||||
AddStore<StoredBookmark, MongoBookmarkStore>();
|
||||
AddStore<ActivityExecutionRecord, MongoActivityExecutionLogStore>();
|
||||
AddStore<WorkflowInboxMessage, MongoWorkflowInboxStore>();
|
||||
AddStore<WorkflowInboxMessage, MongoWorkflowInboxMessageStore>();
|
||||
|
||||
Services.AddHostedService<CreateIndices>();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
using Elsa.Common.Models;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
using Elsa.Workflows.Runtime.Filters;
|
||||
|
||||
|
|
@ -6,7 +7,7 @@ namespace Elsa.Workflows.Runtime.Contracts;
|
|||
/// <summary>
|
||||
/// A store for workflow inbox messages.
|
||||
/// </summary>
|
||||
public interface IWorkflowInboxStore
|
||||
public interface IWorkflowInboxMessageStore
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds a message to the store.
|
||||
|
|
@ -37,7 +38,8 @@ public interface IWorkflowInboxStore
|
|||
/// Deletes all messages matching the specified filter.
|
||||
/// </summary>
|
||||
/// <param name="filter">The filter to apply.</param>
|
||||
/// <param name="pageArgs">An optional page arguments.</param>
|
||||
/// <param name="cancellationToken">An optional cancellation token.</param>
|
||||
/// <returns>The number of deleted messages.</returns>
|
||||
ValueTask<long> DeleteAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default);
|
||||
ValueTask<long> DeleteManyAsync(WorkflowInboxMessageFilter filter, PageArgs? pageArgs = default, CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
|
@ -25,4 +25,5 @@
|
|||
|
||||
|
||||
|
||||
|
||||
</Project>
|
||||
|
|
|
|||
|
|
@ -71,9 +71,9 @@ public class WorkflowRuntimeFeature : FeatureBase
|
|||
public Func<IServiceProvider, IActivityExecutionStore> ActivityExecutionLogStore { get; set; } = sp => sp.GetRequiredService<NoopActivityExecutionStore>();
|
||||
|
||||
/// <summary>
|
||||
/// A factory that instantiates an <see cref="IWorkflowInboxStore"/>.
|
||||
/// A factory that instantiates an <see cref="IWorkflowInboxMessageStore"/>.
|
||||
/// </summary>
|
||||
public Func<IServiceProvider, IWorkflowInboxStore> WorkflowInboxStore { get; set; } = sp => sp.GetRequiredService<MemoryWorkflowInboxStore>();
|
||||
public Func<IServiceProvider, IWorkflowInboxMessageStore> WorkflowInboxStore { get; set; } = sp => sp.GetRequiredService<MemoryWorkflowInboxMessageStore>();
|
||||
|
||||
/// <summary>
|
||||
/// A factory that instantiates an <see cref="IDistributedLockProvider"/>.
|
||||
|
|
@ -99,6 +99,11 @@ public class WorkflowRuntimeFeature : FeatureBase
|
|||
/// A delegate to configure the <see cref="DistributedLockingOptions"/>.
|
||||
/// </summary>
|
||||
public Action<DistributedLockingOptions> DistributedLockingOptions { get; set; } = _ => { };
|
||||
|
||||
/// <summary>
|
||||
/// A delegate to configure the <see cref="WorkflowInboxCleanupOptions"/>.
|
||||
/// </summary>
|
||||
public Action<WorkflowInboxCleanupOptions> WorkflowInboxCleanupOptions { get; set; } = _ => { };
|
||||
|
||||
/// <summary>
|
||||
/// Register the specified workflow type.
|
||||
|
|
@ -135,7 +140,11 @@ public class WorkflowRuntimeFeature : FeatureBase
|
|||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override void ConfigureHostedServices() => Module.ConfigureHostedService<PopulateRegistriesHostedService>();
|
||||
public override void ConfigureHostedServices()
|
||||
{
|
||||
Module.ConfigureHostedService<PopulateRegistriesHostedService>();
|
||||
Module.ConfigureHostedService<WorkflowInboxCleanupHostedService>();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override void Apply()
|
||||
|
|
@ -143,6 +152,7 @@ public class WorkflowRuntimeFeature : FeatureBase
|
|||
// Options.
|
||||
Services.Configure(DistributedLockingOptions);
|
||||
Services.Configure<RuntimeOptions>(options => { options.Workflows = Workflows; });
|
||||
Services.Configure(WorkflowInboxCleanupOptions);
|
||||
|
||||
Services
|
||||
// Core.
|
||||
|
|
@ -183,7 +193,7 @@ public class WorkflowRuntimeFeature : FeatureBase
|
|||
.AddMemoryStore<StoredTrigger, MemoryTriggerStore>()
|
||||
.AddMemoryStore<WorkflowExecutionLogRecord, MemoryWorkflowExecutionLogStore>()
|
||||
.AddMemoryStore<ActivityExecutionRecord, MemoryActivityExecutionStore>()
|
||||
.AddMemoryStore<WorkflowInboxMessage, MemoryWorkflowInboxStore>()
|
||||
.AddMemoryStore<WorkflowInboxMessage, MemoryWorkflowInboxMessageStore>()
|
||||
|
||||
// Distributed locking.
|
||||
.AddSingleton(DistributedLockProvider)
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
using Elsa.Common.Contracts;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
|
||||
namespace Elsa.Workflows.Runtime.Filters;
|
||||
|
|
@ -37,11 +38,16 @@ public class WorkflowInboxMessageFilter
|
|||
/// </summary>
|
||||
public bool? IsHandled { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// A flag indicating whether to filter by messages that have expired.
|
||||
/// </summary>
|
||||
public bool? IsExpired { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Applies the filter to the specified query.
|
||||
/// </summary>
|
||||
/// <returns>The filtered query.</returns>
|
||||
public IQueryable<WorkflowInboxMessage> Apply(IQueryable<WorkflowInboxMessage> query)
|
||||
public IQueryable<WorkflowInboxMessage> Apply(IQueryable<WorkflowInboxMessage> query, DateTimeOffset now)
|
||||
{
|
||||
var filter = this;
|
||||
if (filter.CorrelationId != null) query = query.Where(x => filter.CorrelationId == x.CorrelationId);
|
||||
|
|
@ -50,6 +56,7 @@ public class WorkflowInboxMessageFilter
|
|||
if (filter.Hash != null) query = query.Where(x => filter.Hash == x.Hash);
|
||||
if (filter.ActivityTypeName != null) query = query.Where(x => filter.ActivityTypeName == x.ActivityTypeName);
|
||||
if (filter.ActivityInstanceId != null) query = query.Where(x => filter.ActivityInstanceId == x.ActivityInstanceId);
|
||||
if (filter.IsExpired != null) query = query.Where(x => x.ExpiresAt <= now);
|
||||
|
||||
return query;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,64 @@
|
|||
using Elsa.Common.Models;
|
||||
using Elsa.Workflows.Runtime.Contracts;
|
||||
using Elsa.Workflows.Runtime.Filters;
|
||||
using Elsa.Workflows.Runtime.Options;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace Elsa.Workflows.Runtime.HostedServices;
|
||||
|
||||
/// <summary>
|
||||
/// Cleans up expired messages from the database.
|
||||
/// </summary>
|
||||
public class WorkflowInboxCleanupHostedService : BackgroundService
|
||||
{
|
||||
private readonly IWorkflowInboxMessageStore _workflowInboxMessageStore;
|
||||
private readonly IOptions<WorkflowInboxCleanupOptions> _options;
|
||||
private readonly ILogger _logger;
|
||||
|
||||
/// <inheritdoc />
|
||||
public WorkflowInboxCleanupHostedService(
|
||||
IWorkflowInboxMessageStore workflowInboxMessageStore,
|
||||
IOptions<WorkflowInboxCleanupOptions> options,
|
||||
ILogger<WorkflowInboxCleanupHostedService> logger)
|
||||
{
|
||||
_workflowInboxMessageStore = workflowInboxMessageStore;
|
||||
_options = options;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogInformation("Entering expired workflow inbox messages cleanup service loop");
|
||||
|
||||
try
|
||||
{
|
||||
await CleanupExpiredMessages(stoppingToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "An error occurred while cleaning up expired workflow inbox messages");
|
||||
}
|
||||
|
||||
var sweepInterval = _options.Value.SweepInterval;
|
||||
_logger.LogInformation("Expired messages cleanup service is going to sleep for {SweepInterval} minutes", sweepInterval.TotalMinutes);
|
||||
await Task.Delay(sweepInterval, stoppingToken);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task CleanupExpiredMessages(CancellationToken cancellationToken)
|
||||
{
|
||||
var filter = new WorkflowInboxMessageFilter
|
||||
{
|
||||
IsExpired = true
|
||||
};
|
||||
|
||||
var pageArgs = PageArgs.FromRange(0, _options.Value.BatchSize);
|
||||
var deleteCount = await _workflowInboxMessageStore.DeleteManyAsync(filter, pageArgs, cancellationToken);
|
||||
_logger.LogInformation("Cleaned up {DeleteCount} expired messages", deleteCount);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
namespace Elsa.Workflows.Runtime.Options;
|
||||
|
||||
/// <summary>
|
||||
/// Options for cleaning up the workflow inbox.
|
||||
/// </summary>
|
||||
public class WorkflowInboxCleanupOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// The sweep interval at which to clean up the workflow inbox.
|
||||
/// </summary>
|
||||
public TimeSpan SweepInterval { get; set; } = TimeSpan.FromMinutes(5);
|
||||
|
||||
/// <summary>
|
||||
/// The number of messages to clean up per sweep.
|
||||
/// </summary>
|
||||
public int BatchSize { get; set; } = 1000;
|
||||
}
|
||||
|
|
@ -1,3 +1,5 @@
|
|||
namespace Elsa.Workflows.Runtime.Contracts;
|
||||
using Elsa.Workflows.Runtime.Contracts;
|
||||
|
||||
namespace Elsa.Workflows.Runtime.Results;
|
||||
|
||||
public record DeliverWorkflowInboxMessageResult(ICollection<WorkflowExecutionResult> WorkflowExecutionResults);
|
||||
|
|
@ -15,7 +15,7 @@ namespace Elsa.Workflows.Runtime.Services;
|
|||
public class DefaultWorkflowInbox : IWorkflowInbox
|
||||
{
|
||||
private readonly IWorkflowRuntime _workflowRuntime;
|
||||
private readonly IWorkflowInboxStore _store;
|
||||
private readonly IWorkflowInboxMessageStore _messageStore;
|
||||
private readonly INotificationSender _notificationSender;
|
||||
private readonly ISystemClock _systemClock;
|
||||
private readonly IIdentityGenerator _identityGenerator;
|
||||
|
|
@ -26,14 +26,14 @@ public class DefaultWorkflowInbox : IWorkflowInbox
|
|||
/// </summary>
|
||||
public DefaultWorkflowInbox(
|
||||
IWorkflowRuntime workflowRuntime,
|
||||
IWorkflowInboxStore store,
|
||||
IWorkflowInboxMessageStore messageStore,
|
||||
INotificationSender notificationSender,
|
||||
ISystemClock systemClock,
|
||||
IIdentityGenerator identityGenerator,
|
||||
IBookmarkHasher bookmarkHasher)
|
||||
{
|
||||
_workflowRuntime = workflowRuntime;
|
||||
_store = store;
|
||||
_messageStore = messageStore;
|
||||
_notificationSender = notificationSender;
|
||||
_systemClock = systemClock;
|
||||
_identityGenerator = identityGenerator;
|
||||
|
|
@ -68,7 +68,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox
|
|||
};
|
||||
|
||||
// Store the message.
|
||||
await _store.SaveAsync(message, cancellationToken);
|
||||
await _messageStore.SaveAsync(message, cancellationToken);
|
||||
|
||||
// Send a notification.
|
||||
var strategy = options.EventPublishingStrategy;
|
||||
|
|
@ -91,7 +91,7 @@ public class DefaultWorkflowInbox : IWorkflowInbox
|
|||
{
|
||||
message.AffectedWorkflowInstancesIds = triggeredWorkflows.Select(x => x.WorkflowInstanceId).ToList();
|
||||
message.HandledAt = _systemClock.UtcNow;
|
||||
await _store.SaveAsync(message, cancellationToken);
|
||||
await _messageStore.SaveAsync(message, cancellationToken);
|
||||
}
|
||||
|
||||
return new DeliverWorkflowInboxMessageResult(triggeredWorkflows);
|
||||
|
|
@ -117,12 +117,12 @@ public class DefaultWorkflowInbox : IWorkflowInbox
|
|||
/// <inheritdoc />
|
||||
public async ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return await _store.FindManyAsync(filter, cancellationToken);
|
||||
return await _messageStore.FindManyAsync(filter, cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(IEnumerable<WorkflowInboxMessageFilter> filters, CancellationToken cancellationToken = default)
|
||||
{
|
||||
return await _store.FindManyAsync(filters, cancellationToken);
|
||||
return await _messageStore.FindManyAsync(filters, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,69 @@
|
|||
using Elsa.Common.Contracts;
|
||||
using Elsa.Common.Models;
|
||||
using Elsa.Common.Services;
|
||||
using Elsa.Workflows.Runtime.Contracts;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
using Elsa.Workflows.Runtime.Filters;
|
||||
|
||||
namespace Elsa.Workflows.Runtime.Stores;
|
||||
|
||||
/// <summary>
|
||||
/// An in-memory implementation of <see cref="IWorkflowInboxMessageStore"/>.
|
||||
/// </summary>
|
||||
public class MemoryWorkflowInboxMessageStore : IWorkflowInboxMessageStore
|
||||
{
|
||||
private readonly MemoryStore<WorkflowInboxMessage> _store;
|
||||
private readonly ISystemClock _systemClock;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="MemoryWorkflowInboxMessageStore"/> class.
|
||||
/// </summary>
|
||||
public MemoryWorkflowInboxMessageStore(MemoryStore<WorkflowInboxMessage> store, ISystemClock systemClock)
|
||||
{
|
||||
_store = store;
|
||||
_systemClock = systemClock;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask SaveAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_store.Save(message, x => x.Id);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var entities = _store.Query(query => Filter(query, filter)).ToList();
|
||||
return new(entities);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(IEnumerable<WorkflowInboxMessageFilter> filters, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var entities = _store.Query(query => Filter(query, filters.ToArray())).ToList();
|
||||
return new(entities);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<long> DeleteManyAsync(WorkflowInboxMessageFilter filter, PageArgs? pageArgs = default, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var entities = _store.Query(query => Paginate(Filter(query, filter), pageArgs)).ToList();
|
||||
var ids = entities.Select(x => x.Id);
|
||||
var deleteCount = _store.DeleteMany(ids);
|
||||
return new (deleteCount);
|
||||
}
|
||||
|
||||
private IQueryable<WorkflowInboxMessage> Filter(IQueryable<WorkflowInboxMessage> query, params WorkflowInboxMessageFilter[] filters)
|
||||
{
|
||||
foreach (var filter in filters) filter.Apply(query, _systemClock.UtcNow);
|
||||
return query;
|
||||
}
|
||||
|
||||
private static IQueryable<WorkflowInboxMessage> Paginate(IQueryable<WorkflowInboxMessage> queryable, PageArgs? pageArgs)
|
||||
{
|
||||
if (pageArgs?.Offset != null) queryable = queryable.Skip(pageArgs.Offset.Value);
|
||||
if (pageArgs?.Limit != null) queryable = queryable.Take(pageArgs.Limit.Value);
|
||||
return queryable;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,56 +0,0 @@
|
|||
using Elsa.Common.Services;
|
||||
using Elsa.Workflows.Runtime.Contracts;
|
||||
using Elsa.Workflows.Runtime.Entities;
|
||||
using Elsa.Workflows.Runtime.Filters;
|
||||
|
||||
namespace Elsa.Workflows.Runtime.Stores;
|
||||
|
||||
/// <summary>
|
||||
/// An in-memory implementation of <see cref="IWorkflowInboxStore"/>.
|
||||
/// </summary>
|
||||
public class MemoryWorkflowInboxStore : IWorkflowInboxStore
|
||||
{
|
||||
private readonly MemoryStore<WorkflowInboxMessage> _store;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="MemoryWorkflowInboxStore"/> class.
|
||||
/// </summary>
|
||||
public MemoryWorkflowInboxStore(MemoryStore<WorkflowInboxMessage> store)
|
||||
{
|
||||
_store = store;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask SaveAsync(WorkflowInboxMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_store.Save(message, x => x.Id);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var entities = _store.Query(query => Filter(query, filter)).ToList();
|
||||
return new(entities);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<IEnumerable<WorkflowInboxMessage>> FindManyAsync(IEnumerable<WorkflowInboxMessageFilter> filters, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var entities = _store.Query(query => Filter(query, filters.ToArray())).ToList();
|
||||
return new(entities);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask<long> DeleteAsync(WorkflowInboxMessageFilter filter, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var ids = (await FindManyAsync(filter, cancellationToken)).Select(x => x.Id);
|
||||
return _store.DeleteMany(ids);
|
||||
}
|
||||
|
||||
private static IQueryable<WorkflowInboxMessage> Filter(IQueryable<WorkflowInboxMessage> query, params WorkflowInboxMessageFilter[] filters)
|
||||
{
|
||||
foreach (var filter in filters) filter.Apply(query);
|
||||
return query;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue