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

WIP: Orleans grains

Browse source
This commit is contained in:
Sipke Schoorstra 2021-03-30 15:18:48 +02:00
parent 4cf43882dc
commit 0d3e19dcd3
21 changed files with 344 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
Elsa.sln
View file

@ -245,6 +245,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Telnyx", "s
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Webhooks", "src\activities\Elsa.Activities.Webhooks\Elsa.Activities.Webhooks.csproj", "{2B67E954-3B04-402D-A9A7-AAAB1D6C3215}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Server.Orleans", "src\server\Elsa.Server.Orleans\Elsa.Server.Orleans.csproj", "{76BD888E-F0FD-40DD-B025-2ED773C1C50B}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
@ -591,6 +593,10 @@ Global
{2B67E954-3B04-402D-A9A7-AAAB1D6C3215}.Debug|Any CPU.Build.0 = Debug|Any CPU
{2B67E954-3B04-402D-A9A7-AAAB1D6C3215}.Release|Any CPU.ActiveCfg = Release|Any CPU
{2B67E954-3B04-402D-A9A7-AAAB1D6C3215}.Release|Any CPU.Build.0 = Release|Any CPU
{76BD888E-F0FD-40DD-B025-2ED773C1C50B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{76BD888E-F0FD-40DD-B025-2ED773C1C50B}.Debug|Any CPU.Build.0 = Debug|Any CPU
{76BD888E-F0FD-40DD-B025-2ED773C1C50B}.Release|Any CPU.ActiveCfg = Release|Any CPU
{76BD888E-F0FD-40DD-B025-2ED773C1C50B}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
@ -708,6 +714,7 @@ Global
{E3EA6449-28EC-48E4-91C4-E82DE08A7DD9} = {C865B0FD-E505-48F0-BFAF-0D4D7C1B5CA1}
{93878BAD-855D-48D1-97ED-C776EC54E19E} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180}
{2B67E954-3B04-402D-A9A7-AAAB1D6C3215} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180}
{76BD888E-F0FD-40DD-B025-2ED773C1C50B} = {468E498C-59DB-4541-8D8A-0D89DE9083AB}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158}

2
src/core/Elsa.Abstractions/Dispatch/ICorrelatingWorkflowDispatcher.cs
View file

@ -9,6 +9,6 @@ namespace Elsa.Dispatch
/// </summary>
public interface ICorrelatingWorkflowDispatcher
{
Task DispatchAsync(ExecuteCorrelatedWorkflowRequest request, CancellationToken? cancellationToken = default);
Task DispatchAsync(ExecuteCorrelatedWorkflowRequest request, CancellationToken cancellationToken = default);
}
}

2
src/core/Elsa.Abstractions/Dispatch/IWorkflowDefinitionDispatcher.cs
View file

@ -8,6 +8,6 @@ namespace Elsa.Dispatch
/// </summary>
public interface IWorkflowDefinitionDispatcher
{
Task DispatchAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken? cancellationToken = default);
Task DispatchAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default);
}
}

2
src/core/Elsa.Abstractions/Dispatch/IWorkflowInstanceDispatcher.cs
View file

@ -8,6 +8,6 @@ namespace Elsa.Dispatch
/// </summary>
public interface IWorkflowInstanceDispatcher
{
Task DispatchAsync(ExecuteWorkflowInstanceRequest request, CancellationToken? cancellationToken = default);
Task DispatchAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default);
}
}

1
src/core/Elsa.Core/Dispatch/Consumers/ExecuteCorrelatedWorkflowRequestConsumer.cs
View file

@ -67,7 +67,6 @@ namespace Elsa.Dispatch.Consumers
}
else
{
// Trigger new workflow.
_logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId);
await TriggerNewWorkflowAsync(message);
}

6
src/core/Elsa.Core/Dispatch/QueuingWorkflowDispatcher.cs
View file

@ -11,8 +11,8 @@ namespace Elsa.Dispatch
{
private readonly ICommandSender _commandSender;
public QueuingWorkflowDispatcher(ICommandSender commandSender) => _commandSender = commandSender;
public async Task DispatchAsync(ExecuteWorkflowInstanceRequest request, CancellationToken? cancellationToken = default) => await _commandSender.SendAsync(request);
public async Task DispatchAsync(ExecuteCorrelatedWorkflowRequest request, CancellationToken? cancellationToken = default) => await _commandSender.SendAsync(request);
public async Task DispatchAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken? cancellationToken = default) => await _commandSender.SendAsync(request);
public async Task DispatchAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default) => await _commandSender.SendAsync(request);
public async Task DispatchAsync(ExecuteCorrelatedWorkflowRequest request, CancellationToken cancellationToken = default) => await _commandSender.SendAsync(request);
public async Task DispatchAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default) => await _commandSender.SendAsync(request);
}
}

37
src/core/Elsa.Core/Extensions/WorkflowRegistryExtensions.cs
View file

@ -24,33 +24,14 @@ namespace Elsa
string id,
VersionOptions versionOptions,
CancellationToken cancellationToken = default) =>
workflowRegistry.GetAsync(id, default, versionOptions, cancellationToken);
// public static async Task<IEnumerable<(IWorkflowBlueprint Workflow, IActivityBlueprint Activity)>>
// GetWorkflowsByStartActivityAsync<T>(
// this IWorkflowRegistry workflowRegistry,
// CancellationToken cancellationToken = default)
// where T : IActivity
// {
// var results = await workflowRegistry.GetWorkflowsByStartActivityAsync(typeof(T).Name, cancellationToken);
// return results.Select(x => (x.Workflow, x.Activity));
// }
// public static async Task<IEnumerable<(IWorkflowBlueprint Workflow, IActivityBlueprint Activity)>> GetWorkflowsByStartActivityAsync(
// this IWorkflowRegistry workflowRegistry,
// string activityType,
// CancellationToken cancellationToken = default)
// {
// var workflows = await workflowRegistry.GetWorkflowsAsync(cancellationToken).ToListAsync(cancellationToken);
//
// var query =
// from workflow in workflows
// where workflow.IsPublished
// from activity in workflow.GetStartActivities()
// where activity.Type == activityType
// select (workflow, activity);
//
// return query.Distinct();
// }
workflowRegistry.GetWorkflowAsync(id, default, versionOptions, cancellationToken);
public static Task<IWorkflowBlueprint?> GetWorkflowAsync(
this IWorkflowRegistry workflowRegistry,
string id,
string? tenantId,
VersionOptions versionOptions,
CancellationToken cancellationToken = default) =>
workflowRegistry.GetAsync(id, tenantId, versionOptions, cancellationToken);
}
}

19
src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj
View file

@ -1,6 +1,6 @@
<Project Sdk="Microsoft.NET.Sdk.Web">
<Import Project="..\..\..\..\configureawait.props" />
<Import Project="..\..\..\..\configureawait.props"/>
<PropertyGroup>
<TargetFramework>net5.0</TargetFramework>
<LangVersion>latest</LangVersion>
@ -9,17 +9,14 @@
</PropertyGroup>
<ItemGroup>
<PackageReference Include="YesSql.Provider.Sqlite" Version="1.0.0-beta-1637" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Telnyx\Elsa.Activities.Telnyx.csproj"/>
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Temporal.Quartz\Elsa.Activities.Temporal.Quartz.csproj"/>
<ProjectReference Include="..\..\..\persistence\Elsa.Persistence.EntityFramework\Elsa.Persistence.EntityFramework.Sqlite\Elsa.Persistence.EntityFramework.Sqlite.csproj"/>
<ProjectReference Include="..\..\..\persistence\Elsa.Persistence.YesSql\Elsa.Persistence.YesSql.csproj"/>
<ProjectReference Include="..\..\..\server\Elsa.Server.Api\Elsa.Server.Api.csproj"/>
<ProjectReference Include="..\..\..\server\Elsa.Server.Orleans\Elsa.Server.Orleans.csproj"/>
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Telnyx\Elsa.Activities.Telnyx.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Temporal.Quartz\Elsa.Activities.Temporal.Quartz.csproj" />
<ProjectReference Include="..\..\..\persistence\Elsa.Persistence.EntityFramework\Elsa.Persistence.EntityFramework.Sqlite\Elsa.Persistence.EntityFramework.Sqlite.csproj" />
<ProjectReference Include="..\..\..\persistence\Elsa.Persistence.YesSql\Elsa.Persistence.YesSql.csproj" />
<ProjectReference Include="..\..\..\server\Elsa.Server.Api\Elsa.Server.Api.csproj" />
</ItemGroup>
<ItemGroup>
<Compile Update=".\Workflows\HelloWorld.*.cs">
<DependentUpon>.\Workflows\HelloWorld.cs</DependentUpon>

16
src/samples/server/Elsa.Samples.Server.Host/Program.cs
View file

@ -1,5 +1,10 @@
using System.Net;
using Elsa.Server.Orleans.Grains.Contracts;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.Hosting;
using Orleans;
using Orleans.Configuration;
using Orleans.Hosting;
namespace Elsa.Samples.Server.Host
{
@ -12,6 +17,15 @@ namespace Elsa.Samples.Server.Host
public static IHostBuilder CreateHostBuilder(string[] args) =>
Microsoft.Extensions.Hosting.Host.CreateDefaultBuilder(args)
.ConfigureWebHostDefaults(webBuilder => { webBuilder.UseStartup<Startup>(); });
.ConfigureWebHostDefaults(webBuilder => webBuilder.UseStartup<Startup>())
.UseOrleans(siloBuilder => siloBuilder
.UseLocalhostClustering()
.Configure<ClusterOptions>(options =>
{
options.ClusterId = "localhost";
options.ServiceId = "elsa-workflows";
})
.ConfigureApplicationParts(parts => parts.AddApplicationPart(typeof(IWorkflowDefinitionGrain).Assembly).WithReferences())
.Configure<EndpointOptions>(options => options.AdvertisedIPAddress = IPAddress.Loopback));
}
}

2
src/samples/server/Elsa.Samples.Server.Host/Startup.cs
View file

@ -1,6 +1,7 @@
using Elsa.Activities.Telnyx.Extensions;
using Elsa.Persistence.EntityFramework.Core.Extensions;
using Elsa.Persistence.EntityFramework.Sqlite;
using Elsa.Server.Orleans.Extensions;
using Microsoft.AspNetCore.Builder;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.Configuration;
@ -27,6 +28,7 @@ namespace Elsa.Samples.Server.Host
services
.AddElsa(elsa => elsa
.UseEntityFrameworkPersistence(ef => ef.UseSqlite())
.UseOrleansDispatchers()
.AddConsoleActivities()
.AddHttpActivities(elsaSection.GetSection("Http").Bind)
.AddEmailActivities(elsaSection.GetSection("Smtp").Bind)

21
src/samples/server/Elsa.Samples.Server.Host/Workflows/HeartbeatWorkflow.cs Normal file
View file

@ -0,0 +1,21 @@
using Elsa.Activities.Console;
using Elsa.Activities.Temporal;
using Elsa.Builders;
using NodaTime;
namespace Elsa.Samples.Server.Host.Workflows
{
public class HeartbeatWorkflow : IWorkflow
{
private readonly IClock _clock;
public HeartbeatWorkflow(IClock clock) => _clock = clock;
public void Build(IWorkflowBuilder builder)
{
builder
.WithDisplayName("Timer")
.Timer(Duration.FromSeconds(10))
.WriteLine(() => $"Heartbeat at {_clock.GetCurrentInstant()}");
}
}
}

37
src/server/Elsa.Server.Orleans/Dispatch/GrainWorkflowDispatcher.cs Normal file
View file

@ -0,0 +1,37 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Server.Orleans.Grains;
using Elsa.Dispatch;
using Elsa.Server.Orleans.Grains.Contracts;
using Orleans;
namespace Elsa.Server.Orleans.Dispatch
{
public class GrainWorkflowDispatcher : IWorkflowDefinitionDispatcher, IWorkflowInstanceDispatcher, ICorrelatingWorkflowDispatcher
{
private readonly IClusterClient _clusterClient;
public GrainWorkflowDispatcher(IClusterClient clusterClient)
{
_clusterClient = clusterClient;
}
public async Task DispatchAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
{
var grain = _clusterClient.GetGrain<IWorkflowDefinitionGrain>(request.WorkflowDefinitionId);
await grain.ExecuteWorkflowAsync(request, cancellationToken);
}
public async Task DispatchAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var grain = _clusterClient.GetGrain<IWorkflowInstanceGrain>(request.WorkflowInstanceId);
await grain.ExecuteWorkflowAsync(request, cancellationToken);
}
public async Task DispatchAsync(ExecuteCorrelatedWorkflowRequest request, CancellationToken cancellationToken = default)
{
var grain = _clusterClient.GetGrain<ICorrelatedWorkflowGrain>(request.CorrelationId);
await grain.ExecutedCorrelatedWorkflowAsync(request, cancellationToken);
}
}
}

23
src/server/Elsa.Server.Orleans/Elsa.Server.Orleans.csproj Normal file
View file

@ -0,0 +1,23 @@
<Project Sdk="Microsoft.NET.Sdk">
<Import Project="..\..\..\common.props" />
<Import Project="..\..\..\configureawait.props" />
<PropertyGroup>
<TargetFramework>net5.0</TargetFramework>
<Description>
Elsa is a set of workflow libraries and tools that enable lean and mean workflowing capabilities in any .NET Core application.
This package provides Orleans grains that handle workflow dispatch.
</Description>
<PackageTags>elsa, workflows, orleans, actor model</PackageTags>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Orleans.Server" Version="3.4.1" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />
</ItemGroup>
</Project>

23
src/server/Elsa.Server.Orleans/Extensions/ServiceCollectionExtensions.cs Normal file
View file

@ -0,0 +1,23 @@
using Elsa.Dispatch;
using Elsa.Server.Orleans.Dispatch;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Server.Orleans.Extensions
{
public static class ServiceCollectionExtensions
{
public static ElsaOptions UseOrleansDispatchers(this ElsaOptions elsaOptions)
{
var services = elsaOptions.Services;
services.AddSingleton<GrainWorkflowDispatcher>();
elsaOptions
.UseCorrelatingWorkflowDispatcher(sp => sp.GetRequiredService<GrainWorkflowDispatcher>())
.UseWorkflowDefinitionDispatcher(sp => sp.GetRequiredService<GrainWorkflowDispatcher>())
.UseWorkflowInstanceDispatcher(sp => sp.GetRequiredService<GrainWorkflowDispatcher>());
return elsaOptions;
}
}
}

3
src/server/Elsa.Server.Orleans/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>

12
src/server/Elsa.Server.Orleans/Grains/Contracts/ICorrelatedWorkflowGrain.cs Normal file
View file

@ -0,0 +1,12 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Orleans;
namespace Elsa.Server.Orleans.Grains.Contracts
{
public interface ICorrelatedWorkflowGrain : IGrainWithStringKey
{
Task ExecutedCorrelatedWorkflowAsync(ExecuteCorrelatedWorkflowRequest request, CancellationToken cancellationToken = default);
}
}

12
src/server/Elsa.Server.Orleans/Grains/Contracts/IWorkflowDefinitionGrain.cs Normal file
View file

@ -0,0 +1,12 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Orleans;
namespace Elsa.Server.Orleans.Grains.Contracts
{
public interface IWorkflowDefinitionGrain : IGrainWithStringKey
{
Task ExecuteWorkflowAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default);
}
}

12
src/server/Elsa.Server.Orleans/Grains/Contracts/IWorkflowInstanceGrain.cs Normal file
View file

@ -0,0 +1,12 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Orleans;
namespace Elsa.Server.Orleans.Grains.Contracts
{
public interface IWorkflowInstanceGrain : IGrainWithStringKey
{
Task ExecuteWorkflowAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default);
}
}

78
src/server/Elsa.Server.Orleans/Grains/CorrelatedWorkflowDefinitionGrain.cs Normal file
View file

@ -0,0 +1,78 @@
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Elsa.Bookmarks;
using Elsa.Dispatch;
using Elsa.Models;
using Elsa.Persistence;
using Elsa.Persistence.Specifications;
using Elsa.Server.Orleans.Grains.Contracts;
using Elsa.Triggers;
using Microsoft.Extensions.Logging;
using Open.Linq.AsyncExtensions;
using Orleans;
namespace Elsa.Server.Orleans.Grains
{
public class CorrelatedWorkflowDefinitionGrain : Grain, ICorrelatedWorkflowGrain
{
private readonly IWorkflowInstanceStore _workflowInstanceStore;
private readonly IBookmarkFinder _bookmarkFinder;
private readonly ITriggerFinder _triggerFinder;
private readonly IWorkflowDefinitionDispatcher _workflowDefinitionDispatcher;
private readonly IWorkflowInstanceDispatcher _workflowInstanceDispatcher;
private readonly ILogger<CorrelatedWorkflowDefinitionGrain> _logger;
public CorrelatedWorkflowDefinitionGrain(
IWorkflowInstanceStore workflowInstanceStore,
IBookmarkFinder bookmarkFinder,
ITriggerFinder triggerFinder,
IWorkflowDefinitionDispatcher workflowDefinitionDispatcher,
IWorkflowInstanceDispatcher workflowInstanceDispatcher,
ILogger<CorrelatedWorkflowDefinitionGrain> logger)
{
_workflowInstanceStore = workflowInstanceStore;
_bookmarkFinder = bookmarkFinder;
_triggerFinder = triggerFinder;
_workflowDefinitionDispatcher = workflowDefinitionDispatcher;
_workflowInstanceDispatcher = workflowInstanceDispatcher;
_logger = logger;
}
public async Task ExecutedCorrelatedWorkflowAsync(ExecuteCorrelatedWorkflowRequest request, CancellationToken cancellationToken = default)
{
var correlationId = request.CorrelationId;
var correlatedWorkflowInstanceCount = await _workflowInstanceStore.CountAsync(new CorrelationIdSpecification<WorkflowInstance>(correlationId), cancellationToken);
if (correlatedWorkflowInstanceCount > 0)
{
_logger.LogDebug("{WorkflowInstanceCount} existing workflows found with correlation ID '{CorrelationId}' will be queued for execution", correlatedWorkflowInstanceCount, correlationId);
var existingWorkflows = await _bookmarkFinder.FindBookmarksAsync(request.ActivityType, request.Bookmark, request.TenantId, cancellationToken).ToList();
await ResumeWorkflowsAsync(existingWorkflows, request.Input, cancellationToken);
}
else
{
_logger.LogDebug("No existing workflows found with correlation ID '{CorrelationId}'. Starting new workflow", correlationId);
await StartWorkflowsAsync(request, cancellationToken);
}
}
private async Task StartWorkflowsAsync(ExecuteCorrelatedWorkflowRequest request, CancellationToken cancellationToken)
{
var filter = request.Trigger;
var triggers = await _triggerFinder.FindTriggersAsync(request.ActivityType, filter, request.TenantId, cancellationToken);
foreach (var trigger in triggers)
{
var workflowBlueprint = trigger.WorkflowBlueprint;
await _workflowDefinitionDispatcher.DispatchAsync(new ExecuteWorkflowDefinitionRequest(workflowBlueprint.Id, trigger.ActivityId, request.Input, request.CorrelationId, request.ContextId, workflowBlueprint.TenantId), cancellationToken);
}
}
private async Task ResumeWorkflowsAsync(IEnumerable<BookmarkFinderResult> results, object? input, CancellationToken cancellationToken)
{
foreach (var result in results)
await _workflowInstanceDispatcher.DispatchAsync(new ExecuteWorkflowInstanceRequest(result.WorkflowInstanceId, result.ActivityId, input), cancellationToken);
}
}
}

38
src/server/Elsa.Server.Orleans/Grains/WorkflowDefinitionGrain.cs Normal file
View file

@ -0,0 +1,38 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Elsa.Models;
using Elsa.Server.Orleans.Grains.Contracts;
using Elsa.Services;
using Microsoft.Extensions.Logging;
using Orleans;
namespace Elsa.Server.Orleans.Grains
{
public class WorkflowDefinitionGrain : Grain, IWorkflowDefinitionGrain
{
private readonly IWorkflowRegistry _workflowRegistry;
private readonly IWorkflowRunner _workflowRunner;
private readonly ILogger<WorkflowDefinitionGrain> _logger;
public WorkflowDefinitionGrain(IWorkflowRegistry workflowRegistry, IWorkflowRunner workflowRunner, ILogger<WorkflowDefinitionGrain> logger)
{
_workflowRegistry = workflowRegistry;
_workflowRunner = workflowRunner;
_logger = logger;
}
public async Task ExecuteWorkflowAsync(ExecuteWorkflowDefinitionRequest request, CancellationToken cancellationToken = default)
{
var workflowBlueprint = await _workflowRegistry.GetWorkflowAsync(request.WorkflowDefinitionId, request.TenantId, VersionOptions.Published, cancellationToken);
if (workflowBlueprint == null)
{
_logger.LogWarning("No published workflow definition {WorkflowDefinitionId} found", request.WorkflowDefinitionId);
return;
}
await _workflowRunner.RunWorkflowAsync(workflowBlueprint, request.ActivityId, request.Input, request.CorrelationId, request.ContextId, cancellationToken);
}
}
}

38
src/server/Elsa.Server.Orleans/Grains/WorkflowInstanceGrain.cs Normal file
View file

@ -0,0 +1,38 @@
using System.Threading;
using System.Threading.Tasks;
using Elsa.Dispatch;
using Elsa.Persistence;
using Elsa.Server.Orleans.Grains.Contracts;
using Elsa.Services;
using Microsoft.Extensions.Logging;
using Orleans;
namespace Elsa.Server.Orleans.Grains
{
public class WorkflowInstanceGrain : Grain, IWorkflowInstanceGrain
{
private readonly IWorkflowInstanceStore _store;
private readonly IWorkflowRunner _workflowRunner;
private readonly ILogger<WorkflowInstanceGrain> _logger;
public WorkflowInstanceGrain(IWorkflowInstanceStore store, IWorkflowRunner workflowRunner, ILogger<WorkflowInstanceGrain> logger)
{
_store = store;
_workflowRunner = workflowRunner;
_logger = logger;
}
public async Task ExecuteWorkflowAsync(ExecuteWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var workflowInstance = await _store.FindByIdAsync(request.WorkflowInstanceId, cancellationToken);
if(workflowInstance == null)
{
_logger.LogWarning("Workflow instance {WorkflowInstanceId} not found", request.WorkflowInstanceId);
return;
}
await _workflowRunner.RunWorkflowAsync(workflowInstance, request.ActivityId, request.Input, cancellationToken);
}
}
}