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

Implemented Elasticsearch for workflow instance and log record persistence

Browse source
This commit is contained in:
Gürkan Güran 2023-01-05 15:10:55 +01:00
parent 30031510d1
commit fcfd092003
36 changed files with 671 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
Elsa.sln
View file

@ -144,6 +144,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.EntityFrameworkCore.Po
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.RunTaskIntegration", "src\samples\aspnet\Elsa.Samples.RunTaskIntegration\Elsa.Samples.RunTaskIntegration.csproj", "{51050209-EC2F-4DC7-8F46-07E22B7811CD}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Elasticsearch", "src\modules\Elsa.Elasticsearch\Elsa.Elasticsearch.csproj", "{3246883E-2FA7-4B4A-BDC5-99039A2869BC}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -362,6 +364,10 @@ Global
{51050209-EC2F-4DC7-8F46-07E22B7811CD}.Debug|Any CPU.Build.0 = Debug|Any CPU
{51050209-EC2F-4DC7-8F46-07E22B7811CD}.Release|Any CPU.ActiveCfg = Release|Any CPU
{51050209-EC2F-4DC7-8F46-07E22B7811CD}.Release|Any CPU.Build.0 = Release|Any CPU
{3246883E-2FA7-4B4A-BDC5-99039A2869BC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{3246883E-2FA7-4B4A-BDC5-99039A2869BC}.Debug|Any CPU.Build.0 = Debug|Any CPU
{3246883E-2FA7-4B4A-BDC5-99039A2869BC}.Release|Any CPU.ActiveCfg = Release|Any CPU
{3246883E-2FA7-4B4A-BDC5-99039A2869BC}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(NestedProjects) = preSolution
{155227F0-A33B-40AA-A4B4-06F813EB921B} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F}
@ -424,5 +430,6 @@ Global
{536EFB55-EEA1-4761-9A72-D4EA84A6DB70} = {56C2FFB8-EA54-45B5-A095-4A78142EB4B5}
{AA84DBF7-F70F-4673-8DC2-6EFBE3E9BF83} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79}
{51050209-EC2F-4DC7-8F46-07E22B7811CD} = {56C2FFB8-EA54-45B5-A095-4A78142EB4B5}
{3246883E-2FA7-4B4A-BDC5-99039A2869BC} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79}
EndGlobalSection
EndGlobal

6
src/bundles/Elsa.AllInOne.Web/Program.cs
View file

@ -30,7 +30,11 @@ services
identity.TokenOptions = identityTokenOptions;
})
.UseDefaultAuthentication()
.UseWorkflowRuntime(runtime => { runtime.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)); })
.UseWorkflowRuntime(runtime =>
{
runtime.UseDefaultRuntime(dr => dr.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)));
runtime.UseExecutionLogRecords(e => e.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)));
})
.UseLabels(labels => labels.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)))
.UseActivityDefinitions(feature => feature.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)))
.UseJobs()

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

@ -9,6 +9,8 @@
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\modules\Elsa.Elasticsearch\Elsa.Elasticsearch.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.PostgreSql\Elsa.EntityFrameworkCore.PostgreSql.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Jobs.Activities\Elsa.Jobs.Activities.csproj" />
<ProjectReference Include="..\..\modules\Elsa.ActivityDefinitions\Elsa.ActivityDefinitions.csproj" />
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.Sqlite\Elsa.EntityFrameworkCore.Sqlite.csproj" />

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

@ -1,3 +1,6 @@
using Elsa.Elasticsearch.Modules.Management;
using Elsa.Elasticsearch.Modules.Runtime;
using Elsa.Elasticsearch.Options;
using Elsa.EntityFrameworkCore.Extensions;
using Elsa.Extensions;
using Elsa.Identity;
@ -24,6 +27,9 @@ var identitySection = configuration.GetSection("Identity");
var identityTokenSection = identitySection.GetSection("Tokens");
identitySection.Bind(identityOptions);
identityTokenSection.Bind(identityTokenOptions);
var postgreSqlConnectionString = configuration.GetConnectionString("PostgreSql")!;
var elasticOptions = new ElasticsearchOptions();
configuration.GetSection(ElasticsearchOptions.Elasticsearch).Bind(elasticOptions);
// Add Elsa services.
services
@ -35,11 +41,16 @@ services
identity.TokenOptions = identityTokenOptions;
})
.UseDefaultAuthentication()
.UseWorkflowManagement(management => management.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)))
.UseWorkflowManagement(management =>
{
management.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString));
management.UseWorkflowInstances(w => w.UseElasticsearch(elasticOptions));
})
.UseWorkflowRuntime(runtime =>
{
runtime.UseProtoActor(proto => proto.PersistenceProvider = _ => new SqliteProvider(new SqliteConnectionStringBuilder(sqliteConnectionString)));
runtime.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString));
runtime.UseDefaultRuntime(df => df.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)));
runtime.UseExecutionLogRecords(e => e.UseElasticsearch(elasticOptions));
runtime.UseAsyncWorkflowStateExporter();
runtime.UseMassTransitDispatcher();
})

2
src/bundles/Elsa.WorkflowServer.Web/Properties/launchSettings.json
View file

@ -12,7 +12,7 @@
"commandName": "Project",
"dotnetRunMessages": true,
"launchBrowser": true,
"applicationUrl": "https://localhost:7072;http://localhost:5072",
"applicationUrl": "https://localhost:5001;http://localhost:5072",
"environmentVariables": {
"ASPNETCORE_ENVIRONMENT": "Development"
}

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

@ -9,6 +9,7 @@
"AllowedHosts": "*",
"ConnectionStrings": {
"Sqlite": "Data Source=elsa.sqlite.db;Cache=Shared;",
"PostgreSql": "Server=localhost;Port=5455;Database=myPostgresDb;Username=guest;Password=guest",
"RabbitMq": "rabbitmq://guest:guest@localhost"
},
"Identity": {
@ -17,5 +18,8 @@
"SigningKey": "secret-signing-key",
"Lifetime": "8:00:00"
}
},
"Elasticsearch": {
"Endpoint": "http://localhost:9200/"
}
}
}

35
src/modules/Elsa.Elasticsearch/Common/Configuration.cs Normal file
View file

@ -0,0 +1,35 @@
using Elasticsearch.Net;
using Elsa.Elasticsearch.Options;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Runtime.Entities;
using Nest;
namespace Elsa.Elasticsearch.Common;
public static class Configuration
{
public static ConnectionSettings SetupAuthentication(this ConnectionSettings settings, ElasticsearchOptions options)
{
if (!string.IsNullOrEmpty(options.ApiKey))
{
settings.ApiKeyAuthentication(new ApiKeyAuthenticationCredentials(options.ApiKey));
}
else if (!string.IsNullOrEmpty(options.Username) && !string.IsNullOrEmpty(options.Password))
{
settings.BasicAuthentication(options.Username, options.Password);
}
return settings;
}
public static ConnectionSettings SetupMappingsAndIndices(this ConnectionSettings settings)
{
settings.DefaultMappingFor<WorkflowInstance>(s =>
s.IndexName(ElasticsearchOptions.Indices[typeof(WorkflowInstance)]));
settings.DefaultMappingFor<WorkflowExecutionLogRecord>(s =>
s.IndexName(ElasticsearchOptions.Indices[typeof(WorkflowExecutionLogRecord)]));
return settings;
}
}

39
src/modules/Elsa.Elasticsearch/Common/ElasticFeatureBase.cs Normal file
View file

@ -0,0 +1,39 @@
using Elasticsearch.Net;
using Elsa.Elasticsearch.Options;
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
using Microsoft.Extensions.DependencyInjection;
using Nest;
namespace Elsa.Elasticsearch.Common;
public abstract class ElasticFeatureBase : FeatureBase
{
public ElasticFeatureBase(IModule module) : base(module)
{
}
internal ElasticsearchOptions Options { get; set; } = new();
public override void Apply()
{
if (Services.All(x => x.ServiceType != typeof(ElasticClient)))
{
Services.AddSingleton(new ElasticClient(GetSettings()));
}
}
private ConnectionSettings GetSettings()
{
return new ConnectionSettings(new Uri(Options.Endpoint))
.SetupAuthentication(Options)
.SetupMappingsAndIndices();
}
protected void AddStore<TModel, TStore>() where TModel : class where TStore : class
{
Services
.AddSingleton<ElasticStore<TModel>>()
.AddSingleton<TStore>();
}
}

121
src/modules/Elsa.Elasticsearch/Common/ElasticStore.cs Normal file
View file

@ -0,0 +1,121 @@
using System.Collections.ObjectModel;
using Elsa.Common.Models;
using Microsoft.Extensions.Logging;
using Nest;
namespace Elsa.Elasticsearch.Common;
public class ElasticStore<T> where T : class
{
private readonly ElasticClient _elasticClient;
private readonly ILogger _logger;
public ElasticStore(ElasticClient elasticClient, ILogger<ElasticStore<T>> logger)
{
_elasticClient = elasticClient;
_logger = logger;
}
public async Task<T?> GetByIdAsync(string id, CancellationToken cancellationToken)
{
var response = await _elasticClient.GetAsync(DocumentPath<T>.Id(id), ct: cancellationToken);
if (response.IsValid) return response.Source;
_logger.LogError("Failed to fetch data from Elasticsearch: {message}", response.ServerError?.ToString());
return null;
}
public async Task<Page<T>> SearchAsync(Func<QueryContainerDescriptor<T>, QueryContainer> query, PageArgs? pageArgs, CancellationToken cancellationToken)
{
var search = new SearchDescriptor<T>().Query(query);
if (pageArgs != default)
{
search = search.From(pageArgs.Offset).Size(pageArgs.Limit);
}
var response = await _elasticClient.SearchAsync<T>(search, cancellationToken);
if (response.IsValid) return new Page<T>(response.Hits.Select(hit => hit.Source).ToList(), response.Total);
_logger.LogError("Failed to search data in Elasticsearch: {message}", response.ServerError?.ToString());
return new Page<T>(new Collection<T>(), 0);
}
public async Task<Page<T>> SearchAsync(SearchDescriptor<T> search, PageArgs? pageArgs, CancellationToken cancellationToken)
{
if (pageArgs != default)
{
search = search.From(pageArgs.Offset).Size(pageArgs.Limit);
}
var response = await _elasticClient.SearchAsync<T>(search, cancellationToken);
if (response.IsValid) return new Page<T>(response.Hits.Select(hit => hit.Source).ToList(), response.Total);
_logger.LogError("Failed to search data in Elasticsearch: {message}", response.ServerError?.ToString());
return new Page<T>(new Collection<T>(), 0);
}
public async Task<bool> SaveAsync(T model, CancellationToken cancellationToken)
{
var response = await _elasticClient.IndexAsync(model, descriptor => descriptor, cancellationToken);
if (response.IsValid) return true;
_logger.LogError("Failed to save data in Elasticsearch: {message}", response.ServerError?.ToString());
return false;
}
public async Task<bool> SaveManyAsync(IEnumerable<T> documents, CancellationToken cancellationToken)
{
var response = await _elasticClient.IndexManyAsync(documents, cancellationToken: cancellationToken);
if (response.IsValid) return true;
_logger.LogError("Failed to save data in Elasticsearch: {message}", response.ServerError?.ToString());
return false;
}
public async Task<bool> DeleteByIdAsync(string id, CancellationToken cancellationToken)
{
var response = await _elasticClient.DeleteAsync(DocumentPath<T>.Id(id), ct: cancellationToken);
if (response.IsValid) return true;
_logger.LogError("Failed to delete data in Elasticsearch: {message}", response.ServerError?.ToString());
return false;
}
public async Task<int> DeleteManyAsync(IEnumerable<T> list, CancellationToken cancellationToken)
{
var response = await _elasticClient.DeleteManyAsync(list, cancellationToken: cancellationToken);
if (response.IsValid) return 0;
_logger.LogError("Failed to delete data in Elasticsearch: {message}", response.ServerError?.ToString());
return response.Items.Count;
}
public async Task<bool> DeleteByQueryAsync(Func<QueryContainerDescriptor<T>, QueryContainer> query, CancellationToken cancellationToken)
{
var response = await _elasticClient.DeleteByQueryAsync<T>(q => q
.Query(query), cancellationToken);
if (response.IsValid) return true;
_logger.LogError("Failed to delete data in Elasticsearch: {message}", response.ServerError?.ToString());
return false;
}
public async Task<long> CountAsync(Func<QueryContainerDescriptor<T>, QueryContainer> query, CancellationToken cancellationToken)
{
var response = await _elasticClient.CountAsync<T>(s =>
s.Query(query), cancellationToken);
if (response.IsValid) return response.Count;
_logger.LogError("Failed to count data in Elasticsearch: {message}", response.ServerError?.ToString());
return 0;
}
}

23
src/modules/Elsa.Elasticsearch/Elsa.Elasticsearch.csproj Normal file
View file

@ -0,0 +1,23 @@
<Project Sdk="Microsoft.NET.Sdk">
<Import Project="..\..\..\common.props" />
<Import Project="..\..\..\configureawait.props" />
<PropertyGroup>
<TargetFrameworks>net6.0;net7.0</TargetFrameworks>
<Description>
Provides Elasticsearch implementation for workflow state persistence.
</Description>
<PackageTags>elsa module</PackageTags>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\common\Elsa.Features\Elsa.Features.csproj" />
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="NEST" Version="7.17.5" />
</ItemGroup>
</Project>

3
src/modules/Elsa.Elasticsearch/FodyWeavers.xml Normal file
View file

@ -0,0 +1,3 @@
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd">
<ConfigureAwait />
</Weavers>

31
src/modules/Elsa.Elasticsearch/Modules/Management/ElasticWorkflowInstanceFeature.cs Normal file
View file

@ -0,0 +1,31 @@
using Elsa.Elasticsearch.Common;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Features;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Elasticsearch.Modules.Management;
[DependsOn(typeof(WorkflowManagementFeature))]
public class ElasticWorkflowInstanceFeature : ElasticFeatureBase
{
public ElasticWorkflowInstanceFeature(IModule module) : base(module)
{
}
public override void Configure()
{
Module.Configure<WorkflowManagementFeature>(feature =>
{
feature.WorkflowInstanceStore = sp => sp.GetRequiredService<ElasticWorkflowInstanceStore>();
});
}
public override void Apply()
{
base.Apply();
AddStore<WorkflowInstance, ElasticWorkflowInstanceStore>();
}
}

17
src/modules/Elsa.Elasticsearch/Modules/Management/Extensions.cs Normal file
View file

@ -0,0 +1,17 @@
using Elsa.Elasticsearch.Options;
using Elsa.Workflows.Management.Features;
namespace Elsa.Elasticsearch.Modules.Management;
public static class Extensions
{
/// <summary>
/// Configures the <see cref="WorkflowInstanceFeature"/> to use the <see cref="ElasticWorkflowInstanceFeature"/>.
/// </summary>
public static WorkflowInstanceFeature UseElasticsearch(this WorkflowInstanceFeature feature, ElasticsearchOptions options, Action<ElasticWorkflowInstanceFeature>? configure = default)
{
configure += f => f.Options = options;
feature.Module.Configure(configure);
return feature;
}
}

99
src/modules/Elsa.Elasticsearch/Modules/Management/WorkflowInstanceStore.cs Normal file
View file

@ -0,0 +1,99 @@
using Elsa.Common.Entities;
using Elsa.Common.Models;
using Elsa.Elasticsearch.Common;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Models;
using Elsa.Workflows.Management.Services;
using Nest;
namespace Elsa.Elasticsearch.Modules.Management;
public class ElasticWorkflowInstanceStore : IWorkflowInstanceStore
{
private readonly ElasticStore<WorkflowInstance> _store;
public ElasticWorkflowInstanceStore(ElasticStore<WorkflowInstance> store)
{
_store = store;
}
public async Task<WorkflowInstance?> FindByIdAsync(string id, CancellationToken cancellationToken = default) =>
await _store.GetByIdAsync(id, cancellationToken);
public async Task SaveAsync(WorkflowInstance record, CancellationToken cancellationToken = default) =>
await _store.SaveAsync(record, cancellationToken);
public async Task SaveManyAsync(IEnumerable<WorkflowInstance> records, CancellationToken cancellationToken = default) =>
await _store.SaveManyAsync(records, cancellationToken);
public async Task<bool> DeleteAsync(string id, CancellationToken cancellationToken = default) =>
await _store.DeleteByIdAsync(id, cancellationToken);
public async Task<int> DeleteManyAsync(IEnumerable<string> ids, CancellationToken cancellationToken = default) =>
await _store.DeleteManyAsync(ids.Select(id => new WorkflowInstance { Id = id}), cancellationToken);
public async Task DeleteManyByDefinitionIdAsync(string definitionId, CancellationToken cancellationToken = default) =>
await _store.DeleteByQueryAsync(q => q
.Term(t => t
.Field(f => f.DefinitionId)
.Value(definitionId)),
cancellationToken);
public async Task<Page<WorkflowInstanceSummary>> FindManyAsync(FindWorkflowInstancesArgs args,
CancellationToken cancellationToken = default)
{
var queryDescriptor = new QueryContainerDescriptor<WorkflowInstance>();
var searchDescriptor = new SearchDescriptor<WorkflowInstance>();
var query = new QueryContainer();
var (searchTerm, definitionId, version, correlationId, workflowStatus, workflowSubStatus, pageArgs, orderBy,
orderDirection) = args;
if (!string.IsNullOrWhiteSpace(definitionId))
query = queryDescriptor.Match(m => m.Field(f => f.DefinitionId).Query(args.DefinitionId));
if (version != null)
query &= queryDescriptor.Match(m => m.Field(f => f.Version).Query(args.Version.ToString()));
if (!string.IsNullOrWhiteSpace(correlationId))
query &= queryDescriptor.Match(m => m.Field(f => f.CorrelationId).Query(args.CorrelationId));
if (workflowStatus != null)
query &= queryDescriptor.Match(m => m.Field(f => f.Status).Query(args.WorkflowStatus?.ToString()));
if (workflowSubStatus != null)
query &= queryDescriptor.Match(m => m.Field(f => f.SubStatus).Query(args.WorkflowSubStatus?.ToString()));
if(!string.IsNullOrWhiteSpace(searchTerm))
{
query &= queryDescriptor
.MultiMatch(c => c
.Type(TextQueryType.Phrase)
.Fields(f => f
.Field(p => p.Name)
.Field(p => p.Id)
.Field(p => p.DefinitionId)
.Field(p => p.CorrelationId))
.Lenient()
.Query(searchTerm));
}
searchDescriptor = orderBy switch
{
OrderBy.Finished => orderDirection == OrderDirection.Ascending
? searchDescriptor.Sort(x => x.Ascending(f => f.FinishedAt))
: searchDescriptor.Sort(x => x.Descending(f => f.FinishedAt)),
OrderBy.LastExecuted => orderDirection == OrderDirection.Ascending
? searchDescriptor.Sort(x => x.Ascending(f => f.LastExecutedAt))
: searchDescriptor.Sort(x => x.Descending(f => f.LastExecutedAt)),
OrderBy.Created => orderDirection == OrderDirection.Ascending
? searchDescriptor.Sort(x => x.Ascending(f => f.CreatedAt))
: searchDescriptor.Sort(x => x.Descending(f => f.CreatedAt)),
_ => searchDescriptor
};
var result = await _store.SearchAsync(searchDescriptor.Query(_ => query), args.PageArgs, cancellationToken);
return new Page<WorkflowInstanceSummary>(result.Items.Select(WorkflowInstanceSummary.FromInstance).ToList(),
result.TotalCount);
}
}

31
src/modules/Elsa.Elasticsearch/Modules/Runtime/ElasticExecutionLogRecordFeature.cs Normal file
View file

@ -0,0 +1,31 @@
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Features;
using Elsa.Elasticsearch.Common;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Elasticsearch.Modules.Runtime;
[DependsOn(typeof(WorkflowRuntimeFeature))]
public class ElasticExecutionLogRecordFeature : ElasticFeatureBase
{
public ElasticExecutionLogRecordFeature(IModule module) : base(module)
{
}
public override void Configure()
{
Module.Configure<WorkflowRuntimeFeature>(feature =>
{
feature.WorkflowExecutionLogStore = sp => sp.GetRequiredService<ElasticWorkflowExecutionLogStore>();
});
}
public override void Apply()
{
base.Apply();
AddStore<WorkflowExecutionLogRecord, ElasticWorkflowExecutionLogStore>();
}
}

17
src/modules/Elsa.Elasticsearch/Modules/Runtime/Extensions.cs Normal file
View file

@ -0,0 +1,17 @@
using Elsa.Elasticsearch.Options;
using Elsa.Workflows.Runtime.Features;
namespace Elsa.Elasticsearch.Modules.Runtime;
public static class Extensions
{
/// <summary>
/// Configures the <see cref="ExecutionLogRecordFeature"/> to use the <see cref="ElasticExecutionLogRecordFeature"/>.
/// </summary>
public static ExecutionLogRecordFeature UseElasticsearch(this ExecutionLogRecordFeature feature, ElasticsearchOptions options, Action<ElasticExecutionLogRecordFeature>? configure = default)
{
configure += f => f.Options = options;
feature.Module.Configure(configure);
return feature;
}
}

31
src/modules/Elsa.Elasticsearch/Modules/Runtime/WorkflowExecutionLogStore.cs Normal file
View file

@ -0,0 +1,31 @@
using Elsa.Common.Models;
using Elsa.Elasticsearch.Common;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Services;
namespace Elsa.Elasticsearch.Modules.Runtime;
public class ElasticWorkflowExecutionLogStore : IWorkflowExecutionLogStore
{
private readonly ElasticStore<WorkflowExecutionLogRecord> _store;
public ElasticWorkflowExecutionLogStore(ElasticStore<WorkflowExecutionLogRecord> store)
{
_store = store;
}
public async Task SaveAsync(WorkflowExecutionLogRecord record, CancellationToken cancellationToken = default) =>
await _store.SaveAsync(record, cancellationToken);
public async Task SaveManyAsync(IEnumerable<WorkflowExecutionLogRecord> records, CancellationToken cancellationToken = default) =>
await _store.SaveManyAsync(records, cancellationToken);
public async Task<Page<WorkflowExecutionLogRecord>> FindManyByWorkflowInstanceIdAsync(string workflowInstanceId, PageArgs? pageArgs = default, CancellationToken cancellationToken = default) =>
await _store
.SearchAsync(q => q
.Match(m => m
.Field(f => f.WorkflowInstanceId)
.Query(workflowInstanceId)),
pageArgs,
cancellationToken);
}

20
src/modules/Elsa.Elasticsearch/Options/ElasticsearchOptions.cs Normal file
View file

@ -0,0 +1,20 @@
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Runtime.Entities;
namespace Elsa.Elasticsearch.Options;
public class ElasticsearchOptions
{
public const string Elasticsearch = "Elasticsearch";
public static readonly Dictionary<Type, string> Indices = new()
{
{typeof(WorkflowInstance),"workflow-instances"},
{typeof(WorkflowExecutionLogRecord),"execution-log-records"}
};
public string Endpoint { get; set; }
public string Username { get; set; }
public string Password { get; set; }
public string ApiKey { get; set; }
}

7
src/modules/Elsa.EntityFrameworkCore.PostgreSql/Modules/Management/Extensions.cs
View file

@ -1,4 +1,5 @@
using Elsa.EntityFrameworkCore.Modules.Management;
using Elsa.EntityFrameworkCore.Modules.Runtime;
// ReSharper disable once CheckNamespace
namespace Elsa.EntityFrameworkCore.Extensions;
@ -10,4 +11,10 @@ public static partial class Extensions
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaPostgreSql(connectionString);
return feature;
}
public static EFCoreWorkflowInstancePersistenceFeature UsePostgreSql(this EFCoreWorkflowInstancePersistenceFeature feature, string connectionString)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaPostgreSql(connectionString);
return feature;
}
}

8
src/modules/Elsa.EntityFrameworkCore.PostgreSql/Modules/Runtime/Extensions.cs
View file

@ -5,7 +5,13 @@ namespace Elsa.EntityFrameworkCore.Extensions;
public static partial class Extensions
{
public static EFCoreRuntimePersistenceFeature UsePostgreSql(this EFCoreRuntimePersistenceFeature feature, string connectionString)
public static EFCoreDefaultRuntimePersistenceFeature UsePostgreSql(this EFCoreDefaultRuntimePersistenceFeature feature, string connectionString)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaPostgreSql(connectionString);
return feature;
}
public static EFCoreExecutionLogRecordPersistenceFeature UsePostgreSql(this EFCoreExecutionLogRecordPersistenceFeature feature, string connectionString)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaPostgreSql(connectionString);
return feature;

2
src/modules/Elsa.EntityFrameworkCore.SqlServer/Modules/Runtime/Extensions.cs
View file

@ -5,7 +5,7 @@ namespace Elsa.EntityFrameworkCore.Extensions;
public static partial class Extensions
{
public static EFCoreRuntimePersistenceFeature UseSqlServer(this EFCoreRuntimePersistenceFeature feature, string connectionString)
public static EFCoreDefaultRuntimePersistenceFeature UseSqlServer(this EFCoreDefaultRuntimePersistenceFeature feature, string connectionString)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlServer(connectionString);
return feature;

6
src/modules/Elsa.EntityFrameworkCore.Sqlite/Modules/Management/Extensions.cs
View file

@ -11,4 +11,10 @@ public static partial class Extensions
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlite(connectionString);
return feature;
}
public static EFCoreWorkflowInstancePersistenceFeature UseSqlite(this EFCoreWorkflowInstancePersistenceFeature feature, string connectionString = Constants.DefaultConnectionString)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlite(connectionString);
return feature;
}
}

8
src/modules/Elsa.EntityFrameworkCore.Sqlite/Modules/Runtime/Extensions.cs
View file

@ -6,7 +6,13 @@ namespace Elsa.EntityFrameworkCore.Extensions;
public static partial class Extensions
{
public static EFCoreRuntimePersistenceFeature UseSqlite(this EFCoreRuntimePersistenceFeature feature, string connectionString = Constants.DefaultConnectionString)
public static EFCoreDefaultRuntimePersistenceFeature UseSqlite(this EFCoreDefaultRuntimePersistenceFeature feature, string connectionString = Constants.DefaultConnectionString)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlite(connectionString);
return feature;
}
public static EFCoreExecutionLogRecordPersistenceFeature UseSqlite(this EFCoreExecutionLogRecordPersistenceFeature feature, string connectionString = Constants.DefaultConnectionString)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlite(connectionString);
return feature;

6
src/modules/Elsa.EntityFrameworkCore/Common/Store.cs
View file

@ -16,12 +16,6 @@ public class Store<TDbContext, TEntity> where TDbContext : DbContext where TEnti
_dbContextFactory = dbContextFactory;
}
public async Task<TDbContext> GetDbContextAsync(CancellationToken cancellationToken)
{
await using var dbContext = await CreateDbContextAsync(cancellationToken);
return dbContext;
}
public async Task<TDbContext> CreateDbContextAsync(CancellationToken cancellationToken = default) => await _dbContextFactory.CreateDbContextAsync(cancellationToken);
public async Task SaveAsync(TEntity entity, CancellationToken cancellationToken = default) => await SaveAsync(entity, default, cancellationToken);

12
src/modules/Elsa.EntityFrameworkCore/Modules/Management/Extensions.cs
View file

@ -10,7 +10,17 @@ public static class WorkflowManagementFeatureExtensions
/// <summary>
/// Sets up the EF Core persistence provider.
/// </summary>
public static WorkflowManagementFeature UseEntityFrameworkCore(this WorkflowManagementFeature feature, Action<EFCoreManagementPersistenceFeature>? configure = default)
public static WorkflowManagementFeature UseEntityFrameworkCore(this WorkflowManagementFeature feature,
Action<EFCoreManagementPersistenceFeature>? configure = default)
{
feature.Module.Configure(configure);
return feature;
}
/// <summary>
/// Sets up the EF Core persistence provider.
/// </summary>
public static WorkflowInstanceFeature UseEntityFrameworkCore(this WorkflowInstanceFeature feature, Action<EFCoreWorkflowInstancePersistenceFeature>? configure = default)
{
feature.Module.Configure(configure);
return feature;

2
src/modules/Elsa.EntityFrameworkCore/Modules/Management/Feature.cs → src/modules/Elsa.EntityFrameworkCore/Modules/Management/ManagementPersistenceFeature.cs
View file

@ -19,7 +19,6 @@ public class EFCoreManagementPersistenceFeature : PersistenceFeatureBase<Managem
Module.Configure<WorkflowManagementFeature>(feature =>
{
feature.WorkflowDefinitionStore = sp => sp.GetRequiredService<EFCoreWorkflowDefinitionStore>();
feature.WorkflowInstanceStore = sp => sp.GetRequiredService<EFCoreWorkflowInstanceStore>();
});
}
@ -28,6 +27,5 @@ public class EFCoreManagementPersistenceFeature : PersistenceFeatureBase<Managem
base.Apply();
AddStore<WorkflowDefinition, EFCoreWorkflowDefinitionStore>();
AddStore<WorkflowInstance, EFCoreWorkflowInstanceStore>();
}
}

31
src/modules/Elsa.EntityFrameworkCore/Modules/Management/WorkflowInstancePersistenceFeature.cs Normal file
View file

@ -0,0 +1,31 @@
using Elsa.EntityFrameworkCore.Common;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Workflows.Management.Entities;
using Elsa.Workflows.Management.Features;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.EntityFrameworkCore.Modules.Management;
[DependsOn(typeof(WorkflowManagementFeature))]
public class EFCoreWorkflowInstancePersistenceFeature : PersistenceFeatureBase<ManagementElsaDbContext>
{
public EFCoreWorkflowInstancePersistenceFeature(IModule module) : base(module)
{
}
public override void Configure()
{
Module.Configure<WorkflowManagementFeature>(feature =>
{
feature.WorkflowInstanceStore = sp => sp.GetRequiredService<EFCoreWorkflowInstanceStore>();
});
}
public override void Apply()
{
base.Apply();
AddStore<WorkflowInstance, EFCoreWorkflowInstanceStore>();
}
}

4
src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/RuntimePersistenceFeature.cs → src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/DefaultRuntimePersistenceFeature.cs
View file

@ -10,9 +10,9 @@ using Microsoft.Extensions.DependencyInjection;
namespace Elsa.EntityFrameworkCore.Modules.Runtime;
[DependsOn(typeof(WorkflowRuntimeFeature))]
public class EFCoreRuntimePersistenceFeature : PersistenceFeatureBase<RuntimeElsaDbContext>
public class EFCoreDefaultRuntimePersistenceFeature : PersistenceFeatureBase<RuntimeElsaDbContext>
{
public EFCoreRuntimePersistenceFeature(IModule module) : base(module)
public EFCoreDefaultRuntimePersistenceFeature(IModule module) : base(module)
{
}

32
src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/ExecutionLogRecordPersistenceFeature.cs Normal file
View file

@ -0,0 +1,32 @@
using Elsa.EntityFrameworkCore.Common;
using Elsa.Features.Abstractions;
using Elsa.Features.Attributes;
using Elsa.Features.Services;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Features;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.EntityFrameworkCore.Modules.Runtime;
[DependsOn(typeof(WorkflowRuntimeFeature))]
public class EFCoreExecutionLogRecordPersistenceFeature : PersistenceFeatureBase<RuntimeElsaDbContext>
{
public EFCoreExecutionLogRecordPersistenceFeature(IModule module) : base(module)
{
}
public override void Configure()
{
Module.Configure<WorkflowRuntimeFeature>(feature =>
{
feature.WorkflowExecutionLogStore = sp => sp.GetRequiredService<EFCoreWorkflowExecutionLogStore>();
});
}
public override void Apply()
{
base.Apply();
AddStore<WorkflowExecutionLogRecord, EFCoreWorkflowExecutionLogStore>();
}
}

13
src/modules/Elsa.EntityFrameworkCore/Modules/Runtime/Extensions.cs
View file

@ -10,9 +10,18 @@ namespace Elsa.EntityFrameworkCore.Modules.Runtime;
public static class Extensions
{
/// <summary>
/// Configures the <see cref="WorkflowRuntimeFeature"/> to use the <see cref="EFCoreRuntimePersistenceFeature"/>.
/// Configures the <see cref="DefaultRuntimeFeature"/> to use the <see cref="EFCoreDefaultRuntimePersistenceFeature"/>.
/// </summary>
public static WorkflowRuntimeFeature UseEntityFrameworkCore(this WorkflowRuntimeFeature feature, Action<EFCoreRuntimePersistenceFeature>? configure = default)
public static DefaultRuntimeFeature UseEntityFrameworkCore(this DefaultRuntimeFeature feature, Action<EFCoreDefaultRuntimePersistenceFeature>? configure = default)
{
feature.Module.Configure(configure);
return feature;
}
/// <summary>
/// Configures the <see cref="ExecutionLogRecordFeature"/> to use the <see cref="EFCoreWorkflowExecutionLogStore"/>.
/// </summary>
public static ExecutionLogRecordFeature UseEntityFrameworkCore(this ExecutionLogRecordFeature feature, Action<EFCoreExecutionLogRecordPersistenceFeature>? configure = default)
{
feature.Module.Configure(configure);
return feature;

9
src/modules/Elsa.Workflows.Management/Extensions/ModuleExtensions.cs
View file

@ -23,6 +23,15 @@ public static class ModuleExtensions
});
return module;
}
/// <summary>
/// Adds the workflow instance feature to workflow management module.
/// </summary>
public static WorkflowManagementFeature UseWorkflowInstances(this WorkflowManagementFeature feature, Action<WorkflowInstanceFeature>? configure = default)
{
feature.Module.Configure(configure);
return feature;
}
/// <summary>
/// Adds all types implementing <see cref="IActivity"/> to the system.

11
src/modules/Elsa.Workflows.Management/Features/WorkflowInstanceFeature.cs Normal file
View file

@ -0,0 +1,11 @@
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
namespace Elsa.Workflows.Management.Features;
public class WorkflowInstanceFeature : FeatureBase
{
public WorkflowInstanceFeature(IModule module) : base(module)
{
}
}

12
src/modules/Elsa.Workflows.Runtime/Extensions/ModuleExtensions.cs
View file

@ -27,4 +27,16 @@ public static class ModuleExtensions
module.Configure<WorkflowRuntimeFeature>().AddWorkflow<T>();
return module;
}
public static WorkflowRuntimeFeature UseDefaultRuntime(this WorkflowRuntimeFeature feature, Action<DefaultRuntimeFeature>? configure = default)
{
feature.Module.Configure(configure);
return feature;
}
public static WorkflowRuntimeFeature UseExecutionLogRecords(this WorkflowRuntimeFeature feature, Action<ExecutionLogRecordFeature>? configure = default)
{
feature.Module.Configure(configure);
return feature;
}
}

11
src/modules/Elsa.Workflows.Runtime/Features/DefaultRuntimeFeature.cs Normal file
View file

@ -0,0 +1,11 @@
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
namespace Elsa.Workflows.Runtime.Features;
public class DefaultRuntimeFeature : FeatureBase
{
public DefaultRuntimeFeature(IModule module) : base(module)
{
}
}

11
src/modules/Elsa.Workflows.Runtime/Features/ExecutionLogRecordFeature.cs Normal file
View file

@ -0,0 +1,11 @@
using Elsa.Features.Abstractions;
using Elsa.Features.Services;
namespace Elsa.Workflows.Runtime.Features;
public class ExecutionLogRecordFeature : FeatureBase
{
public ExecutionLogRecordFeature(IModule module) : base(module)
{
}
}

3
src/samples/aspnet/Elsa.Samples.TelnyxIntegration/Program.cs
View file

@ -57,7 +57,8 @@ services
{
//proto.PersistenceProvider = _ => new SqliteProvider(new SqliteConnectionStringBuilder(sqliteConnectionString));
});
runtime.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString));
runtime.UseDefaultRuntime(dr => dr.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)));
runtime.UseExecutionLogRecords(e => e.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)));
runtime.WorkflowStateExporter = sp => sp.GetRequiredService<AsyncWorkflowStateExporter>();
})
.UseActivityDefinitions(feature => feature.UseEntityFrameworkCore(ef => ef.UseSqlite(sqliteConnectionString)))