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

Incremental work on restoring previous workflow runtime API surface for backward compatibility

Browse source
This commit is contained in:
Sipke Schoorstra 2025-01-28 00:11:25 +01:00
parent a2efe5a941
commit 8e88de546d
20 changed files with 831 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs
View file

@ -20,7 +20,12 @@ public interface IWorkflowInstanceManager
/// Finds the first workflow instance that matches the specified filter.
/// </summary>
Task<WorkflowInstance?> FindAsync(WorkflowInstanceFilter filter, CancellationToken cancellationToken = default);
/// <summary>
/// Determines whether a workflow instance with the specified ID exists.
/// </summary>
Task<bool> ExistsAsync(string instanceId, CancellationToken cancellationToken = default);
/// <summary>
/// Saves the specified workflow instance.
/// </summary>

10
src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs
View file

@ -33,6 +33,16 @@ public class WorkflowInstanceManager(
return await store.FindAsync(filter, cancellationToken);
}
public async Task<bool> ExistsAsync(string instanceId, CancellationToken cancellationToken = default)
{
var filter = new WorkflowInstanceFilter
{
Id = instanceId
};
var count = await store.CountAsync(filter, cancellationToken);
return count > 0;
}
/// <inheritdoc />
public async Task SaveAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken = default)
{

1
src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto
View file

@ -15,4 +15,5 @@ service WorkflowInstance {
rpc Cancel (Empty) returns (Empty);
rpc ExportState(Empty) returns (ExportWorkflowStateResponse);
rpc ImportState(ImportWorkflowStateRequest) returns (Empty);
rpc InstanceExists(Empty) returns (bool);
}

5
src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs
View file

@ -85,6 +85,11 @@ public class ProtoActorWorkflowClient : IWorkflowClient
await _actorClient.ImportState(request, CreateHeaders(), cancellationToken);
}
public Task<bool> InstanceExistsAsync(CancellationToken cancellationToken = default)
{
throw new NotImplementedException();
}
private IDictionary<string, string> CreateHeaders()
{
var headers = new Dictionary<string, string>();

375
src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs Normal file
View file

@ -0,0 +1,375 @@
using System.Diagnostics.CodeAnalysis;
using Elsa.Common.Models;
using Elsa.Extensions;
using Elsa.Workflows.Management.Filters;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Filters;
using Elsa.Workflows.Runtime.Matches;
using Elsa.Workflows.Runtime.Options;
using Elsa.Workflows.Runtime.Parameters;
using Elsa.Workflows.Runtime.Params;
using Elsa.Workflows.Runtime.ProtoActor.Extensions;
using Elsa.Workflows.Runtime.ProtoActor.ProtoBuf;
using Elsa.Workflows.Runtime.Requests;
using Elsa.Workflows.Runtime.Results;
using Elsa.Workflows.State;
namespace Elsa.Workflows.Runtime.ProtoActor.Services;
public partial class ProtoActorWorkflowRuntime
{
/// <inheritdoc />
public async Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionId, options?.VersionOptions ?? VersionOptions.Published, cancellationToken);
var workflow = workflowGraph!.Workflow;
var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new()
{
Workflow = workflow,
CorrelationId = options?.CorrelationId,
CancellationToken = cancellationToken
});
return new CanStartWorkflowResult(
{
CanStart = canStart,
InstanceId = null
};
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken);
var createRequest = new CreateAndRunWorkflowInstanceRequest
{
Properties = options?.Properties,
CorrelationId = options?.CorrelationId,
Input = options?.Input,
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definitionId, options?.VersionOptions ?? VersionOptions.Published),
ParentId = options?.ParentWorkflowInstanceId,
TriggerActivityId = options?.TriggerActivityId
};
var response = await client.CreateAndRunInstanceAsync(createRequest, cancellationToken);
return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents);
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken);
var createRequest = new Workflows.Runtime.Messages.CreateAndRunWorkflowInstanceRequest
{
Properties = options?.Properties,
CorrelationId = options?.CorrelationId,
Input = options?.Input,
WorkflowDefinitionHandle = Workflows.Models.WorkflowDefinitionHandle.ByDefinitionId(definitionId, options?.VersionOptions ?? VersionOptions.Published),
ParentId = options?.ParentWorkflowInstanceId,
TriggerActivityId = options?.TriggerActivityId
};
var response = await client.CreateAndRunInstanceAsync(createRequest, cancellationToken);
return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents);
}
/// <inheritdoc />
public async Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default)
{
var hash = _hasher.Hash(activityTypeName, bookmarkPayload);
var filter = new TriggerFilter
{
Hash = hash
};
var systemCancellationToken = options?.CancellationTokens.SystemCancellationToken ?? default;
var triggers = await _triggerStore.FindManyAsync(filter, systemCancellationToken);
var results = new List<WorkflowExecutionResult>();
foreach (var trigger in triggers)
{
var definitionId = trigger.WorkflowDefinitionId;
var startOptions = new StartWorkflowRuntimeParams
{
CorrelationId = options?.CorrelationId,
Input = options?.Input,
Properties = options?.Properties,
VersionOptions = VersionOptions.Published,
TriggerActivityId = trigger.ActivityId,
InstanceId = options?.WorkflowInstanceId,
CancellationTokens = options?.CancellationTokens ?? default
};
var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions);
// If we can't start the workflow, don't try it.
if (!canStartResult.CanStart)
continue;
var startResult = await StartWorkflowAsync(definitionId, startOptions);
results.Add(startResult);
}
return results;
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = default)
{
var request = new ResumeWorkflowRequest
{
InstanceId = workflowInstanceId,
CorrelationId = options?.CorrelationId.EmptyIfNull(),
BookmarkId = options?.BookmarkId.EmptyIfNull(),
ActivityId = options?.ActivityId.EmptyIfNull(),
Input = options?.Input?.SerializeInput(),
Properties = options?.Properties?.SerializeProperties(),
};
var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId);
var response = await client.Resume(request, options?.CancellationTokens.SystemCancellationToken ?? default);
return _workflowExecutionResultMapper.Map(response!);
}
/// <inheritdoc />
public async Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default)
{
var hash = _hasher.Hash(activityTypeName, bookmarkPayload, options?.ActivityInstanceId);
var correlationId = options?.CorrelationId;
var workflowInstanceId = options?.WorkflowInstanceId;
var filter = new BookmarkFilter
{
Hash = hash,
CorrelationId = correlationId,
WorkflowInstanceId = workflowInstanceId
};
var bookmarks = await _bookmarkStore.FindManyAsync(filter, options?.CancellationTokens.SystemCancellationToken ?? default);
return await ResumeWorkflowsAsync(
bookmarks,
new ResumeWorkflowRuntimeParams
{
CorrelationId = correlationId,
Input = options?.Input,
Properties = options?.Properties,
CancellationTokens = options?.CancellationTokens ?? default
}
);
}
/// <inheritdoc />
public async Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default)
{
var startedWorkflows = await StartWorkflowsAsync(activityTypeName, bookmarkPayload, options);
var resumedWorkflows = await ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options);
var results = startedWorkflows.Concat(resumedWorkflows).ToList();
return new TriggerWorkflowsResult(results);
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult> ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default)
{
if (match is StartableWorkflowMatch collectedStartableWorkflow)
{
var startOptions = new StartWorkflowRuntimeParams
{
CorrelationId = collectedStartableWorkflow.CorrelationId,
Input = options?.Input,
Properties = options?.Properties,
VersionOptions = VersionOptions.Published,
TriggerActivityId = collectedStartableWorkflow.ActivityId,
InstanceId = collectedStartableWorkflow.WorkflowInstanceId,
CancellationTokens = options?.CancellationTokens ?? default
};
return await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions);
}
var collectedResumableWorkflow = (match as ResumableWorkflowMatch)!;
var runtimeOptions = new ResumeWorkflowRuntimeParams
{
CorrelationId = collectedResumableWorkflow.CorrelationId,
Input = options?.Input,
Properties = options?.Properties,
BookmarkId = collectedResumableWorkflow.BookmarkId,
CancellationTokens = options?.CancellationTokens ?? default
};
var result = await ResumeWorkflowAsync(match.WorkflowInstanceId, runtimeOptions);
return result!;
}
/// <inheritdoc />
public async Task<CancellationResult> CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken)
{
var filter = new WorkflowInstanceFilter
{
Id = workflowInstanceId
};
var instance = await _workflowInstanceStore.FindAsync(filter, cancellationToken);
if (instance is null)
return new CancellationResult(false, FailureReason.NotFound);
var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId);
var result = await client.Cancel(cancellationToken);
return new CancellationResult(result?.Result ?? false);
}
/// <inheritdoc />
public async Task<IEnumerable<WorkflowMatch>> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default)
{
var startableWorkflows = await FindStartableWorkflowsAsync(filter, cancellationToken);
var resumableWorkflows = await FindResumableWorkflowsAsync(filter, cancellationToken);
var results = startableWorkflows.Concat(resumableWorkflows).ToList();
return results;
}
/// <inheritdoc />
[RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.DeserializeAsync(String, CancellationToken)")]
public async Task<WorkflowState?> ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
{
var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId);
var response = await client.ExportState(new ExportWorkflowStateRequest(), cancellationToken);
var json = response!.SerializedWorkflowState.Text;
var workflowState = await _workflowStateSerializer.DeserializeAsync(json, cancellationToken);
return workflowState;
}
/// <inheritdoc />
[RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.SerializeAsync(WorkflowState, CancellationToken)")]
public async Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default)
{
var client = _cluster.GetNamedWorkflowGrain(workflowState.Id);
var json = await _workflowStateSerializer.SerializeAsync(workflowState, cancellationToken);
var request = new ImportWorkflowStateRequest
{
SerializedWorkflowState = new Json
{
Text = json
}
};
await client.ImportState(request, cancellationToken);
}
/// <inheritdoc />
public async Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default)
{
await _bookmarkStore.SaveAsync(bookmark, cancellationToken);
}
/// <inheritdoc />
public async Task<long> CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default)
{
var filter = new WorkflowInstanceFilter
{
DefinitionId = request.DefinitionId,
Version = request.Version,
CorrelationId = request.CorrelationId,
WorkflowStatus = WorkflowStatus.Running
};
return await _workflowInstanceStore.CountAsync(filter, cancellationToken);
}
private async Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(IEnumerable<StoredBookmark> bookmarks, ResumeWorkflowRuntimeParams runtimeParams)
{
var resumedWorkflows = new List<WorkflowExecutionResult>();
foreach (var bookmark in bookmarks)
{
var workflowInstanceId = bookmark.WorkflowInstanceId;
var newRuntimeOptions = new ResumeWorkflowRuntimeParams
{
CorrelationId = runtimeParams.CorrelationId,
Input = runtimeParams.Input,
Properties = runtimeParams.Properties,
BookmarkId = bookmark.BookmarkId,
ActivityId = runtimeParams.ActivityId,
ActivityNodeId = runtimeParams.ActivityNodeId,
ActivityInstanceId = runtimeParams.ActivityInstanceId,
ActivityHash = runtimeParams.ActivityHash,
CancellationTokens = runtimeParams.CancellationTokens
};
var resumeResult = await ResumeWorkflowAsync(workflowInstanceId, newRuntimeOptions);
resumedWorkflows.Add(resumeResult!);
}
return resumedWorkflows;
}
private async Task<IEnumerable<WorkflowMatch>> FindStartableWorkflowsAsync(WorkflowsFilter workflowsFilter, CancellationToken cancellationToken)
{
var hash = _hasher.Hash(workflowsFilter.ActivityTypeName, workflowsFilter.BookmarkPayload);
var filter = new TriggerFilter
{
Hash = hash
};
var triggers = await _triggerStore.FindManyAsync(filter, cancellationToken);
var results = new List<WorkflowMatch>();
foreach (var trigger in triggers)
{
var definitionId = trigger.WorkflowDefinitionId;
var startOptions = new StartWorkflowRuntimeParams
{
CorrelationId = workflowsFilter.Options?.CorrelationId,
Input = workflowsFilter.Options.Input,
Properties = workflowsFilter.Options.Properties,
VersionOptions = VersionOptions.Published,
TriggerActivityId = trigger.ActivityId,
CancellationTokens = cancellationToken
};
var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions);
var workflowGraph = await _workflowDefinitionService.FindWorkflowGraphAsync(trigger.WorkflowDefinitionVersionId, cancellationToken);
if (workflowGraph == null)
{
_logger.LogWarning("Workflow version ID {DefinitionVersionId} not found", trigger.WorkflowDefinitionVersionId);
continue;
}
var workflow = workflowGraph.Workflow;
var createWorkflowInstanceRequest = new CreateWorkflowInstanceRequest
{
Workflow = workflow,
CorrelationId = workflowsFilter.Options.CorrelationId,
WorkflowInstanceId = workflowsFilter.Options?.WorkflowInstanceId,
Input = workflowsFilter.Options?.Input,
Properties = workflowsFilter.Options?.Properties
};
var workflowInstance = _workflowInstanceFactory.CreateWorkflowInstance(createWorkflowInstanceRequest);
if (canStartResult.CanStart)
results.Add(new StartableWorkflowMatch(workflowInstance.Id, workflowInstance, workflowsFilter.Options?.CorrelationId, trigger.ActivityId, definitionId, trigger.Payload));
}
return results;
}
private async Task<IEnumerable<WorkflowMatch>> FindResumableWorkflowsAsync(WorkflowsFilter workflowsFilter, CancellationToken cancellationToken)
{
var hash = _hasher.Hash(workflowsFilter.ActivityTypeName, workflowsFilter.BookmarkPayload);
var correlationId = workflowsFilter.Options.CorrelationId;
var workflowInstanceId = workflowsFilter.Options.WorkflowInstanceId;
var activityInstanceId = workflowsFilter.Options.ActivityInstanceId;
var filter = new BookmarkFilter
{
Hash = hash,
CorrelationId = correlationId,
WorkflowInstanceId = workflowInstanceId,
ActivityInstanceId = activityInstanceId
};
var bookmarks = await _bookmarkStore.FindManyAsync(filter, cancellationToken);
var collectedWorkflows = bookmarks.Select(b => new ResumableWorkflowMatch(b.WorkflowInstanceId, default, correlationId, b.BookmarkId, b.Payload)).ToList();
return collectedWorkflows;
}
}

9
src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs
View file

@ -1,11 +1,18 @@
using Elsa.Workflows.Management;
using Microsoft.Extensions.DependencyInjection;
using Proto.Cluster;
namespace Elsa.Workflows.Runtime.ProtoActor.Services;
/// <summary>
/// Represents a Proto.Actor implementation of the workflows runtime.
/// </summary>
public class ProtoActorWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) : IWorkflowRuntime
public partial class ProtoActorWorkflowRuntime(
IServiceProvider serviceProvider,
IWorkflowDefinitionService workflowDefinitionService,
IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator,
Cluster cluster,
IIdentityGenerator identityGenerator) : IWorkflowRuntime
{
/// <inheritdoc />
public async ValueTask<IWorkflowClient> CreateClientAsync(CancellationToken cancellationToken = default)

2
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs
View file

@ -42,4 +42,6 @@ public interface IWorkflowClient
/// Imports the specified <see cref="WorkflowState"/>.
/// </summary>
Task ImportStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default);
Task<bool> InstanceExistsAsync(CancellationToken cancellationToken = default);
}

107
src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs
View file

@ -1,3 +1,13 @@
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Filters;
using Elsa.Workflows.Runtime.Matches;
using Elsa.Workflows.Runtime.Options;
using Elsa.Workflows.Runtime.Parameters;
using Elsa.Workflows.Runtime.Params;
using Elsa.Workflows.Runtime.Requests;
using Elsa.Workflows.Runtime.Results;
using Elsa.Workflows.State;
namespace Elsa.Workflows.Runtime;
/// <summary>
@ -21,4 +31,101 @@ public interface IWorkflowRuntime
/// <returns>A new <see cref="IWorkflowClient"/> instance.</returns>
/// <remarks>The workflow instance itself doesn't have to exist yet.</remarks>
ValueTask<IWorkflowClient> CreateClientAsync(string? workflowInstanceId, CancellationToken cancellationToken = default);
/// <summary>
/// Returns a value whether the specified workflow definition can create a new instance.
/// </summary>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default);
/// <summary>
/// Creates a new workflow instance of the specified definition ID and executes it.
/// </summary>
/// <param name="definitionId">The workflow definition ID to run.</param>
/// <param name="options">Options for starting the workflow.</param>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default);
/// <summary>
/// Starts all workflows with triggers matching the specified activity type and bookmark payload.
/// </summary>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default);
/// <summary>
/// Tries to start a workflow and returns the result if successful.
/// </summary>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default);
/// <summary>
/// Resumes an existing workflow instance.
/// </summary>
/// <param name="workflowInstanceId">The ID of the workflow instance to resume.</param>
/// <param name="options">Options for resuming the workflow.</param>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = default);
/// <summary>
/// Resumes all workflows that are bookmarked on the specified activity type.
/// </summary>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default);
/// <summary>
/// Starts all workflows and resumes existing workflow instances based on the specified activity type and bookmark payload.
/// </summary>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default);
/// <summary>
/// Executes a pending workflow.
/// </summary>
/// <param name="match">A workflow match to execute.</param>
/// <param name="options">Options for executing the workflow.</param>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<WorkflowExecutionResult> ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default);
/// <summary>
/// Cancels the execution of a workflow.
/// </summary>
/// <param name="workflowInstanceId">The ID of the workflow instance to cancel.</param>
/// <param name="cancellationToken"></param>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<CancellationResult> CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default);
/// <summary>
/// Finds all the workflows that can be started or resumed based on a query model.
/// </summary>
/// <param name="filter"></param>
/// <param name="cancellationToken"></param>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<IEnumerable<WorkflowMatch>> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default);
/// <summary>
/// Exports the <see cref="WorkflowState"/> of the specified workflow instance.
/// </summary>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<WorkflowState?> ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default);
/// <summary>
/// Imports the specified <see cref="WorkflowState"/>.
/// </summary>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default);
/// <summary>
/// Updates the specified bookmark.
/// </summary>
/// <param name="bookmark">The bookmark to update.</param>
/// <param name="cancellationToken">The cancellation token.</param>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default);
/// <summary>
/// Counts the number of workflow instances based on the provided query args.
/// </summary>
[Obsolete("Use the client API instead, retrieved from CreateClientAsync")]
Task<long> CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default);
}

11
src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs Normal file
View file

@ -0,0 +1,11 @@
using Elsa.Workflows.Runtime.Options;
namespace Elsa.Workflows.Runtime.Filters;
/// <summary>
/// A filter for finding workflows to trigger.
/// </summary>
/// <param name="ActivityTypeName">The activity type name to trigger workflows for.</param>
/// <param name="BookmarkPayload">The bookmark payload to trigger workflows for.</param>
/// <param name="Options">The options to use when triggering workflows.</param>
public record WorkflowsFilter(string ActivityTypeName, object BookmarkPayload, TriggerWorkflowsOptions Options);

6
src/modules/Elsa.Workflows.Runtime/Matches/ResumableWorkflowMatch.cs Normal file
View file

@ -0,0 +1,6 @@
using Elsa.Workflows.Management.Entities;
namespace Elsa.Workflows.Runtime.Matches;
public record ResumableWorkflowMatch(string WorkflowInstanceId, string? CorrelationId, string? BookmarkId, object? Payload)
: WorkflowMatch(CorrelationId, Payload);

4
src/modules/Elsa.Workflows.Runtime/Matches/StartableWorkflowMatch.cs Normal file
View file

@ -0,0 +1,4 @@
namespace Elsa.Workflows.Runtime.Matches;
public record StartableWorkflowMatch(string? CorrelationId, string? ActivityId, string? DefinitionId, object? Payload)
: WorkflowMatch(CorrelationId, Payload);

5
src/modules/Elsa.Workflows.Runtime/Matches/WorkflowMatch.cs Normal file
View file

@ -0,0 +1,5 @@
using Elsa.Workflows.Management.Entities;
namespace Elsa.Workflows.Runtime.Matches;
public record WorkflowMatch(string? CorrelationId, object? Payload);

2
src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs
View file

@ -22,6 +22,8 @@ public record RunWorkflowInstanceResponse
/// </summary>
public WorkflowSubStatus SubStatus { get; set; }
public ICollection<Bookmark> Bookmarks { get; set; } = new List<Bookmark>();
/// <summary>
/// Any incidents that occurred during the execution of the workflow instance.
/// </summary>

2
src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs
View file

@ -2,6 +2,7 @@ using Elsa.Workflows.Models;
namespace Elsa.Workflows.Runtime.Params;
[Obsolete("This type is obsolete.")]
public class ExecuteWorkflowParams
{
public string? CorrelationId { get; set; }
@ -11,4 +12,5 @@ public class ExecuteWorkflowParams
public IDictionary<string, object>? Properties { get; set; }
public string? TriggerActivityId { get; set; }
public string? ParentWorkflowInstanceId { get; set; }
public CancellationToken CancellationToken { get; set; }
}

22
src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs Normal file
View file

@ -0,0 +1,22 @@
namespace Elsa.Workflows.Runtime.Requests;
/// <summary>
/// Contains arguments to use for counting the number of workflow instances.
/// </summary>
public class CountRunningWorkflowsRequest
{
/// <summary>
/// The workflow definition ID to include in the query.
/// </summary>
public string? DefinitionId { get; set; }
/// <summary>
/// The workflow definition version to include in the query.
/// </summary>
public int? Version { get; set; }
/// <summary>
/// The correlation ID to include in the query.
/// </summary>
public string? CorrelationId { get; set; }
}

2
src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs
View file

@ -24,6 +24,8 @@ public record StartWorkflowResponse
/// The sub-status of the workflow instance.
/// </summary>
public WorkflowSubStatus? SubStatus { get; set; }
public ICollection<Bookmark> Bookmarks { get; set; } = new List<Bookmark>();
/// <summary>
/// Any incidents that occurred during the execution of the workflow instance.

5
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs
View file

@ -12,7 +12,7 @@ public class DefaultWorkflowStarter(IWorkflowDefinitionService workflowDefinitio
{
var workflow = await GetWorkflowAsync(request, cancellationToken);
var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new WorkflowActivationStrategyEvaluationContext
var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new()
{
Workflow = workflow,
CorrelationId = request.CorrelationId
@ -41,6 +41,7 @@ public class DefaultWorkflowStarter(IWorkflowDefinitionService workflowDefinitio
WorkflowInstanceId = runWorkflowResponse.WorkflowInstanceId,
Status = runWorkflowResponse.Status,
SubStatus = runWorkflowResponse.SubStatus,
Bookmarks = runWorkflowResponse.Bookmarks,
Incidents = runWorkflowResponse.Incidents
};
}
@ -56,7 +57,7 @@ public class DefaultWorkflowStarter(IWorkflowDefinitionService workflowDefinitio
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(request.WorkflowDefinitionHandle, cancellationToken);
if (workflowGraph == null)
throw new WorkflowGraphNotFoundException($"Workflow definition not found.", request.WorkflowDefinitionHandle);
throw new WorkflowGraphNotFoundException("Workflow definition not found.", request.WorkflowDefinitionHandle);
return workflowGraph.Workflow;
}

5
src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs
View file

@ -96,6 +96,11 @@ public class LocalWorkflowClient(
await workflowInstanceManager.SaveAsync(workflowInstance, cancellationToken);
}
public Task<bool> InstanceExistsAsync(CancellationToken cancellationToken = default)
{
return workflowInstanceManager.ExistsAsync(workflowInstanceId, cancellationToken);
}
private async Task<RunWorkflowInstanceResponse> RunInstanceAsync(WorkflowInstance workflowInstance, RunWorkflowInstanceRequest request, CancellationToken cancellationToken = default)
{
var workflowState = workflowInstance.WorkflowState;

241
src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs Normal file
View file

@ -0,0 +1,241 @@
using Elsa.Common.Models;
using Elsa.Workflows.Management.Filters;
using Elsa.Workflows.Models;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Filters;
using Elsa.Workflows.Runtime.Matches;
using Elsa.Workflows.Runtime.Messages;
using Elsa.Workflows.Runtime.Options;
using Elsa.Workflows.Runtime.Parameters;
using Elsa.Workflows.Runtime.Params;
using Elsa.Workflows.Runtime.Requests;
using Elsa.Workflows.Runtime.Results;
using Elsa.Workflows.State;
using Open.Linq.AsyncExtensions;
namespace Elsa.Workflows.Runtime;
public partial class LocalWorkflowRuntime
{
public async Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionId, options?.VersionOptions ?? VersionOptions.Published, cancellationToken);
var workflow = workflowGraph!.Workflow;
var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new()
{
Workflow = workflow,
CorrelationId = options?.CorrelationId,
CancellationToken = cancellationToken
});
return new CanStartWorkflowResult(
{
CanStart = canStart,
InstanceId = null
};
}
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken);
var createRequest = new CreateAndRunWorkflowInstanceRequest
{
Properties = options?.Properties,
CorrelationId = options?.CorrelationId,
Input = options?.Input,
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definitionId, options?.VersionOptions ?? VersionOptions.Published),
ParentId = options?.ParentWorkflowInstanceId,
TriggerActivityId = options?.TriggerActivityId
};
var response = await client.CreateAndRunInstanceAsync(createRequest, cancellationToken);
return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents);
}
public async Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var metadata = new StimulusMetadata
{
CorrelationId = options?.CorrelationId,
WorkflowInstanceId = options?.WorkflowInstanceId,
Properties = options?.Properties,
ActivityInstanceId = options?.ActivityInstanceId,
Input = options?.Input
};
var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken);
var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList();
return results;
}
public async Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null)
{
return await StartWorkflowAsync(definitionId, options);
}
public async Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var workflowClient = await CreateClientAsync(workflowInstanceId, cancellationToken);
var exists = await workflowClient.InstanceExistsAsync(cancellationToken);
if (!exists)
return null;
var runWorkflowRequest = new RunWorkflowInstanceRequest
{
Input = options?.Input,
Properties = options?.Properties,
ActivityHandle = options?.ActivityHandle,
BookmarkId = options?.BookmarkId
};
var response = await workflowClient.RunInstanceAsync(runWorkflowRequest, cancellationToken);
return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents);
}
public async Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var metadata = new StimulusMetadata
{
CorrelationId = options?.CorrelationId,
WorkflowInstanceId = options?.WorkflowInstanceId,
Properties = options?.Properties,
ActivityInstanceId = options?.ActivityInstanceId,
Input = options?.Input
};
var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken);
var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList();
return results;
}
public async Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var metadata = new StimulusMetadata
{
CorrelationId = options?.CorrelationId,
WorkflowInstanceId = options?.WorkflowInstanceId,
Properties = options?.Properties,
ActivityInstanceId = options?.ActivityInstanceId,
Input = options?.Input
};
var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken);
var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList();
return new(results);
}
public async Task<WorkflowExecutionResult> ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
if (match is StartableWorkflowMatch collectedStartableWorkflow)
{
var startOptions = new StartWorkflowRuntimeParams
{
CorrelationId = collectedStartableWorkflow.CorrelationId,
Input = options?.Input,
Properties = options?.Properties,
VersionOptions = VersionOptions.Published,
TriggerActivityId = collectedStartableWorkflow.ActivityId,
CancellationToken = cancellationToken
};
var startResult = await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions);
return startResult with
{
TriggeredActivityId = collectedStartableWorkflow.ActivityId
};
}
var collectedResumableWorkflow = (match as ResumableWorkflowMatch)!;
var runtimeOptions = new ResumeWorkflowRuntimeParams
{
CorrelationId = collectedResumableWorkflow.CorrelationId,
BookmarkId = collectedResumableWorkflow.BookmarkId,
Input = options?.Input,
Properties = options?.Properties,
CancellationToken = cancellationToken,
};
return (await ResumeWorkflowAsync(collectedResumableWorkflow.WorkflowInstanceId, runtimeOptions))!;
}
public async Task<CancellationResult> CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
{
var client = await CreateClientAsync(workflowInstanceId, cancellationToken);
await client.CancelAsync(cancellationToken);
return new(true);
}
public async Task<IEnumerable<WorkflowMatch>> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default)
{
var startableWorkflows = await FindStartableWorkflowsAsync(filter, cancellationToken);
var resumableWorkflows = await FindResumableWorkflowsAsync(filter, cancellationToken);
var results = startableWorkflows.Concat(resumableWorkflows).ToList();
return results;
}
public async Task<WorkflowState?> ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
{
var client = await CreateClientAsync(workflowInstanceId, cancellationToken);
return await client.ExportStateAsync(cancellationToken);
}
public async Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default)
{
var client = await CreateClientAsync(workflowState.Id, cancellationToken);
await client.ImportStateAsync(workflowState, cancellationToken);
}
public async Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default)
{
await bookmarkStore.SaveAsync(bookmark, cancellationToken);
}
public async Task<long> CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default)
{
var filter = new WorkflowInstanceFilter
{
DefinitionId = request.DefinitionId,
Version = request.Version,
CorrelationId = request.CorrelationId,
WorkflowStatus = WorkflowStatus.Running
};
return await workflowInstanceStore.CountAsync(filter, cancellationToken);
}
private async Task<IEnumerable<WorkflowMatch>> FindStartableWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default)
{
var stimulusHash = stimulusHasher.Hash(filter.ActivityTypeName, filter.BookmarkPayload, filter.Options.ActivityInstanceId);
var triggerBoundWorkflows = await triggerBoundWorkflowService.FindManyAsync(stimulusHash, cancellationToken).ToList();
var correlationId = filter.Options.CorrelationId;
var query =
from triggerBoundWorkflow in triggerBoundWorkflows
from trigger in triggerBoundWorkflow.Triggers
select new StartableWorkflowMatch(correlationId, trigger.ActivityId, triggerBoundWorkflow.WorkflowGraph.Workflow.Identity.DefinitionId, filter.BookmarkPayload);
return query.ToList();
}
private async Task<IEnumerable<WorkflowMatch>> FindResumableWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken)
{
var bookmarkOptions = new FindBookmarkOptions
{
CorrelationId = filter.Options.CorrelationId,
WorkflowInstanceId = filter.Options.WorkflowInstanceId,
ActivityInstanceId = filter.Options.ActivityInstanceId
};
var bookmarkBoundWorkflows = await bookmarkBoundWorkflowService.FindManyAsync(filter.ActivityTypeName, filter.BookmarkPayload, bookmarkOptions, cancellationToken).ToList();
return (
from bookmarkBoundWorkflow in bookmarkBoundWorkflows
from bookmark in bookmarkBoundWorkflow.Bookmarks
select new ResumableWorkflowMatch(bookmarkBoundWorkflow.WorkflowInstanceId, bookmark.CorrelationId, bookmark.Id, bookmark.Payload))
.ToList();
}
}

15
src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs
View file

@ -1,3 +1,4 @@
using Elsa.Workflows.Management;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Workflows.Runtime;
@ -7,7 +8,19 @@ namespace Elsa.Workflows.Runtime;
/// It does not support clustering and is intended for single-node deployments only.
/// For distributed deployments, use Proto.Actor or another distributed runtime.
/// </summary>
public class LocalWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) : IWorkflowRuntime
public partial class LocalWorkflowRuntime(
IServiceProvider serviceProvider,
IIdentityGenerator identityGenerator,
IWorkflowDefinitionService workflowDefinitionService,
IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator,
IStimulusSender stimulusSender,
IBookmarkResumer bookmarkResumer,
IStimulusHasher stimulusHasher,
IWorkflowCanceler workflowCanceler,
IBookmarkStore bookmarkStore,
IWorkflowInstanceStore workflowInstanceStore,
ITriggerBoundWorkflowService triggerBoundWorkflowService,
IBookmarkBoundWorkflowService bookmarkBoundWorkflowService) : IWorkflowRuntime
{
/// <inheritdoc />
public async ValueTask<IWorkflowClient> CreateClientAsync(CancellationToken cancellationToken = default)