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

Refactor proto files and fix snapshot loading of running workflows grain (#4371)

Browse source
This commit is contained in:
Sipke Schoorstra 2023-08-29 23:29:31 +02:00 committed by GitHub
parent deb6b34c98
commit e2c3ab0efd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
27 changed files with 209 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -9,6 +9,7 @@
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.SqlServer\Elsa.EntityFrameworkCore.SqlServer.csproj" />
<ProjectReference Include="..\Elsa\Elsa.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Elasticsearch\Elsa.Elasticsearch.csproj" />
<ProjectReference Include="..\..\modules\Elsa.Email\Elsa.Email.csproj" />
@ -34,6 +35,7 @@
<ItemGroup>
<PackageReference Include="Azure.Identity" Version="1.8.2" />
<PackageReference Include="Proto.Persistence.Sqlite" Version="1.1.0" />
<PackageReference Include="Proto.Persistence.SqlServer" Version="1.3.1-alpha.0.2" />
</ItemGroup>
</Project>

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

@ -1,3 +1,4 @@
using Elsa.EntityFrameworkCore.Extensions;
using Elsa.EntityFrameworkCore.Modules.Identity;
using Elsa.EntityFrameworkCore.Modules.Management;
using Elsa.EntityFrameworkCore.Modules.Runtime;
@ -10,9 +11,11 @@ using Elsa.MongoDb.Modules.Runtime;
using Elsa.WorkflowServer.Web;
using Microsoft.Data.Sqlite;
using Proto.Persistence.Sqlite;
using Proto.Persistence.SqlServer;
const bool useMongoDb = false;
const bool useProtoActor = false;
const bool useSqlServer = true;
const bool useProtoActor = true;
const bool useHangfire = false;
var builder = WebApplication.CreateBuilder(args);
@ -21,6 +24,7 @@ var configuration = builder.Configuration;
var identitySection = configuration.GetSection("Identity");
var identityTokenSection = identitySection.GetSection("Tokens");
var sqliteConnectionString = configuration.GetConnectionString("Sqlite");
var sqlServerConnectionString = configuration.GetConnectionString("SqlServer");
var mongoDbConnectionString = configuration.GetConnectionString("MongoDb")!;
// Add Elsa services.
@ -43,7 +47,13 @@ services
if(useMongoDb)
identity.UseMongoDb();
else
identity.UseEntityFrameworkCore();
identity.UseEntityFrameworkCore(ef =>
{
if(useSqlServer)
ef.UseSqlServer(sqlServerConnectionString!);
else
ef.UseSqlite(sqliteConnectionString);
});
identity.IdentityOptions = options => identitySection.Bind(options);
identity.TokenOptions = options => identityTokenSection.Bind(options);
@ -57,7 +67,13 @@ services
if(useMongoDb)
management.UseMongoDb();
else
management.UseEntityFrameworkCore();
management.UseEntityFrameworkCore(ef =>
{
if(useSqlServer)
ef.UseSqlServer(sqlServerConnectionString!);
else
ef.UseSqlite(sqliteConnectionString);
});
management.AddVariableType<ApiResponse<User>>("Api");
management.AddVariableType<User>("Api");
@ -68,17 +84,37 @@ services
if(useMongoDb)
runtime.UseMongoDb();
else
runtime.UseEntityFrameworkCore();
runtime.UseEntityFrameworkCore(ef =>
{
if(useSqlServer)
ef.UseSqlServer(sqlServerConnectionString!);
else
ef.UseSqlite(sqliteConnectionString);
});
if(useProtoActor)
runtime.UseProtoActor(proto => proto.PersistenceProvider = _ => new SqliteProvider(new SqliteConnectionStringBuilder(sqliteConnectionString)));
{
runtime.UseProtoActor(proto => proto.PersistenceProvider = _ =>
{
if(useSqlServer)
return new SqlServerProvider(sqlServerConnectionString!, true, "", "proto_actor");
else
return new SqliteProvider(new SqliteConnectionStringBuilder(sqliteConnectionString));
});
}
else
runtime.UseDefaultRuntime(dr =>
{
if(useMongoDb)
dr.UseMongoDb();
else
dr.UseEntityFrameworkCore();
dr.UseEntityFrameworkCore(ef =>
{
if(useSqlServer)
ef.UseSqlServer(sqlServerConnectionString!);
else
ef.UseSqlite(sqliteConnectionString);
});
});
runtime.UseExecutionLogRecords(e =>
@ -86,7 +122,13 @@ services
if(useMongoDb)
e.UseMongoDb();
else
e.UseEntityFrameworkCore();
e.UseEntityFrameworkCore(ef =>
{
if(useSqlServer)
ef.UseSqlServer(sqlServerConnectionString!);
else
ef.UseSqlite(sqliteConnectionString);
});
});
runtime.UseMassTransitDispatcher();

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

@ -6,6 +6,12 @@ namespace Elsa.EntityFrameworkCore.Extensions;
public static partial class Extensions
{
public static EFCoreWorkflowRuntimePersistenceFeature UseSqlServer(this EFCoreWorkflowRuntimePersistenceFeature feature, string connectionString, ElsaDbContextOptions? options = default)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlServer(connectionString, options);
return feature;
}
public static EFCoreDefaultWorkflowRuntimePersistenceFeature UseSqlServer(this EFCoreDefaultWorkflowRuntimePersistenceFeature feature, string connectionString, ElsaDbContextOptions? options = default)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlServer(connectionString, options);

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

@ -11,8 +11,8 @@ public static partial class Extensions
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlite(connectionString, options);
return feature;
}
public static EFCoreDefaultWorkflowRuntimePersistenceFeature UseSqlite(this EFCoreDefaultWorkflowRuntimePersistenceFeature feature, string connectionString = Constants.DefaultConnectionString, ElsaDbContextOptions? options = default)
{
feature.DbContextOptionsBuilder = (_, db) => db.UseElsaSqlite(connectionString, options);

34
src/modules/Elsa.ProtoActor/Elsa.ProtoActor.csproj
View file

@ -1,7 +1,7 @@
<Project Sdk="Microsoft.NET.Sdk">
<Import Project="..\..\..\common.props" />
<Import Project="..\..\..\configureawait.props" />
<Import Project="..\..\..\common.props"/>
<Import Project="..\..\..\configureawait.props"/>
<PropertyGroup>
<TargetFrameworks>net6.0;net7.0</TargetFrameworks>
@ -12,31 +12,35 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Google.Protobuf" Version="3.23.4" />
<PackageReference Include="Google.Protobuf" Version="3.23.4"/>
<PackageReference Include="Grpc.Tools" Version="2.51.0">
<PrivateAssets>all</PrivateAssets>
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
</PackageReference>
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="7.0.0" />
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="7.0.0"/>
</ItemGroup>
<ItemGroup Label="ProtoActor">
<PackageReference Include="Proto.Actor" Version="1.3.0" />
<PackageReference Include="Proto.Cluster" Version="1.3.0" />
<PackageReference Include="Proto.Cluster.CodeGen" Version="1.3.0" />
<PackageReference Include="Proto.Cluster.TestProvider" Version="1.3.0" />
<PackageReference Include="Proto.Persistence" Version="1.3.0" />
<PackageReference Include="Proto.Remote" Version="1.3.0" />
<PackageReference Include="Proto.Actor" Version="1.3.0"/>
<PackageReference Include="Proto.Cluster" Version="1.3.0"/>
<PackageReference Include="Proto.Cluster.CodeGen" Version="1.3.0"/>
<PackageReference Include="Proto.Cluster.TestProvider" Version="1.3.0"/>
<PackageReference Include="Proto.Persistence" Version="1.3.0"/>
<PackageReference Include="Proto.Remote" Version="1.3.0"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj" />
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
<ProjectReference Include="..\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj"/>
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj"/>
</ItemGroup>
<ItemGroup Label="Protos">
<Protobuf Include="Protos\Messages.proto" />
<ProtoGrain Include="Protos\Grains.proto" AdditionalImportDirs="Protos" />
<ItemGroup Label="Proto">
<Protobuf Include="Proto\Shared.proto" AdditionalImportDirs="./Proto"/>
<Protobuf Include="Proto\RunningWorkflows.Messages.proto" AdditionalImportDirs="./Proto"/>
<Protobuf Include="Proto\WorkflowInstance.Messages.proto" AdditionalImportDirs="./Proto"/>
<ProtoGrain Include="Proto\RunningWorkflows.proto" AdditionalImportDirs="./Proto"/>
<ProtoGrain Include="Proto\WorkflowInstance.proto" AdditionalImportDirs="./Proto"/>
</ItemGroup>
</Project>

6
src/modules/Elsa.ProtoActor/Extensions/ClusterExtensions.cs
View file

@ -1,5 +1,5 @@
using Elsa.ProtoActor.Grains;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.ProtoBuf;
using Proto.Cluster;
// ReSharper disable once CheckNamespace
@ -7,6 +7,6 @@ namespace Elsa.Extensions;
internal static class ClusterExtensions
{
public static RunningWorkflowsGrainClient GetNamedRunningWorkflowsGrain(this Cluster cluster) => cluster.GetRunningWorkflowsGrain(nameof(RunningWorkflowsGrain));
public static WorkflowGrainClient GetNamedWorkflowGrain(this Cluster cluster, string workflowInstanceId) => cluster.GetWorkflowGrain($"{nameof(WorkflowGrain)}-{workflowInstanceId}");
public static RunningWorkflowsClient GetNamedRunningWorkflowsGrain(this Cluster cluster) => cluster.GetRunningWorkflows(nameof(RunningWorkflows));
public static WorkflowInstanceClient GetNamedWorkflowGrain(this Cluster cluster, string workflowInstanceId) => cluster.GetWorkflowInstance($"{nameof(WorkflowInstance)}-{workflowInstanceId}");
}

2
src/modules/Elsa.ProtoActor/Extensions/ProtoInputExtensions.cs
View file

@ -1,5 +1,5 @@
using System.Text.Json;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.ProtoBuf;
using Elsa.Workflows.Core.Serialization.Converters;
namespace Elsa.ProtoActor.Extensions;

16
src/modules/Elsa.ProtoActor/Features/ProtoActorFeature.cs
View file

@ -5,7 +5,7 @@ using Elsa.Features.Services;
using Elsa.ProtoActor.Grains;
using Elsa.ProtoActor.HostedServices;
using Elsa.ProtoActor.Mappers;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.ProtoBuf;
using Elsa.ProtoActor.Services;
using Elsa.Workflows.Core.Features;
using Elsa.Workflows.Runtime.Features;
@ -95,8 +95,8 @@ public class ProtoActorFeature : FeatureBase
var clusterProvider = ClusterProvider(sp);
var system = new ActorSystem(systemConfig).WithServiceProvider(sp);
var workflowGrainProps = system.DI().PropsFor<WorkflowGrainActor>();
var workflowRegistryGrainProps = system.DI().PropsFor<RunningWorkflowsGrainActor>();
var workflowGrainProps = system.DI().PropsFor<WorkflowInstanceActor>();
var workflowRegistryGrainProps = system.DI().PropsFor<RunningWorkflowsActor>();
var clusterConfig = ClusterConfig
.Setup(ClusterName, clusterProvider, new PartitionIdentityLookup())
@ -104,8 +104,8 @@ public class ProtoActorFeature : FeatureBase
.WithActorRequestTimeout(TimeSpan.FromHours(1))
.WithActorActivationTimeout(TimeSpan.FromHours(1))
.WithActorSpawnVerificationTimeout(TimeSpan.FromHours(1))
.WithClusterKind(WorkflowGrainActor.Kind, workflowGrainProps)
.WithClusterKind(RunningWorkflowsGrainActor.Kind, workflowRegistryGrainProps)
.WithClusterKind(WorkflowInstanceActor.Kind, workflowGrainProps)
.WithClusterKind(RunningWorkflowsActor.Kind, workflowRegistryGrainProps)
;
ActorSystemConfig(sp, systemConfig);
@ -143,8 +143,8 @@ public class ProtoActorFeature : FeatureBase
// Actors.
services
.AddTransient(sp => new WorkflowGrainActor((context, _) => ActivatorUtilities.CreateInstance<WorkflowGrain>(sp, context)))
.AddTransient(sp => new RunningWorkflowsGrainActor((context, _) => ActivatorUtilities.CreateInstance<RunningWorkflowsGrain>(sp, context)))
.AddTransient(sp => new WorkflowInstanceActor((context, _) => ActivatorUtilities.CreateInstance<WorkflowInstance>(sp, context)))
.AddTransient(sp => new RunningWorkflowsActor((context, _) => ActivatorUtilities.CreateInstance<RunningWorkflows>(sp, context)))
;
}
@ -154,6 +154,6 @@ public class ProtoActorFeature : FeatureBase
private static GrpcNetRemoteConfig CreateDefaultRemoteConfig(IServiceProvider serviceProvider) =>
GrpcNetRemoteConfig.BindToLocalhost()
.WithProtoMessages(MessagesReflection.Descriptor)
.WithProtoMessages(SharedReflection.Descriptor)
.WithProtoMessages(EmptyReflection.Descriptor);
}

30
src/modules/Elsa.ProtoActor/Grains/RunningWorkflowsGrain.cs → src/modules/Elsa.ProtoActor/Grains/RunningWorkflows.cs
View file

@ -1,5 +1,6 @@
using Elsa.ProtoActor.Extensions;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.Models;
using Elsa.ProtoActor.ProtoBuf;
using Proto;
using Proto.Cluster;
using Proto.Persistence;
@ -7,19 +8,18 @@ using Proto.Persistence.SnapshotStrategies;
namespace Elsa.ProtoActor.Grains;
// TODO: Replace this grain with a store-based implementation that stores metrics.
/// <summary>
/// Represents a registry of workflow instances for a given workflow definition version.
/// </summary>
public class RunningWorkflowsGrain : RunningWorkflowsGrainBase
public class RunningWorkflows : RunningWorkflowsBase
{
private const int EventsPerSnapshot = 100;
private const int EventsPerSnapshot = 1;
private readonly Persistence _persistence;
private IDictionary<string, WorkflowInstanceEntry> _lookupByInstanceId = new Dictionary<string, WorkflowInstanceEntry>();
private IDictionary<string, WorkflowInstanceEntry> _lookupByCorrelationId = new Dictionary<string, WorkflowInstanceEntry>();
private IDictionary<string, RunningWorkflowInstanceEntry> _lookupByInstanceId = new Dictionary<string, RunningWorkflowInstanceEntry>();
private IDictionary<string, RunningWorkflowInstanceEntry> _lookupByCorrelationId = new Dictionary<string, RunningWorkflowInstanceEntry>();
/// <inheritdoc />
public RunningWorkflowsGrain(IProvider provider, IContext context) : base(context)
public RunningWorkflows(IProvider provider, IContext context) : base(context)
{
_persistence = Persistence.WithEventSourcingAndSnapshotting(
provider,
@ -31,6 +31,12 @@ public class RunningWorkflowsGrain : RunningWorkflowsGrainBase
GetState);
}
/// <inheritdoc />
public override async Task OnStarted()
{
await _persistence.RecoverStateAsync();
}
/// <inheritdoc />
public override async Task Register(RegisterRunningWorkflowRequest request) => await _persistence.PersistRollingEventAsync(request, EventsPerSnapshot);
@ -58,7 +64,7 @@ public class RunningWorkflowsGrain : RunningWorkflowsGrainBase
private void ApplySnapshot(Snapshot snapshot)
{
var registrySnapshot = (WorkflowRegistrySnapshot)snapshot.State;
var registrySnapshot = (RunningWorkflowsSnapshot)snapshot.State;
_lookupByCorrelationId = registrySnapshot.Entries.Where(x => !string.IsNullOrEmpty(x.CorrelationId)).ToDictionary(x => x.CorrelationId!);
_lookupByInstanceId = registrySnapshot.Entries.ToDictionary(x => x.InstanceId);
}
@ -76,11 +82,11 @@ public class RunningWorkflowsGrain : RunningWorkflowsGrainBase
}
}
private object GetState() => new WorkflowRegistrySnapshot(_lookupByInstanceId.Values);
private object GetState() => new RunningWorkflowsSnapshot(_lookupByInstanceId.Values.ToList());
private void RegisterInternal(RegisterRunningWorkflowRequest request)
{
var entry = new WorkflowInstanceEntry(request.DefinitionId, request.Version, request.InstanceId, request.CorrelationId);
var entry = new RunningWorkflowInstanceEntry(request.DefinitionId, request.Version, request.InstanceId, request.CorrelationId);
_lookupByInstanceId[request.InstanceId] = entry;
if (!string.IsNullOrEmpty(request.CorrelationId))
@ -96,6 +102,4 @@ public class RunningWorkflowsGrain : RunningWorkflowsGrainBase
_lookupByInstanceId.Remove(request.InstanceId);
}
}
internal record WorkflowInstanceEntry(string DefinitionId, int Version, string InstanceId, string? CorrelationId);
}

13
src/modules/Elsa.ProtoActor/Grains/WorkflowGrain.cs → src/modules/Elsa.ProtoActor/Grains/WorkflowInstance.cs
View file

@ -1,7 +1,8 @@
using Elsa.Common.Models;
using Elsa.Extensions;
using Elsa.ProtoActor.Extensions;
using Elsa.ProtoActor.Mappers;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.ProtoBuf;
using Elsa.Workflows.Core.Contracts;
using Elsa.Workflows.Core.State;
using Elsa.Workflows.Management.Contracts;
@ -11,14 +12,14 @@ using Proto.Cluster;
using Proto.Persistence;
using Exception = System.Exception;
using WorkflowStatus = Elsa.Workflows.Core.WorkflowStatus;
using ProtoBookmark = Elsa.ProtoActor.Protos.Bookmark;
using ProtoBookmark = Elsa.ProtoActor.ProtoBuf.Bookmark;
namespace Elsa.ProtoActor.Grains;
/// <summary>
/// Executes a workflow.
/// </summary>
internal class WorkflowGrain : WorkflowGrainBase
internal class WorkflowInstance : WorkflowInstanceBase
{
private const int MaxSnapshotsToKeep = 5;
private readonly IWorkflowDefinitionService _workflowDefinitionService;
@ -38,7 +39,7 @@ internal class WorkflowGrain : WorkflowGrainBase
private WorkflowState _workflowState = default!;
/// <inheritdoc />
public WorkflowGrain(
public WorkflowInstance(
IWorkflowDefinitionService workflowDefinitionService,
IWorkflowHostFactory workflowHostFactory,
IWorkflowStateSerializer workflowStateSerializer,
@ -140,7 +141,7 @@ internal class WorkflowGrain : WorkflowGrainBase
var startWorkflowOptions = new StartWorkflowHostOptions(instanceId, correlationId, input, request.TriggerActivityId);
await _workflowHost.StartWorkflowAsync(startWorkflowOptions, cancellationToken);
var workflowState = _workflowHost.WorkflowState;
var result = workflowState.Status == WorkflowStatus.Finished ? Protos.RunWorkflowResult.Finished : Protos.RunWorkflowResult.Suspended;
var result = workflowState.Status == WorkflowStatus.Finished ? ProtoBuf.RunWorkflowResult.Finished : ProtoBuf.RunWorkflowResult.Suspended;
_workflowState = workflowState;
@ -261,7 +262,7 @@ internal class WorkflowGrain : WorkflowGrainBase
await _persistence.PersistRollingSnapshotAsync(GetState(), MaxSnapshotsToKeep);
}
private object GetState() => new WorkflowSnapshot(_definitionId, _instanceId, _version, _workflowState, _input);
private object GetState() => new WorkflowSnapshot(_definitionId, _instanceId, _version, _workflowState, _input?.ToDictionary(x => x.Key, x => x.Value));
private async Task<IWorkflowHost> CreateWorkflowHostAsync(string definitionId, VersionOptions versionOptions, CancellationToken cancellationToken)
{

2
src/modules/Elsa.ProtoActor/Handlers/StopRunningWorkflows.cs
View file

@ -1,6 +1,6 @@
using Elsa.Extensions;
using Elsa.Mediator.Contracts;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.ProtoBuf;
using Elsa.Workflows.Management.Notifications;
using JetBrains.Annotations;
using Proto.Cluster;

4
src/modules/Elsa.ProtoActor/Handlers/UpdateRunningWorkflows.cs
View file

@ -2,7 +2,7 @@ using Elsa.Extensions;
using Elsa.Mediator.Contracts;
using Elsa.ProtoActor.Extensions;
using Elsa.ProtoActor.Grains;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.ProtoBuf;
using Elsa.Workflows.Core.Notifications;
using JetBrains.Annotations;
using Proto.Cluster;
@ -11,7 +11,7 @@ using WorkflowStatus = Elsa.Workflows.Core.WorkflowStatus;
namespace Elsa.ProtoActor.Handlers;
/// <summary>
/// Updates the <see cref="RunningWorkflowsGrain"/> with running workflow instances.
/// Updates the <see cref="RunningWorkflows"/> with running workflow instances.
/// </summary>
[PublicAPI]
internal class UpdateRunningWorkflows : INotificationHandler<WorkflowExecuted>

2
src/modules/Elsa.ProtoActor/Mappers/BookmarkMapper.cs
View file

@ -1,7 +1,7 @@
using Elsa.ProtoActor.Extensions;
using Elsa.Workflows.Core.Contracts;
using Elsa.Workflows.Core.Models;
using ProtoBookmark = Elsa.ProtoActor.Protos.Bookmark;
using ProtoBookmark = Elsa.ProtoActor.ProtoBuf.Bookmark;
namespace Elsa.ProtoActor.Mappers;

2
src/modules/Elsa.ProtoActor/Mappers/ExceptionMapper.cs
View file

@ -1,5 +1,5 @@
using Elsa.Workflows.Core.State;
using ProtoException = Elsa.ProtoActor.Protos.ExceptionState;
using ProtoException = Elsa.ProtoActor.ProtoBuf.ExceptionState;
namespace Elsa.ProtoActor.Mappers;

2
src/modules/Elsa.ProtoActor/Mappers/WorkflowExecutionResultMapper.cs
View file

@ -1,6 +1,6 @@
using Elsa.ProtoActor.Extensions;
using Elsa.Workflows.Runtime.Contracts;
using ProtoWorkflowExecutionResponse = Elsa.ProtoActor.Protos.WorkflowExecutionResponse;
using ProtoWorkflowExecutionResponse = Elsa.ProtoActor.ProtoBuf.WorkflowExecutionResponse;
namespace Elsa.ProtoActor.Mappers;

2
src/modules/Elsa.ProtoActor/Mappers/WorkflowFaultStateMapper.cs
View file

@ -1,5 +1,5 @@
using Elsa.Workflows.Core.State;
using ProtoWorkflowFault = Elsa.ProtoActor.Protos.WorkflowFault;
using ProtoWorkflowFault = Elsa.ProtoActor.ProtoBuf.WorkflowFault;
namespace Elsa.ProtoActor.Mappers;

2
src/modules/Elsa.ProtoActor/Mappers/WorkflowStatusMapper.cs
View file

@ -1,5 +1,5 @@
using Elsa.Workflows.Core;
using ProtoWorkflowStatus = Elsa.ProtoActor.Protos.WorkflowStatus;
using ProtoWorkflowStatus = Elsa.ProtoActor.ProtoBuf.WorkflowStatus;
namespace Elsa.ProtoActor.Mappers;

2
src/modules/Elsa.ProtoActor/Mappers/WorkflowSubStatusMapper.cs
View file

@ -1,5 +1,5 @@
using Elsa.Workflows.Core;
using ProtoWorkflowSubStatus = Elsa.ProtoActor.Protos.WorkflowSubStatus;
using ProtoWorkflowSubStatus = Elsa.ProtoActor.ProtoBuf.WorkflowSubStatus;
namespace Elsa.ProtoActor.Mappers;

10
src/modules/Elsa.ProtoActor/Models/RunningWorkflowInstanceEntry.cs Normal file
View file

@ -0,0 +1,10 @@
namespace Elsa.ProtoActor.Models;
/// <summary>
/// A snapshot of information about a running workflow instance.
/// </summary>
/// <param name="DefinitionId">The workflow definition id.</param>
/// <param name="Version">The workflow definition version.</param>
/// <param name="InstanceId">The workflow instance ID.</param>
/// <param name="CorrelationId">The workflow instance correlation ID.</param>
public record RunningWorkflowInstanceEntry(string DefinitionId, int Version, string InstanceId, string? CorrelationId);

27
src/modules/Elsa.ProtoActor/Proto/RunningWorkflows.Messages.proto Normal file
View file

@ -0,0 +1,27 @@
syntax = "proto3";
option csharp_namespace = "Elsa.ProtoActor.ProtoBuf";
package Elsa.ProtoActor.ProtoBuf;
import "google/protobuf/empty.proto";
import "Shared.proto";
message RegisterRunningWorkflowRequest {
string DefinitionId = 1;
int32 Version = 2;
string InstanceId = 3;
optional string CorrelationId = 6;
}
message UnregisterRunningWorkflowRequest {
string InstanceId = 1;
}
message CountRunningWorkflowsRequest {
string DefinitionId = 1;
int32 Version = 2;
optional string CorrelationId = 3;
}
message CountRunningWorkflowsResponse {
optional int32 Count = 1;
}

13
src/modules/Elsa.ProtoActor/Proto/RunningWorkflows.proto Normal file
View file

@ -0,0 +1,13 @@
syntax = "proto3";
option csharp_namespace = "Elsa.ProtoActor.ProtoBuf";
package Elsa.ProtoActor.ProtoBuf;
import "google/protobuf/empty.proto";
import "Shared.proto";
import "RunningWorkflows.Messages.proto";
service RunningWorkflows {
rpc Register(RegisterRunningWorkflowRequest) returns (google.protobuf.Empty);
rpc Unregister(UnregisterRunningWorkflowRequest) returns (google.protobuf.Empty);
rpc Count (CountRunningWorkflowsRequest) returns (CountRunningWorkflowsResponse);
}

15
src/modules/Elsa.ProtoActor/Proto/Shared.proto Normal file
View file

@ -0,0 +1,15 @@
syntax = "proto3";
option csharp_namespace = "Elsa.ProtoActor.ProtoBuf";
package Elsa.ProtoActor.ProtoBuf;
import "google/protobuf/empty.proto";
import "google/protobuf/wrappers.proto";
// Shared.
message Json {
string text = 1;
}
message Input {
map<string, Json> Data = 1;
}

38
src/modules/Elsa.ProtoActor/Protos/Messages.proto → src/modules/Elsa.ProtoActor/Proto/WorkflowInstance.Messages.proto
View file

@ -1,19 +1,11 @@
syntax = "proto3";
option csharp_namespace = "Elsa.ProtoActor.ProtoBuf";
package Elsa.ProtoActor.ProtoBuf;
import "google/protobuf/empty.proto";
import "google/protobuf/wrappers.proto";
package Elsa.ProtoActor.Protos;
option csharp_namespace = "Elsa.ProtoActor.Protos";
import "Shared.proto";
// Shared.
message Json {
string text = 1;
}
message Input {
map<string, Json> Data = 1;
}
// WorkflowGrain.
message CanStartWorkflowResponse {
bool CanStart = 1;
}
@ -102,26 +94,4 @@ message Bookmark {
optional bool AutoBurn = 8;
optional string CallbackMethodName = 9;
string CreatedAt = 10; // ISO 8601
}
// RunningWorkflowsGrain.
message RegisterRunningWorkflowRequest {
string DefinitionId = 1;
int32 Version = 2;
string InstanceId = 3;
optional string CorrelationId = 6;
}
message UnregisterRunningWorkflowRequest {
string InstanceId = 1;
}
message CountRunningWorkflowsRequest {
string DefinitionId = 1;
int32 Version = 2;
optional string CorrelationId = 3;
}
message CountRunningWorkflowsResponse {
optional int32 Count = 1;
}

16
src/modules/Elsa.ProtoActor/Protos/Grains.proto → src/modules/Elsa.ProtoActor/Proto/WorkflowInstance.proto
View file

@ -1,21 +1,17 @@
syntax = "proto3";
option csharp_namespace = "Elsa.ProtoActor.Protos";
option csharp_namespace = "Elsa.ProtoActor.ProtoBuf";
package Elsa.ProtoActor.ProtoBuf;
import "google/protobuf/empty.proto";
import "Messages.proto";
import "google/protobuf/wrappers.proto";
import "Shared.proto";
import "WorkflowInstance.Messages.proto";
service WorkflowGrain {
service WorkflowInstance {
rpc CanStart (StartWorkflowRequest) returns (CanStartWorkflowResponse);
rpc Start (StartWorkflowRequest) returns (WorkflowExecutionResponse);
rpc Stop (Empty) returns (Empty);
rpc Resume (ResumeWorkflowRequest) returns (WorkflowExecutionResponse);
rpc ExportState(ExportWorkflowStateRequest) returns (ExportWorkflowStateResponse);
rpc ImportState(ImportWorkflowStateRequest) returns (ImportWorkflowStateResponse);
}
service RunningWorkflowsGrain {
rpc Register(RegisterRunningWorkflowRequest) returns (google.protobuf.Empty);
rpc Unregister(UnregisterRunningWorkflowRequest) returns (google.protobuf.Empty);
rpc Count (CountRunningWorkflowsRequest) returns (CountRunningWorkflowsResponse);
}

14
src/modules/Elsa.ProtoActor/Services/ProtoActorWorkflowRuntime.cs
View file

@ -2,7 +2,7 @@ using Elsa.Common.Models;
using Elsa.Extensions;
using Elsa.ProtoActor.Extensions;
using Elsa.ProtoActor.Mappers;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.ProtoBuf;
using Elsa.Workflows.Core.Contracts;
using Elsa.Workflows.Core.State;
using Elsa.Workflows.Management.Contracts;
@ -11,12 +11,12 @@ using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Filters;
using Proto.Cluster;
using Bookmark = Elsa.Workflows.Core.Models.Bookmark;
using ProtoWorkflowStatus = Elsa.ProtoActor.Protos.WorkflowStatus;
using ProtoWorkflowSubStatus = Elsa.ProtoActor.Protos.WorkflowSubStatus;
using ProtoWorkflowFault = Elsa.ProtoActor.Protos.WorkflowFault;
using ProtoException = Elsa.ProtoActor.Protos.ExceptionState;
using ProtoWorkflowExecutionResponse = Elsa.ProtoActor.Protos.WorkflowExecutionResponse;
using ProtoBookmark = Elsa.ProtoActor.Protos.Bookmark;
using ProtoWorkflowStatus = Elsa.ProtoActor.ProtoBuf.WorkflowStatus;
using ProtoWorkflowSubStatus = Elsa.ProtoActor.ProtoBuf.WorkflowSubStatus;
using ProtoWorkflowFault = Elsa.ProtoActor.ProtoBuf.WorkflowFault;
using ProtoException = Elsa.ProtoActor.ProtoBuf.ExceptionState;
using ProtoWorkflowExecutionResponse = Elsa.ProtoActor.ProtoBuf.WorkflowExecutionResponse;
using ProtoBookmark = Elsa.ProtoActor.ProtoBuf.Bookmark;
namespace Elsa.ProtoActor.Services;

6
src/modules/Elsa.ProtoActor/Snapshots.cs
View file

@ -1,8 +1,8 @@
using Elsa.ProtoActor.Grains;
using Elsa.ProtoActor.Models;
using Elsa.Workflows.Core.State;
namespace Elsa.ProtoActor;
internal record WorkflowSnapshot(string DefinitionId, string InstanceId, int Version, WorkflowState WorkflowState, IDictionary<string, object>? Input);
internal record WorkflowSnapshot(string DefinitionId, string InstanceId, int Version, WorkflowState WorkflowState, Dictionary<string, object>? Input);
internal record WorkflowRegistrySnapshot(ICollection<WorkflowInstanceEntry> Entries);
internal record RunningWorkflowsSnapshot(List<RunningWorkflowInstanceEntry> Entries);

4
src/samples/aspnet/Elsa.Samples.AspNet.ProtoActorRuntime.AzureContainerApps/Program.cs
View file

@ -3,7 +3,7 @@ using Elsa.EntityFrameworkCore.Modules.Labels;
using Elsa.EntityFrameworkCore.Modules.Management;
using Elsa.EntityFrameworkCore.Modules.Runtime;
using Elsa.Extensions;
using Elsa.ProtoActor.Protos;
using Elsa.ProtoActor.ProtoBuf;
using Google.Protobuf.WellKnownTypes;
using Microsoft.Data.Sqlite;
using Proto.Cluster.AzureContainerApps;
@ -69,7 +69,7 @@ services
protoActor.RemoteConfig = _ => GrpcNetRemoteConfig
.BindTo(advertisedHost)
.WithProtoMessages(EmptyReflection.Descriptor)
.WithProtoMessages(MessagesReflection.Descriptor)
.WithProtoMessages(SharedReflection.Descriptor)
.WithLogLevelForDeserializationErrors(LogLevel.Critical)
.WithRemoteDiagnostics(true); // required by proto.actor dashboard