From 8e88de546d0235c263018e3f3aede5a818564acc Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 28 Jan 2025 00:11:25 +0100 Subject: [PATCH 1/6] Incremental work on restoring previous workflow runtime API surface for backward compatibility --- .../Contracts/IWorkflowInstanceManager.cs | 7 +- .../Services/WorkflowInstanceManager.cs | 10 + .../Proto/WorkflowInstance.proto | 1 + .../Services/ProtoActorWorkflowClient.cs | 5 + .../ProtoActorWorkflowRuntime.Obsolete.cs | 375 ++++++++++++++++++ .../Services/ProtoActorWorkflowRuntime.cs | 9 +- .../Contracts/IWorkflowClient.cs | 2 + .../Contracts/IWorkflowRuntime.cs | 107 +++++ .../Filters/WorkflowsFilter.cs | 11 + .../Matches/ResumableWorkflowMatch.cs | 6 + .../Matches/StartableWorkflowMatch.cs | 4 + .../Matches/WorkflowMatch.cs | 5 + .../Messages/RunWorkflowInstanceResponse.cs | 2 + .../Params/ExecuteWorkflowParams.cs | 2 + .../Requests/CountRunningWorkflowsRequest.cs | 22 + .../Responses/StartWorkflowResponse.cs | 2 + .../Services/DefaultWorkflowStarter.cs | 5 +- .../Services/LocalWorkflowClient.cs | 5 + .../Services/LocalWorkflowRuntime.Obsolete.cs | 241 +++++++++++ .../Services/LocalWorkflowRuntime.cs | 15 +- 20 files changed, 831 insertions(+), 5 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Matches/ResumableWorkflowMatch.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Matches/StartableWorkflowMatch.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Matches/WorkflowMatch.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs create mode 100644 src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs diff --git a/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs b/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs index 2ae45c492..886160157 100644 --- a/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs +++ b/src/modules/Elsa.Workflows.Management/Contracts/IWorkflowInstanceManager.cs @@ -20,7 +20,12 @@ public interface IWorkflowInstanceManager /// Finds the first workflow instance that matches the specified filter. /// Task FindAsync(WorkflowInstanceFilter filter, CancellationToken cancellationToken = default); - + + /// + /// Determines whether a workflow instance with the specified ID exists. + /// + Task ExistsAsync(string instanceId, CancellationToken cancellationToken = default); + /// /// Saves the specified workflow instance. /// diff --git a/src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs b/src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs index 12bc8ca29..f33e43a6d 100644 --- a/src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs +++ b/src/modules/Elsa.Workflows.Management/Services/WorkflowInstanceManager.cs @@ -33,6 +33,16 @@ public class WorkflowInstanceManager( return await store.FindAsync(filter, cancellationToken); } + public async Task ExistsAsync(string instanceId, CancellationToken cancellationToken = default) + { + var filter = new WorkflowInstanceFilter + { + Id = instanceId + }; + var count = await store.CountAsync(filter, cancellationToken); + return count > 0; + } + /// public async Task SaveAsync(WorkflowInstance workflowInstance, CancellationToken cancellationToken = default) { diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto index a23ca9ab7..47088c1d2 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto @@ -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); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs index 7576d136f..a37cb11b3 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowClient.cs @@ -85,6 +85,11 @@ public class ProtoActorWorkflowClient : IWorkflowClient await _actorClient.ImportState(request, CreateHeaders(), cancellationToken); } + public Task InstanceExistsAsync(CancellationToken cancellationToken = default) + { + throw new NotImplementedException(); + } + private IDictionary CreateHeaders() { var headers = new Dictionary(); diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs new file mode 100644 index 000000000..629322c83 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs @@ -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 +{ + /// + public async Task 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 + }; + } + + /// + public async Task 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); + } + + /// + public async Task 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); + } + + /// + public async Task> 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(); + + 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; + } + + /// + public async Task 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!); + } + + /// + public async Task> 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 + } + ); + } + + /// + public async Task 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); + } + + /// + public async Task 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!; + } + + /// + public async Task 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); + } + + /// + public async Task> 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; + } + + /// + [RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.DeserializeAsync(String, CancellationToken)")] + public async Task 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; + } + + /// + [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); + } + + /// + public async Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default) + { + await _bookmarkStore.SaveAsync(bookmark, cancellationToken); + } + + /// + public async Task 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> ResumeWorkflowsAsync(IEnumerable bookmarks, ResumeWorkflowRuntimeParams runtimeParams) + { + var resumedWorkflows = new List(); + + 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> 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(); + + 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> 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; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index 8ba6c1bd8..ce1266342 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -1,11 +1,18 @@ +using Elsa.Workflows.Management; using Microsoft.Extensions.DependencyInjection; +using Proto.Cluster; namespace Elsa.Workflows.Runtime.ProtoActor.Services; /// /// Represents a Proto.Actor implementation of the workflows runtime. /// -public class ProtoActorWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) : IWorkflowRuntime +public partial class ProtoActorWorkflowRuntime( + IServiceProvider serviceProvider, + IWorkflowDefinitionService workflowDefinitionService, + IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, + Cluster cluster, + IIdentityGenerator identityGenerator) : IWorkflowRuntime { /// public async ValueTask CreateClientAsync(CancellationToken cancellationToken = default) diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs index aecfafdd3..8753a76bd 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowClient.cs @@ -42,4 +42,6 @@ public interface IWorkflowClient /// Imports the specified . /// Task ImportStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default); + + Task InstanceExistsAsync(CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs index cbef32bcc..5ef752118 100644 --- a/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Contracts/IWorkflowRuntime.cs @@ -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; /// @@ -21,4 +31,101 @@ public interface IWorkflowRuntime /// A new instance. /// The workflow instance itself doesn't have to exist yet. ValueTask CreateClientAsync(string? workflowInstanceId, CancellationToken cancellationToken = default); + + + /// + /// Returns a value whether the specified workflow definition can create a new instance. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default); + + /// + /// Creates a new workflow instance of the specified definition ID and executes it. + /// + /// The workflow definition ID to run. + /// Options for starting the workflow. + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default); + + /// + /// Starts all workflows with triggers matching the specified activity type and bookmark payload. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default); + + /// + /// Tries to start a workflow and returns the result if successful. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default); + + /// + /// Resumes an existing workflow instance. + /// + /// The ID of the workflow instance to resume. + /// Options for resuming the workflow. + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = default); + + /// + /// Resumes all workflows that are bookmarked on the specified activity type. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default); + + /// + /// Starts all workflows and resumes existing workflow instances based on the specified activity type and bookmark payload. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default); + + /// + /// Executes a pending workflow. + /// + /// A workflow match to execute. + /// Options for executing the workflow. + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default); + + /// + /// Cancels the execution of a workflow. + /// + /// The ID of the workflow instance to cancel. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default); + + /// + /// Finds all the workflows that can be started or resumed based on a query model. + /// + /// + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default); + + /// + /// Exports the of the specified workflow instance. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default); + + /// + /// Imports the specified . + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default); + + /// + /// Updates the specified bookmark. + /// + /// The bookmark to update. + /// The cancellation token. + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default); + + /// + /// Counts the number of workflow instances based on the provided query args. + /// + [Obsolete("Use the client API instead, retrieved from CreateClientAsync")] + Task CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs b/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs new file mode 100644 index 000000000..438446c5a --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs @@ -0,0 +1,11 @@ +using Elsa.Workflows.Runtime.Options; + +namespace Elsa.Workflows.Runtime.Filters; + +/// +/// A filter for finding workflows to trigger. +/// +/// The activity type name to trigger workflows for. +/// The bookmark payload to trigger workflows for. +/// The options to use when triggering workflows. +public record WorkflowsFilter(string ActivityTypeName, object BookmarkPayload, TriggerWorkflowsOptions Options); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Matches/ResumableWorkflowMatch.cs b/src/modules/Elsa.Workflows.Runtime/Matches/ResumableWorkflowMatch.cs new file mode 100644 index 000000000..ba6f7669b --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Matches/ResumableWorkflowMatch.cs @@ -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); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Matches/StartableWorkflowMatch.cs b/src/modules/Elsa.Workflows.Runtime/Matches/StartableWorkflowMatch.cs new file mode 100644 index 000000000..3103b8c52 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Matches/StartableWorkflowMatch.cs @@ -0,0 +1,4 @@ +namespace Elsa.Workflows.Runtime.Matches; + +public record StartableWorkflowMatch(string? CorrelationId, string? ActivityId, string? DefinitionId, object? Payload) + : WorkflowMatch(CorrelationId, Payload); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Matches/WorkflowMatch.cs b/src/modules/Elsa.Workflows.Runtime/Matches/WorkflowMatch.cs new file mode 100644 index 000000000..541a691c9 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Matches/WorkflowMatch.cs @@ -0,0 +1,5 @@ +using Elsa.Workflows.Management.Entities; + +namespace Elsa.Workflows.Runtime.Matches; + +public record WorkflowMatch(string? CorrelationId, object? Payload); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs b/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs index 73d1a9596..8a49e2273 100644 --- a/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs +++ b/src/modules/Elsa.Workflows.Runtime/Messages/RunWorkflowInstanceResponse.cs @@ -22,6 +22,8 @@ public record RunWorkflowInstanceResponse /// public WorkflowSubStatus SubStatus { get; set; } + public ICollection Bookmarks { get; set; } = new List(); + /// /// Any incidents that occurred during the execution of the workflow instance. /// diff --git a/src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs b/src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs index e6c72ad91..f9d4ab40c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs +++ b/src/modules/Elsa.Workflows.Runtime/Params/ExecuteWorkflowParams.cs @@ -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? Properties { get; set; } public string? TriggerActivityId { get; set; } public string? ParentWorkflowInstanceId { get; set; } + public CancellationToken CancellationToken { get; set; } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs b/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs new file mode 100644 index 000000000..9fc36c54f --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs @@ -0,0 +1,22 @@ +namespace Elsa.Workflows.Runtime.Requests; + +/// +/// Contains arguments to use for counting the number of workflow instances. +/// +public class CountRunningWorkflowsRequest +{ + /// + /// The workflow definition ID to include in the query. + /// + public string? DefinitionId { get; set; } + + /// + /// The workflow definition version to include in the query. + /// + public int? Version { get; set; } + + /// + /// The correlation ID to include in the query. + /// + public string? CorrelationId { get; set; } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs b/src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs index e3472732f..6745e54f7 100644 --- a/src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs +++ b/src/modules/Elsa.Workflows.Runtime/Responses/StartWorkflowResponse.cs @@ -24,6 +24,8 @@ public record StartWorkflowResponse /// The sub-status of the workflow instance. /// public WorkflowSubStatus? SubStatus { get; set; } + + public ICollection Bookmarks { get; set; } = new List(); /// /// Any incidents that occurred during the execution of the workflow instance. diff --git a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs index e1c2e79f7..3778ee265 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowStarter.cs @@ -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; } diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs index 6b73d2906..8497c99dc 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowClient.cs @@ -96,6 +96,11 @@ public class LocalWorkflowClient( await workflowInstanceManager.SaveAsync(workflowInstance, cancellationToken); } + public Task InstanceExistsAsync(CancellationToken cancellationToken = default) + { + return workflowInstanceManager.ExistsAsync(workflowInstanceId, cancellationToken); + } + private async Task RunInstanceAsync(WorkflowInstance workflowInstance, RunWorkflowInstanceRequest request, CancellationToken cancellationToken = default) { var workflowState = workflowInstance.WorkflowState; diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs new file mode 100644 index 000000000..fcd4bb7a4 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs @@ -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 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 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> 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 TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) + { + return await StartWorkflowAsync(definitionId, options); + } + + public async Task 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> 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 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 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 CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + var client = await CreateClientAsync(workflowInstanceId, cancellationToken); + await client.CancelAsync(cancellationToken); + return new(true); + } + + public async Task> 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 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 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> 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> 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(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs index dcfa181c7..cfd5309cf 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs @@ -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. /// -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 { /// public async ValueTask CreateClientAsync(CancellationToken cancellationToken = default) From db1556fd0b35fe283321e91f063c0486c29e9e9b Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 28 Jan 2025 00:21:49 +0100 Subject: [PATCH 2/6] Update ProtoActorWorkflowRuntime to implement backwards-compatible workflow runtime API --- .../ProtoActorWorkflowRuntime.Obsolete.cs | 327 ++++++------------ .../Services/ProtoActorWorkflowRuntime.cs | 8 +- 2 files changed, 116 insertions(+), 219 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs index 629322c83..054afc08b 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs @@ -1,6 +1,5 @@ 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; @@ -8,18 +7,17 @@ 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; +using Open.Linq.AsyncExtensions; namespace Elsa.Workflows.Runtime.ProtoActor.Services; public partial class ProtoActorWorkflowRuntime { /// - public async Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default) + public async Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) { var cancellationToken = options?.CancellationToken ?? CancellationToken.None; var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionId, options?.VersionOptions ?? VersionOptions.Published, cancellationToken); @@ -40,29 +38,11 @@ public partial class ProtoActorWorkflowRuntime } /// - public async Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = default) + public async Task TryStartWorkflowAsync(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 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 + var createRequest = new Messages.CreateAndRunWorkflowInstanceRequest { Properties = options?.Properties, CorrelationId = options?.CorrelationId, @@ -76,103 +56,101 @@ public partial class ProtoActorWorkflowRuntime } /// - public async Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default) + public async Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) { - var hash = _hasher.Hash(activityTypeName, bookmarkPayload); - var filter = new TriggerFilter + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken); + var createRequest = new Messages.CreateAndRunWorkflowInstanceRequest { - Hash = hash + 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 systemCancellationToken = options?.CancellationTokens.SystemCancellationToken ?? default; - var triggers = await _triggerStore.FindManyAsync(filter, systemCancellationToken); - var results = new List(); + var response = await client.CreateAndRunInstanceAsync(createRequest, cancellationToken); + return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents); + } - foreach (var trigger in triggers) + /// + public async Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) + { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var metadata = new StimulusMetadata { - 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); - } - + 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 ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = default) + public async Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null) { - var request = new ResumeWorkflowRequest + 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 Messages.RunWorkflowInstanceRequest { - InstanceId = workflowInstanceId, - CorrelationId = options?.CorrelationId.EmptyIfNull(), - BookmarkId = options?.BookmarkId.EmptyIfNull(), - ActivityId = options?.ActivityId.EmptyIfNull(), - Input = options?.Input?.SerializeInput(), - Properties = options?.Properties?.SerializeProperties(), + Input = options?.Input, + Properties = options?.Properties, + ActivityHandle = options?.ActivityHandle, + BookmarkId = options?.BookmarkId }; - var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId); - var response = await client.Resume(request, options?.CancellationTokens.SystemCancellationToken ?? default); + var response = await workflowClient.RunInstanceAsync(runWorkflowRequest, cancellationToken); - return _workflowExecutionResultMapper.Map(response!); + return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents); } /// - public async Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default) + public async Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) { - var hash = _hasher.Hash(activityTypeName, bookmarkPayload, options?.ActivityInstanceId); - var correlationId = options?.CorrelationId; - var workflowInstanceId = options?.WorkflowInstanceId; - var filter = new BookmarkFilter + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; + var metadata = new StimulusMetadata { - Hash = hash, - CorrelationId = correlationId, - WorkflowInstanceId = workflowInstanceId + CorrelationId = options?.CorrelationId, + WorkflowInstanceId = options?.WorkflowInstanceId, + Properties = options?.Properties, + ActivityInstanceId = options?.ActivityInstanceId, + Input = options?.Input }; - 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 - } - ); + 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 TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default) + public async Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) { - var startedWorkflows = await StartWorkflowsAsync(activityTypeName, bookmarkPayload, options); - var resumedWorkflows = await ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options); - var results = startedWorkflows.Concat(resumedWorkflows).ToList(); - - return new TriggerWorkflowsResult(results); + 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 ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) + public async Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = null) { + var cancellationToken = options?.CancellationToken ?? CancellationToken.None; if (match is StartableWorkflowMatch collectedStartableWorkflow) { var startOptions = new StartWorkflowRuntimeParams @@ -182,41 +160,35 @@ public partial class ProtoActorWorkflowRuntime Properties = options?.Properties, VersionOptions = VersionOptions.Published, TriggerActivityId = collectedStartableWorkflow.ActivityId, - InstanceId = collectedStartableWorkflow.WorkflowInstanceId, - CancellationTokens = options?.CancellationTokens ?? default + CancellationToken = cancellationToken + }; + + var startResult = await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions); + return startResult with + { + TriggeredActivityId = collectedStartableWorkflow.ActivityId }; - return await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions); } var collectedResumableWorkflow = (match as ResumableWorkflowMatch)!; var runtimeOptions = new ResumeWorkflowRuntimeParams { CorrelationId = collectedResumableWorkflow.CorrelationId, + BookmarkId = collectedResumableWorkflow.BookmarkId, Input = options?.Input, Properties = options?.Properties, - BookmarkId = collectedResumableWorkflow.BookmarkId, - CancellationTokens = options?.CancellationTokens ?? default + CancellationToken = cancellationToken, }; - var result = await ResumeWorkflowAsync(match.WorkflowInstanceId, runtimeOptions); - return result!; + return (await ResumeWorkflowAsync(collectedResumableWorkflow.WorkflowInstanceId, runtimeOptions))!; } /// public async Task 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); + var client = await CreateClientAsync(workflowInstanceId, cancellationToken); + await client.CancelAsync(cancellationToken); + return new(true); } /// @@ -232,35 +204,22 @@ public partial class ProtoActorWorkflowRuntime [RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.DeserializeAsync(String, CancellationToken)")] public async Task 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; + var client = await CreateClientAsync(workflowInstanceId, cancellationToken); + return await client.ExportStateAsync(cancellationToken); } /// [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); + 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); + await bookmarkStore.SaveAsync(bookmark, cancellationToken); } /// @@ -273,103 +232,37 @@ public partial class ProtoActorWorkflowRuntime CorrelationId = request.CorrelationId, WorkflowStatus = WorkflowStatus.Running }; - return await _workflowInstanceStore.CountAsync(filter, cancellationToken); + return await workflowInstanceStore.CountAsync(filter, cancellationToken); } - private async Task> ResumeWorkflowsAsync(IEnumerable bookmarks, ResumeWorkflowRuntimeParams runtimeParams) + private async Task> FindStartableWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) { - var resumedWorkflows = new List(); + var stimulusHash = stimulusHasher.Hash(filter.ActivityTypeName, filter.BookmarkPayload, filter.Options.ActivityInstanceId); + var triggerBoundWorkflows = await triggerBoundWorkflowService.FindManyAsync(stimulusHash, cancellationToken).ToList(); + var correlationId = filter.Options.CorrelationId; - foreach (var bookmark in bookmarks) - { - var workflowInstanceId = bookmark.WorkflowInstanceId; + var query = + from triggerBoundWorkflow in triggerBoundWorkflows + from trigger in triggerBoundWorkflow.Triggers + select new StartableWorkflowMatch(correlationId, trigger.ActivityId, triggerBoundWorkflow.WorkflowGraph.Workflow.Identity.DefinitionId, filter.BookmarkPayload); - 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; + return query.ToList(); } - private async Task> FindStartableWorkflowsAsync(WorkflowsFilter workflowsFilter, CancellationToken cancellationToken) + private async Task> FindResumableWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken) { - var hash = _hasher.Hash(workflowsFilter.ActivityTypeName, workflowsFilter.BookmarkPayload); - var filter = new TriggerFilter + var bookmarkOptions = new FindBookmarkOptions { - Hash = hash + CorrelationId = filter.Options.CorrelationId, + WorkflowInstanceId = filter.Options.WorkflowInstanceId, + ActivityInstanceId = filter.Options.ActivityInstanceId }; - var triggers = await _triggerStore.FindManyAsync(filter, cancellationToken); - var results = new List(); + var bookmarkBoundWorkflows = await bookmarkBoundWorkflowService.FindManyAsync(filter.ActivityTypeName, filter.BookmarkPayload, bookmarkOptions, cancellationToken).ToList(); - 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> 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; + return ( + from bookmarkBoundWorkflow in bookmarkBoundWorkflows + from bookmark in bookmarkBoundWorkflow.Bookmarks + select new ResumableWorkflowMatch(bookmarkBoundWorkflow.WorkflowInstanceId, bookmark.CorrelationId, bookmark.Id, bookmark.Payload)) + .ToList(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index ce1266342..335afd1f4 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -1,6 +1,5 @@ using Elsa.Workflows.Management; using Microsoft.Extensions.DependencyInjection; -using Proto.Cluster; namespace Elsa.Workflows.Runtime.ProtoActor.Services; @@ -11,7 +10,12 @@ public partial class ProtoActorWorkflowRuntime( IServiceProvider serviceProvider, IWorkflowDefinitionService workflowDefinitionService, IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, - Cluster cluster, + IStimulusSender stimulusSender, + IStimulusHasher stimulusHasher, + ITriggerBoundWorkflowService triggerBoundWorkflowService, + IBookmarkBoundWorkflowService bookmarkBoundWorkflowService, + IBookmarkStore bookmarkStore, + IWorkflowInstanceStore workflowInstanceStore, IIdentityGenerator identityGenerator) : IWorkflowRuntime { /// From 60323d05227caedea19c7be07f46329b0c4268e3 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 28 Jan 2025 00:22:20 +0100 Subject: [PATCH 3/6] Refactor `CanStartWorkflowResult` to use compact constructor. Simplified the object creation in `CanStartWorkflowResult` by using a concise constructor format. This improves code readability and reduces verbosity without altering functionality. --- .../Services/LocalWorkflowRuntime.Obsolete.cs | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs index fcd4bb7a4..1076ed713 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs @@ -30,11 +30,7 @@ public partial class LocalWorkflowRuntime CancellationToken = cancellationToken }); - return new CanStartWorkflowResult( - { - CanStart = canStart, - InstanceId = null - }; + return new(null, canStart); } public async Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) From e27101032b651003b8613c14b1b8ad21d791367f Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 28 Jan 2025 00:32:43 +0100 Subject: [PATCH 4/6] Incremental work on central obsolete workflow runtime for reuse --- .../Elsa.Workflows.Runtime.Distributed.csproj | 4 - .../Services/DistributedWorkflowClient.cs | 5 + .../DistributedWorkflowRuntime.Obsolete.cs | 237 +++++++++++++++++ .../Services/DistributedWorkflowRuntime.cs | 13 +- .../Services/ObsoleteWorkflowRuntime.cs | 247 ++++++++++++++++++ .../Proto/WorkflowInstance.Messages.proto | 4 + .../Proto/WorkflowInstance.proto | 2 +- .../ProtoActorWorkflowRuntime.Obsolete.cs | 6 +- .../Services/LocalWorkflowRuntime.Obsolete.cs | 2 +- .../Services/LocalWorkflowRuntime.cs | 2 - 10 files changed, 508 insertions(+), 14 deletions(-) create mode 100644 src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs create mode 100644 src/modules/Elsa.Workflows.Runtime.Distributed/Services/ObsoleteWorkflowRuntime.cs diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj b/src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj index 8f3983f11..870cd0d78 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj @@ -18,8 +18,4 @@ - - - - diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs index 55179d545..a9468d11d 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowClient.cs @@ -50,6 +50,11 @@ public class DistributedWorkflowClient( await _localWorkflowClient.ImportStateAsync(workflowState, cancellationToken); } + public async Task InstanceExistsAsync(CancellationToken cancellationToken = default) + { + return await _localWorkflowClient.InstanceExistsAsync(cancellationToken); + } + private async Task WithLockAsync(Func> func) { var lockKey = $"workflow-instance:{WorkflowInstanceId}"; diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs new file mode 100644 index 000000000..a28f34048 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs @@ -0,0 +1,237 @@ +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.Distributed; + +public partial class DistributedWorkflowRuntime +{ + public async Task 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 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> 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 TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) + { + return await StartWorkflowAsync(definitionId, options); + } + + public async Task 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> 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 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 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 CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + var client = await CreateClientAsync(workflowInstanceId, cancellationToken); + await client.CancelAsync(cancellationToken); + return new(true); + } + + public async Task> 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 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 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> 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> 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(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs index 8069d0587..8285f2d69 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs @@ -1,3 +1,4 @@ +using Elsa.Workflows.Management; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime.Distributed; @@ -5,7 +6,17 @@ namespace Elsa.Workflows.Runtime.Distributed; /// /// Represents a distributed workflow runtime that can create instances connected to a workflow instance. /// -public class DistributedWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) : IWorkflowRuntime +public partial class DistributedWorkflowRuntime( + IServiceProvider serviceProvider, + IIdentityGenerator identityGenerator, + IWorkflowDefinitionService workflowDefinitionService, + IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, + IStimulusSender stimulusSender, + IStimulusHasher stimulusHasher, + IBookmarkStore bookmarkStore, + IWorkflowInstanceStore workflowInstanceStore, + ITriggerBoundWorkflowService triggerBoundWorkflowService, + IBookmarkBoundWorkflowService bookmarkBoundWorkflowService) : IWorkflowRuntime { /// public async ValueTask CreateClientAsync(CancellationToken cancellationToken = default) diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/ObsoleteWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/ObsoleteWorkflowRuntime.cs new file mode 100644 index 000000000..2173a9319 --- /dev/null +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/ObsoleteWorkflowRuntime.cs @@ -0,0 +1,247 @@ +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.Distributed; + +public class ObsoleteWorkflowRuntime( + Func> createClientAsync, + IWorkflowDefinitionService workflowDefinitionService, + IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, + IStimulusSender stimulusSender, + IStimulusHasher stimulusHasher, + IBookmarkStore bookmarkStore, + IWorkflowInstanceStore workflowInstanceStore, + ITriggerBoundWorkflowService triggerBoundWorkflowService, + IBookmarkBoundWorkflowService bookmarkBoundWorkflowService) +{ + public async Task 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 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> 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 TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) + { + return await StartWorkflowAsync(definitionId, options); + } + + public async Task 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> 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 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 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 CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) + { + var client = await createClientAsync(workflowInstanceId, cancellationToken); + await client.CancelAsync(cancellationToken); + return new(true); + } + + public async Task> 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 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 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> 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> 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(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.Messages.proto b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.Messages.proto index 5eeaa3699..719572c28 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.Messages.proto +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.Messages.proto @@ -85,3 +85,7 @@ message ExportWorkflowStateResponse { message ImportWorkflowStateRequest { Json SerializedWorkflowState = 1; } + +message InstanceExistsResponse { + bool Exists = 1; +} \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto index 47088c1d2..ded6d1d68 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Proto/WorkflowInstance.proto @@ -15,5 +15,5 @@ service WorkflowInstance { rpc Cancel (Empty) returns (Empty); rpc ExportState(Empty) returns (ExportWorkflowStateResponse); rpc ImportState(ImportWorkflowStateRequest) returns (Empty); - rpc InstanceExists(Empty) returns (bool); + rpc InstanceExists(Empty) returns (InstanceExistsResponse); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs index 054afc08b..f6ec2297a 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs @@ -30,11 +30,7 @@ public partial class ProtoActorWorkflowRuntime CancellationToken = cancellationToken }); - return new CanStartWorkflowResult( - { - CanStart = canStart, - InstanceId = null - }; + return new(null, canStart); } /// diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs index 1076ed713..66c5bfb16 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs @@ -36,7 +36,7 @@ public partial class LocalWorkflowRuntime public async Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) { var cancellationToken = options?.CancellationToken ?? CancellationToken.None; - var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken); + var client = await CreateClientAsync(options?.InstanceId, cancellationToken); var createRequest = new CreateAndRunWorkflowInstanceRequest { Properties = options?.Properties, diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs index cfd5309cf..fba019f0f 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs @@ -14,9 +14,7 @@ public partial class LocalWorkflowRuntime( IWorkflowDefinitionService workflowDefinitionService, IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, IStimulusSender stimulusSender, - IBookmarkResumer bookmarkResumer, IStimulusHasher stimulusHasher, - IWorkflowCanceler workflowCanceler, IBookmarkStore bookmarkStore, IWorkflowInstanceStore workflowInstanceStore, ITriggerBoundWorkflowService triggerBoundWorkflowService, From ee883dbd0720d74fd08237e68ddbea39d4c76351 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 28 Jan 2025 09:18:24 +0100 Subject: [PATCH 5/6] Refactor workflow runtime to use ObsoleteWorkflowRuntime delegation Replaced direct method implementations in workflow runtimes with delegations to the new `ObsoleteWorkflowRuntime` class, simplifying the codebase. This consolidates logic and aligns the runtimes under a unified deprecated API layer. --- .../DistributedWorkflowRuntime.Obsolete.cs | 237 ++-------------- .../Services/DistributedWorkflowRuntime.cs | 30 +- .../Actors/WorkflowInstance.cs | 17 +- .../ProtoActorWorkflowRuntime.Obsolete.cs | 265 ++---------------- .../Services/ProtoActorWorkflowRuntime.cs | 32 ++- .../Deprecated}/ObsoleteWorkflowRuntime.cs | 4 +- .../Services/LocalWorkflowRuntime.Obsolete.cs | 237 ++-------------- .../Services/LocalWorkflowRuntime.cs | 33 ++- 8 files changed, 118 insertions(+), 737 deletions(-) rename src/modules/{Elsa.Workflows.Runtime.Distributed/Services => Elsa.Workflows.Runtime/Deprecated}/ObsoleteWorkflowRuntime.cs (98%) diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs index a28f34048..45c9fb00c 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs @@ -1,237 +1,32 @@ -using Elsa.Common.Models; -using Elsa.Workflows.Management.Filters; -using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime.Deprecated; 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.Distributed; public partial class DistributedWorkflowRuntime { - public async Task 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; + private readonly ObsoleteWorkflowRuntime _obsoleteApi; - var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new() - { - Workflow = workflow, - CorrelationId = options?.CorrelationId, - CancellationToken = cancellationToken - }); - - return new(null, canStart); - } - - public async Task 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> 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 TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) - { - return await StartWorkflowAsync(definitionId, options); - } - - public async Task 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> 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 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 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 CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) - { - var client = await CreateClientAsync(workflowInstanceId, cancellationToken); - await client.CancelAsync(cancellationToken); - return new(true); - } - - public async Task> 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 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 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> 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> 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(); - } + public Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.CanStartWorkflowAsync(definitionId, options); + public Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.StartWorkflowAsync(definitionId, options); + public Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.StartWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.TryStartWorkflowAsync(definitionId, options); + public Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null) => _obsoleteApi.ResumeWorkflowAsync(workflowInstanceId, options); + public Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) => _obsoleteApi.ExecuteWorkflowAsync(match, options); + public Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.CancelWorkflowAsync(workflowInstanceId, cancellationToken); + public Task> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) => _obsoleteApi.FindWorkflowsAsync(filter, cancellationToken); + public Task 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 CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default) => _obsoleteApi.CountRunningWorkflowsAsync(request, cancellationToken); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs index 8285f2d69..4b0a6056b 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs @@ -1,4 +1,5 @@ using Elsa.Workflows.Management; +using Elsa.Workflows.Runtime.Deprecated; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime.Distributed; @@ -6,18 +7,21 @@ namespace Elsa.Workflows.Runtime.Distributed; /// /// Represents a distributed workflow runtime that can create instances connected to a workflow instance. /// -public partial class DistributedWorkflowRuntime( - IServiceProvider serviceProvider, - IIdentityGenerator identityGenerator, - IWorkflowDefinitionService workflowDefinitionService, - IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, - IStimulusSender stimulusSender, - IStimulusHasher stimulusHasher, - IBookmarkStore bookmarkStore, - IWorkflowInstanceStore workflowInstanceStore, - ITriggerBoundWorkflowService triggerBoundWorkflowService, - IBookmarkBoundWorkflowService bookmarkBoundWorkflowService) : IWorkflowRuntime +public partial class DistributedWorkflowRuntime : IWorkflowRuntime { + private readonly IServiceProvider _serviceProvider; + private readonly IIdentityGenerator _identityGenerator; + + /// + /// Represents a distributed workflow runtime that can create instances connected to a workflow instance. + /// + public DistributedWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) + { + _serviceProvider = serviceProvider; + _identityGenerator = identityGenerator; + _obsoleteApi = ActivatorUtilities.CreateInstance(serviceProvider, (Func>)CreateClientAsync); + } + /// public async ValueTask CreateClientAsync(CancellationToken cancellationToken = default) { @@ -27,8 +31,8 @@ public partial class DistributedWorkflowRuntime( /// public ValueTask 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); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Actors/WorkflowInstance.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Actors/WorkflowInstance.cs index 370b77e09..748197690 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Actors/WorkflowInstance.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Actors/WorkflowInstance.cs @@ -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 InstanceExists() + { + var exists = _workflowInstanceId != null; + return Task.FromResult(new InstanceExistsResponse + { + Exists = exists + }); + } + private async Task RunAsync(RunWorkflowOptions runWorkflowOptions) { if (_isRunning) { _queuedRunWorkflowOptions.Enqueue(runWorkflowOptions); - return new RunWorkflowResult(null!, null!, null); + return new(null!, null!, null); } _isRunning = true; diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs index f6ec2297a..6393c5149 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs @@ -1,6 +1,5 @@ -using System.Diagnostics.CodeAnalysis; -using Elsa.Common.Models; -using Elsa.Workflows.Management.Filters; +using Elsa.Workflows.Runtime.Deprecated; +using Elsa.Workflows.Runtime.Distributed; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Filters; using Elsa.Workflows.Runtime.Matches; @@ -10,255 +9,25 @@ 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.ProtoActor.Services; public partial class ProtoActorWorkflowRuntime { - /// - public async Task 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; + private readonly ObsoleteWorkflowRuntime _obsoleteApi; - var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new() - { - Workflow = workflow, - CorrelationId = options?.CorrelationId, - CancellationToken = cancellationToken - }); - - return new(null, canStart); - } - - /// - public async Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) - { - var cancellationToken = options?.CancellationToken ?? CancellationToken.None; - var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken); - var createRequest = new 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); - } - - /// - public async Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) - { - var cancellationToken = options?.CancellationToken ?? CancellationToken.None; - var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken); - var createRequest = new 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); - } - - /// - public async Task> 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 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 Messages.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> 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 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 ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = null) - { - 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 CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken) - { - var client = await CreateClientAsync(workflowInstanceId, cancellationToken); - await client.CancelAsync(cancellationToken); - return new(true); - } - - /// - public async Task> 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; - } - - /// - [RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.DeserializeAsync(String, CancellationToken)")] - public async Task ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default) - { - var client = await CreateClientAsync(workflowInstanceId, cancellationToken); - return await client.ExportStateAsync(cancellationToken); - } - - /// - [RequiresUnreferencedCode("Calls Elsa.Workflows.Contracts.IWorkflowStateSerializer.SerializeAsync(WorkflowState, 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 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> 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> 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(); - } + public Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.CanStartWorkflowAsync(definitionId, options); + public Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.StartWorkflowAsync(definitionId, options); + public Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.StartWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.TryStartWorkflowAsync(definitionId, options); + public Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null) => _obsoleteApi.ResumeWorkflowAsync(workflowInstanceId, options); + public Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) => _obsoleteApi.ExecuteWorkflowAsync(match, options); + public Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.CancelWorkflowAsync(workflowInstanceId, cancellationToken); + public Task> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) => _obsoleteApi.FindWorkflowsAsync(filter, cancellationToken); + public Task 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 CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default) => _obsoleteApi.CountRunningWorkflowsAsync(request, cancellationToken); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index 335afd1f4..98122f13b 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -1,4 +1,4 @@ -using Elsa.Workflows.Management; +using Elsa.Workflows.Runtime.Deprecated; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime.ProtoActor.Services; @@ -6,18 +6,22 @@ namespace Elsa.Workflows.Runtime.ProtoActor.Services; /// /// Represents a Proto.Actor implementation of the workflows runtime. /// -public partial class ProtoActorWorkflowRuntime( - IServiceProvider serviceProvider, - IWorkflowDefinitionService workflowDefinitionService, - IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, - IStimulusSender stimulusSender, - IStimulusHasher stimulusHasher, - ITriggerBoundWorkflowService triggerBoundWorkflowService, - IBookmarkBoundWorkflowService bookmarkBoundWorkflowService, - IBookmarkStore bookmarkStore, - IWorkflowInstanceStore workflowInstanceStore, - IIdentityGenerator identityGenerator) : IWorkflowRuntime +public partial class ProtoActorWorkflowRuntime : IWorkflowRuntime { + private readonly IServiceProvider _serviceProvider; + private readonly IIdentityGenerator _identityGenerator; + + /// + /// Represents a Proto.Actor implementation of the workflows runtime. + /// + public ProtoActorWorkflowRuntime(IServiceProvider serviceProvider, + IIdentityGenerator identityGenerator) + { + _serviceProvider = serviceProvider; + _identityGenerator = identityGenerator; + _obsoleteApi = ActivatorUtilities.CreateInstance(serviceProvider, (Func>)CreateClientAsync); + } + /// public async ValueTask CreateClientAsync(CancellationToken cancellationToken = default) { @@ -27,8 +31,8 @@ public partial class ProtoActorWorkflowRuntime( /// public ValueTask 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); } } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/ObsoleteWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Deprecated/ObsoleteWorkflowRuntime.cs similarity index 98% rename from src/modules/Elsa.Workflows.Runtime.Distributed/Services/ObsoleteWorkflowRuntime.cs rename to src/modules/Elsa.Workflows.Runtime/Deprecated/ObsoleteWorkflowRuntime.cs index 2173a9319..28087e8bc 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/ObsoleteWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Deprecated/ObsoleteWorkflowRuntime.cs @@ -14,10 +14,10 @@ using Elsa.Workflows.Runtime.Results; using Elsa.Workflows.State; using Open.Linq.AsyncExtensions; -namespace Elsa.Workflows.Runtime.Distributed; +namespace Elsa.Workflows.Runtime.Deprecated; public class ObsoleteWorkflowRuntime( - Func> createClientAsync, + Func> createClientAsync, IWorkflowDefinitionService workflowDefinitionService, IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, IStimulusSender stimulusSender, diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs index 66c5bfb16..df5be09f7 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs @@ -1,237 +1,32 @@ -using Elsa.Common.Models; -using Elsa.Workflows.Management.Filters; -using Elsa.Workflows.Models; +using Elsa.Workflows.Runtime.Deprecated; 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 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; + private readonly ObsoleteWorkflowRuntime _obsoleteApi; - var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new() - { - Workflow = workflow, - CorrelationId = options?.CorrelationId, - CancellationToken = cancellationToken - }); - - return new(null, canStart); - } - - public async Task 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> 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 TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) - { - return await StartWorkflowAsync(definitionId, options); - } - - public async Task 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> 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 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 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 CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) - { - var client = await CreateClientAsync(workflowInstanceId, cancellationToken); - await client.CancelAsync(cancellationToken); - return new(true); - } - - public async Task> 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 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 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> 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> 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(); - } + public Task CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.CanStartWorkflowAsync(definitionId, options); + public Task StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.StartWorkflowAsync(definitionId, options); + public Task> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.StartWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null) => _obsoleteApi.TryStartWorkflowAsync(definitionId, options); + public Task ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null) => _obsoleteApi.ResumeWorkflowAsync(workflowInstanceId, options); + public Task> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null) => _obsoleteApi.TriggerWorkflowsAsync(activityTypeName, bookmarkPayload, options); + public Task ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default) => _obsoleteApi.ExecuteWorkflowAsync(match, options); + public Task CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default) => _obsoleteApi.CancelWorkflowAsync(workflowInstanceId, cancellationToken); + public Task> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default) => _obsoleteApi.FindWorkflowsAsync(filter, cancellationToken); + public Task 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 CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default) => _obsoleteApi.CountRunningWorkflowsAsync(request, cancellationToken); } \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs index fba019f0f..3f1414d58 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs @@ -1,4 +1,4 @@ -using Elsa.Workflows.Management; +using Elsa.Workflows.Runtime.Deprecated; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime; @@ -8,18 +8,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. /// -public partial class LocalWorkflowRuntime( - IServiceProvider serviceProvider, - IIdentityGenerator identityGenerator, - IWorkflowDefinitionService workflowDefinitionService, - IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator, - IStimulusSender stimulusSender, - IStimulusHasher stimulusHasher, - IBookmarkStore bookmarkStore, - IWorkflowInstanceStore workflowInstanceStore, - ITriggerBoundWorkflowService triggerBoundWorkflowService, - IBookmarkBoundWorkflowService bookmarkBoundWorkflowService) : IWorkflowRuntime +public partial class LocalWorkflowRuntime : IWorkflowRuntime { + private readonly IServiceProvider _serviceProvider; + private readonly IIdentityGenerator _identityGenerator; + + /// + /// 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. + /// + public LocalWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) + { + _serviceProvider = serviceProvider; + _identityGenerator = identityGenerator; + _obsoleteApi = ActivatorUtilities.CreateInstance(serviceProvider, (Func>)CreateClientAsync); + } + /// public async ValueTask CreateClientAsync(CancellationToken cancellationToken = default) { @@ -29,8 +34,8 @@ public partial class LocalWorkflowRuntime( /// public ValueTask 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); } } \ No newline at end of file From 9f01e148a96c2d7d594a4808fa78fb57a05bc026 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 28 Jan 2025 09:26:46 +0100 Subject: [PATCH 6/6] Deprecate and clean up legacy workflow runtime components. Marked various types as obsolete, advising migration to `CreateClientAsync` methods or `IBookmarkQueue` services. Removed unused `Deprecated` namespace imports, aligning code with updated runtime standards. --- .../Services/DistributedWorkflowRuntime.Obsolete.cs | 1 - .../Services/DistributedWorkflowRuntime.cs | 1 - .../Services/ProtoActorWorkflowRuntime.Obsolete.cs | 1 - .../Services/ProtoActorWorkflowRuntime.cs | 1 - .../Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs | 1 + .../Options/TriggerWorkflowsOptions.cs | 1 + .../Parameters/ResumeWorkflowRuntimeParams.cs | 1 + .../Parameters/StartWorkflowRuntimeParams.cs | 1 + .../Params/WorkflowInboxMessageDeliveryParams.cs | 1 + .../Requests/CountRunningWorkflowsRequest.cs | 1 + .../Elsa.Workflows.Runtime/Results/CancellationResult.cs | 3 ++- .../Services/LocalWorkflowRuntime.Obsolete.cs | 1 - .../Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs | 1 - .../{Deprecated => Services}/ObsoleteWorkflowRuntime.cs | 5 ++++- 14 files changed, 12 insertions(+), 8 deletions(-) rename src/modules/Elsa.Workflows.Runtime/{Deprecated => Services}/ObsoleteWorkflowRuntime.cs (98%) diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs index 45c9fb00c..03860d555 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs @@ -1,4 +1,3 @@ -using Elsa.Workflows.Runtime.Deprecated; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Filters; using Elsa.Workflows.Runtime.Matches; diff --git a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs index 4b0a6056b..9ce03bf52 100644 --- a/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs @@ -1,5 +1,4 @@ using Elsa.Workflows.Management; -using Elsa.Workflows.Runtime.Deprecated; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime.Distributed; diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs index 6393c5149..9da1fe8a3 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.Obsolete.cs @@ -1,4 +1,3 @@ -using Elsa.Workflows.Runtime.Deprecated; using Elsa.Workflows.Runtime.Distributed; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Filters; diff --git a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs index 98122f13b..b66d97ee8 100644 --- a/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime.ProtoActor/Services/ProtoActorWorkflowRuntime.cs @@ -1,4 +1,3 @@ -using Elsa.Workflows.Runtime.Deprecated; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime.ProtoActor.Services; diff --git a/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs b/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs index 438446c5a..a6492d081 100644 --- a/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs +++ b/src/modules/Elsa.Workflows.Runtime/Filters/WorkflowsFilter.cs @@ -8,4 +8,5 @@ namespace Elsa.Workflows.Runtime.Filters; /// The activity type name to trigger workflows for. /// The bookmark payload to trigger workflows for. /// The options to use when triggering workflows. +[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")] public record WorkflowsFilter(string ActivityTypeName, object BookmarkPayload, TriggerWorkflowsOptions Options); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Options/TriggerWorkflowsOptions.cs b/src/modules/Elsa.Workflows.Runtime/Options/TriggerWorkflowsOptions.cs index 5ac0f1837..15a784823 100644 --- a/src/modules/Elsa.Workflows.Runtime/Options/TriggerWorkflowsOptions.cs +++ b/src/modules/Elsa.Workflows.Runtime/Options/TriggerWorkflowsOptions.cs @@ -3,6 +3,7 @@ namespace Elsa.Workflows.Runtime.Options; /// /// Options for triggering workflows. /// +[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")] public class TriggerWorkflowsOptions { public string? CorrelationId { get; set; } diff --git a/src/modules/Elsa.Workflows.Runtime/Parameters/ResumeWorkflowRuntimeParams.cs b/src/modules/Elsa.Workflows.Runtime/Parameters/ResumeWorkflowRuntimeParams.cs index 6836fcc0b..b1b744071 100644 --- a/src/modules/Elsa.Workflows.Runtime/Parameters/ResumeWorkflowRuntimeParams.cs +++ b/src/modules/Elsa.Workflows.Runtime/Parameters/ResumeWorkflowRuntimeParams.cs @@ -5,6 +5,7 @@ namespace Elsa.Workflows.Runtime.Parameters; /// /// Options for resuming workflows. /// +[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")] public class ResumeWorkflowRuntimeParams { public string? CorrelationId { get; set; } diff --git a/src/modules/Elsa.Workflows.Runtime/Parameters/StartWorkflowRuntimeParams.cs b/src/modules/Elsa.Workflows.Runtime/Parameters/StartWorkflowRuntimeParams.cs index d8e337c73..79a1dd72d 100644 --- a/src/modules/Elsa.Workflows.Runtime/Parameters/StartWorkflowRuntimeParams.cs +++ b/src/modules/Elsa.Workflows.Runtime/Parameters/StartWorkflowRuntimeParams.cs @@ -5,6 +5,7 @@ namespace Elsa.Workflows.Runtime.Parameters; /// /// Represents parameters for starting a workflow. /// +[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")] public class StartWorkflowRuntimeParams { public string? CorrelationId { get; set; } diff --git a/src/modules/Elsa.Workflows.Runtime/Params/WorkflowInboxMessageDeliveryParams.cs b/src/modules/Elsa.Workflows.Runtime/Params/WorkflowInboxMessageDeliveryParams.cs index 5f800b381..b275ac64d 100644 --- a/src/modules/Elsa.Workflows.Runtime/Params/WorkflowInboxMessageDeliveryParams.cs +++ b/src/modules/Elsa.Workflows.Runtime/Params/WorkflowInboxMessageDeliveryParams.cs @@ -3,6 +3,7 @@ namespace Elsa.Workflows.Runtime.Params; /// /// Options for delivering a workflow inbox message. /// +[Obsolete("This type is obsolete. Use the new IBookmarkQueue service instead.")] public class WorkflowInboxMessageDeliveryParams { /// diff --git a/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs b/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs index 9fc36c54f..3d4cec8ac 100644 --- a/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs +++ b/src/modules/Elsa.Workflows.Runtime/Requests/CountRunningWorkflowsRequest.cs @@ -3,6 +3,7 @@ namespace Elsa.Workflows.Runtime.Requests; /// /// Contains arguments to use for counting the number of workflow instances. /// +[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")] public class CountRunningWorkflowsRequest { /// diff --git a/src/modules/Elsa.Workflows.Runtime/Results/CancellationResult.cs b/src/modules/Elsa.Workflows.Runtime/Results/CancellationResult.cs index 86092adb7..f40bf1a0b 100644 --- a/src/modules/Elsa.Workflows.Runtime/Results/CancellationResult.cs +++ b/src/modules/Elsa.Workflows.Runtime/Results/CancellationResult.cs @@ -5,4 +5,5 @@ namespace Elsa.Workflows.Runtime.Results; /// /// True if the operation was successful; otherwise, false. /// The reason for the failure, if any. -public record CancellationResult(bool Success, CancellationFailureReason? Reason = default); \ No newline at end of file +[Obsolete("This type is obsolete. Use the new CreateClientAsync methods of IWorkflowRuntime instead.")] +public record CancellationResult(bool Success, CancellationFailureReason? Reason = null); \ No newline at end of file diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs index df5be09f7..7d9cfae0c 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.Obsolete.cs @@ -1,4 +1,3 @@ -using Elsa.Workflows.Runtime.Deprecated; using Elsa.Workflows.Runtime.Entities; using Elsa.Workflows.Runtime.Filters; using Elsa.Workflows.Runtime.Matches; diff --git a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs index 3f1414d58..b5f028fde 100644 --- a/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/LocalWorkflowRuntime.cs @@ -1,4 +1,3 @@ -using Elsa.Workflows.Runtime.Deprecated; using Microsoft.Extensions.DependencyInjection; namespace Elsa.Workflows.Runtime; diff --git a/src/modules/Elsa.Workflows.Runtime/Deprecated/ObsoleteWorkflowRuntime.cs b/src/modules/Elsa.Workflows.Runtime/Services/ObsoleteWorkflowRuntime.cs similarity index 98% rename from src/modules/Elsa.Workflows.Runtime/Deprecated/ObsoleteWorkflowRuntime.cs rename to src/modules/Elsa.Workflows.Runtime/Services/ObsoleteWorkflowRuntime.cs index 28087e8bc..9eb5c5af3 100644 --- a/src/modules/Elsa.Workflows.Runtime/Deprecated/ObsoleteWorkflowRuntime.cs +++ b/src/modules/Elsa.Workflows.Runtime/Services/ObsoleteWorkflowRuntime.cs @@ -14,8 +14,11 @@ using Elsa.Workflows.Runtime.Results; using Elsa.Workflows.State; using Open.Linq.AsyncExtensions; -namespace Elsa.Workflows.Runtime.Deprecated; +namespace Elsa.Workflows.Runtime; +/// +/// Implements the now deprecated workflow runtime API methods. +/// public class ObsoleteWorkflowRuntime( Func> createClientAsync, IWorkflowDefinitionService workflowDefinitionService,