Merge pull request #6344 from elsa-workflows/task/restore-previous-workflow-runtime-api

Restore old workflow runtime API
This commit is contained in:
Sipke Schoorstra 2025-01-28 10:37:46 +01:00 committed by GitHub
commit e2a57aca35
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
32 changed files with 619 additions and 21 deletions

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>

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)
{

View file

@ -18,8 +18,4 @@
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
</ItemGroup>
<ItemGroup>
<Folder Include="Handlers\" />
</ItemGroup>
</Project>

View file

@ -50,6 +50,11 @@ public class DistributedWorkflowClient(
await _localWorkflowClient.ImportStateAsync(workflowState, cancellationToken);
}
public async Task<bool> InstanceExistsAsync(CancellationToken cancellationToken = default)
{
return await _localWorkflowClient.InstanceExistsAsync(cancellationToken);
}
private async Task<R> WithLockAsync<R>(Func<Task<R>> func)
{
var lockKey = $"workflow-instance:{WorkflowInstanceId}";

View file

@ -0,0 +1,31 @@
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.Distributed;
public partial class DistributedWorkflowRuntime
{
private readonly ObsoleteWorkflowRuntime _obsoleteApi;
public Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.CanStartWorkflowAsync(definitionId, options);
public Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.StartWorkflowAsync(definitionId, options);
public Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.StartWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.TryStartWorkflowAsync(definitionId, options);
public Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null) => _obsoleteApi.ResumeWorkflowAsync(workflowInstanceId, options);
public Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<WorkflowExecutionResult> ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) => _obsoleteApi.ExecuteWorkflowAsync(match, options);
public Task<CancellationResult> CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.CancelWorkflowAsync(workflowInstanceId, cancellationToken);
public Task<IEnumerable<WorkflowMatch>> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) => _obsoleteApi.FindWorkflowsAsync(filter, cancellationToken);
public Task<WorkflowState?> ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.ExportWorkflowStateAsync(workflowInstanceId, cancellationToken);
public Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default) => _obsoleteApi.ImportWorkflowStateAsync(workflowState, cancellationToken);
public Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default) => _obsoleteApi.UpdateBookmarkAsync(bookmark, cancellationToken);
public Task<long> CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default) => _obsoleteApi.CountRunningWorkflowsAsync(request, cancellationToken);
}

View file

@ -1,3 +1,4 @@
using Elsa.Workflows.Management;
using Microsoft.Extensions.DependencyInjection;
namespace Elsa.Workflows.Runtime.Distributed;
@ -5,8 +6,21 @@ namespace Elsa.Workflows.Runtime.Distributed;
/// <summary>
/// Represents a distributed workflow runtime that can create <see cref="IWorkflowClient"/> instances connected to a workflow instance.
/// </summary>
public class DistributedWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) : IWorkflowRuntime
public partial class DistributedWorkflowRuntime : IWorkflowRuntime
{
private readonly IServiceProvider _serviceProvider;
private readonly IIdentityGenerator _identityGenerator;
/// <summary>
/// Represents a distributed workflow runtime that can create <see cref="IWorkflowClient"/> instances connected to a workflow instance.
/// </summary>
public DistributedWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator)
{
_serviceProvider = serviceProvider;
_identityGenerator = identityGenerator;
_obsoleteApi = ActivatorUtilities.CreateInstance<ObsoleteWorkflowRuntime>(serviceProvider, (Func<string?, CancellationToken, ValueTask<IWorkflowClient>>)CreateClientAsync);
}
/// <inheritdoc />
public async ValueTask<IWorkflowClient> CreateClientAsync(CancellationToken cancellationToken = default)
{
@ -16,8 +30,8 @@ public class DistributedWorkflowRuntime(IServiceProvider serviceProvider, IIdent
/// <inheritdoc />
public ValueTask<IWorkflowClient> CreateClientAsync(string? workflowInstanceId, CancellationToken cancellationToken = default)
{
workflowInstanceId ??= identityGenerator.GenerateId();
var client = (IWorkflowClient)ActivatorUtilities.CreateInstance(serviceProvider, typeof(DistributedWorkflowClient), workflowInstanceId);
workflowInstanceId ??= _identityGenerator.GenerateId();
var client = (IWorkflowClient)ActivatorUtilities.CreateInstance(_serviceProvider, typeof(DistributedWorkflowClient), workflowInstanceId);
return new(client);
}
}

View file

@ -41,7 +41,7 @@ internal class WorkflowInstance(
public override Task OnStarted()
{
_linkedTokenSource = new CancellationTokenSource();
_linkedTokenSource = new();
_linkedCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(Context.CancellationToken, _linkedTokenSource.Token).Token;
return Task.CompletedTask;
}
@ -61,7 +61,7 @@ internal class WorkflowInstance(
if (result.IsFaulted)
onError(result.Exception.Message);
else
respond(new CreateWorkflowInstanceResponse());
respond(new());
});
return Task.CompletedTask;
@ -170,7 +170,7 @@ internal class WorkflowInstance(
{
await EnsureStateAsync();
var json = mappers.WorkflowStateJsonMapper.Map(WorkflowState);
return new ExportWorkflowStateResponse
return new()
{
SerializedWorkflowState = json
};
@ -186,12 +186,21 @@ internal class WorkflowInstance(
await workflowInstanceManager.SaveAsync(WorkflowState, Context.CancellationToken);
}
public override Task<InstanceExistsResponse> InstanceExists()
{
var exists = _workflowInstanceId != null;
return Task.FromResult(new InstanceExistsResponse
{
Exists = exists
});
}
private async Task<RunWorkflowResult> RunAsync(RunWorkflowOptions runWorkflowOptions)
{
if (_isRunning)
{
_queuedRunWorkflowOptions.Enqueue(runWorkflowOptions);
return new RunWorkflowResult(null!, null!, null);
return new(null!, null!, null);
}
_isRunning = true;

View file

@ -85,3 +85,7 @@ message ExportWorkflowStateResponse {
message ImportWorkflowStateRequest {
Json SerializedWorkflowState = 1;
}
message InstanceExistsResponse {
bool Exists = 1;
}

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 (InstanceExistsResponse);
}

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>();

View file

@ -0,0 +1,32 @@
using Elsa.Workflows.Runtime.Distributed;
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.ProtoActor.Services;
public partial class ProtoActorWorkflowRuntime
{
private readonly ObsoleteWorkflowRuntime _obsoleteApi;
public Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.CanStartWorkflowAsync(definitionId, options);
public Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.StartWorkflowAsync(definitionId, options);
public Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.StartWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.TryStartWorkflowAsync(definitionId, options);
public Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null) => _obsoleteApi.ResumeWorkflowAsync(workflowInstanceId, options);
public Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<WorkflowExecutionResult> ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) => _obsoleteApi.ExecuteWorkflowAsync(match, options);
public Task<CancellationResult> CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.CancelWorkflowAsync(workflowInstanceId, cancellationToken);
public Task<IEnumerable<WorkflowMatch>> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) => _obsoleteApi.FindWorkflowsAsync(filter, cancellationToken);
public Task<WorkflowState?> ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.ExportWorkflowStateAsync(workflowInstanceId, cancellationToken);
public Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default) => _obsoleteApi.ImportWorkflowStateAsync(workflowState, cancellationToken);
public Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default) => _obsoleteApi.UpdateBookmarkAsync(bookmark, cancellationToken);
public Task<long> CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default) => _obsoleteApi.CountRunningWorkflowsAsync(request, cancellationToken);
}

View file

@ -5,8 +5,22 @@ 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 : IWorkflowRuntime
{
private readonly IServiceProvider _serviceProvider;
private readonly IIdentityGenerator _identityGenerator;
/// <summary>
/// Represents a Proto.Actor implementation of the workflows runtime.
/// </summary>
public ProtoActorWorkflowRuntime(IServiceProvider serviceProvider,
IIdentityGenerator identityGenerator)
{
_serviceProvider = serviceProvider;
_identityGenerator = identityGenerator;
_obsoleteApi = ActivatorUtilities.CreateInstance<ObsoleteWorkflowRuntime>(serviceProvider, (Func<string?, CancellationToken, ValueTask<IWorkflowClient>>)CreateClientAsync);
}
/// <inheritdoc />
public async ValueTask<IWorkflowClient> CreateClientAsync(CancellationToken cancellationToken = default)
{
@ -16,8 +30,8 @@ public class ProtoActorWorkflowRuntime(IServiceProvider serviceProvider, IIdenti
/// <inheritdoc />
public ValueTask<IWorkflowClient> CreateClientAsync(string? workflowInstanceId, CancellationToken cancellationToken = default)
{
workflowInstanceId ??= identityGenerator.GenerateId();
var client = (IWorkflowClient)ActivatorUtilities.CreateInstance(serviceProvider, typeof(ProtoActorWorkflowClient), workflowInstanceId);
workflowInstanceId ??= _identityGenerator.GenerateId();
var client = (IWorkflowClient)ActivatorUtilities.CreateInstance(_serviceProvider, typeof(ProtoActorWorkflowClient), workflowInstanceId);
return new(client);
}
}

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);
}

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);
}

View file

@ -0,0 +1,12 @@
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>
[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")]
public record WorkflowsFilter(string ActivityTypeName, object BookmarkPayload, TriggerWorkflowsOptions Options);

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);

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);

View file

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

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>

View file

@ -3,6 +3,7 @@ namespace Elsa.Workflows.Runtime.Options;
/// <summary>
/// Options for triggering workflows.
/// </summary>
[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")]
public class TriggerWorkflowsOptions
{
public string? CorrelationId { get; set; }

View file

@ -5,6 +5,7 @@ namespace Elsa.Workflows.Runtime.Parameters;
/// <summary>
/// Options for resuming workflows.
/// </summary>
[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")]
public class ResumeWorkflowRuntimeParams
{
public string? CorrelationId { get; set; }

View file

@ -5,6 +5,7 @@ namespace Elsa.Workflows.Runtime.Parameters;
/// <summary>
/// Represents parameters for starting a workflow.
/// </summary>
[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")]
public class StartWorkflowRuntimeParams
{
public string? CorrelationId { get; set; }

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; }
}

View file

@ -3,6 +3,7 @@ namespace Elsa.Workflows.Runtime.Params;
/// <summary>
/// Options for delivering a workflow inbox message.
/// </summary>
[Obsolete("This type is obsolete. Use the new IBookmarkQueue service instead.")]
public class WorkflowInboxMessageDeliveryParams
{
/// <summary>

View file

@ -0,0 +1,23 @@
namespace Elsa.Workflows.Runtime.Requests;
/// <summary>
/// Contains arguments to use for counting the number of workflow instances.
/// </summary>
[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")]
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; }
}

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.

View file

@ -5,4 +5,5 @@ namespace Elsa.Workflows.Runtime.Results;
/// </summary>
/// <param name="Success">True if the operation was successful; otherwise, false.</param>
/// <param name="Reason">The reason for the failure, if any.</param>
public record CancellationResult(bool Success, CancellationFailureReason? Reason = default);
[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")]
public record CancellationResult(bool Success, CancellationFailureReason? Reason = null);

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;
}

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;

View file

@ -0,0 +1,31 @@
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;
public partial class LocalWorkflowRuntime
{
private readonly ObsoleteWorkflowRuntime _obsoleteApi;
public Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.CanStartWorkflowAsync(definitionId, options);
public Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.StartWorkflowAsync(definitionId, options);
public Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.StartWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.TryStartWorkflowAsync(definitionId, options);
public Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null) => _obsoleteApi.ResumeWorkflowAsync(workflowInstanceId, options);
public Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options);
public Task<WorkflowExecutionResult> ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) => _obsoleteApi.ExecuteWorkflowAsync(match, options);
public Task<CancellationResult> CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.CancelWorkflowAsync(workflowInstanceId, cancellationToken);
public Task<IEnumerable<WorkflowMatch>> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) => _obsoleteApi.FindWorkflowsAsync(filter, cancellationToken);
public Task<WorkflowState?> ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.ExportWorkflowStateAsync(workflowInstanceId, cancellationToken);
public Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default) => _obsoleteApi.ImportWorkflowStateAsync(workflowState, cancellationToken);
public Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default) => _obsoleteApi.UpdateBookmarkAsync(bookmark, cancellationToken);
public Task<long> CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default) => _obsoleteApi.CountRunningWorkflowsAsync(request, cancellationToken);
}

View file

@ -7,8 +7,23 @@ 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 : IWorkflowRuntime
{
private readonly IServiceProvider _serviceProvider;
private readonly IIdentityGenerator _identityGenerator;
/// <summary>
/// Represents a local implementation of the distributed runtime for running workflows.
/// 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 LocalWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator)
{
_serviceProvider = serviceProvider;
_identityGenerator = identityGenerator;
_obsoleteApi = ActivatorUtilities.CreateInstance<ObsoleteWorkflowRuntime>(serviceProvider, (Func<string?, CancellationToken, ValueTask<IWorkflowClient>>)CreateClientAsync);
}
/// <inheritdoc />
public async ValueTask<IWorkflowClient> CreateClientAsync(CancellationToken cancellationToken = default)
{
@ -18,8 +33,8 @@ public class LocalWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGen
/// <inheritdoc />
public ValueTask<IWorkflowClient> CreateClientAsync(string? workflowInstanceId, CancellationToken cancellationToken = default)
{
workflowInstanceId ??= identityGenerator.GenerateId();
var client = (IWorkflowClient)ActivatorUtilities.CreateInstance(serviceProvider, typeof(LocalWorkflowClient), workflowInstanceId);
workflowInstanceId ??= _identityGenerator.GenerateId();
var client = (IWorkflowClient)ActivatorUtilities.CreateInstance(_serviceProvider, typeof(LocalWorkflowClient), workflowInstanceId);
return new(client);
}
}

View file

@ -0,0 +1,250 @@
using Elsa.Common.Models;
using Elsa.Workflows.Management;
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;
/// <summary>
/// Implements the now deprecated workflow runtime API methods.
/// </summary>
public class ObsoleteWorkflowRuntime(
Func<string?, CancellationToken, ValueTask<IWorkflowClient>> createClientAsync,
IWorkflowDefinitionService workflowDefinitionService,
IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator,
IStimulusSender stimulusSender,
IStimulusHasher stimulusHasher,
IBookmarkStore bookmarkStore,
IWorkflowInstanceStore workflowInstanceStore,
ITriggerBoundWorkflowService triggerBoundWorkflowService,
IBookmarkBoundWorkflowService bookmarkBoundWorkflowService)
{
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(null, canStart);
}
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var client = 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();
}
}