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

Merge pull request #4881 from elsa-workflows/cleanup(workflow-runtime-api)

Browse source 
Make workflow runtime methods accept nullable options
This commit is contained in:
Sipke Schoorstra 2024-02-05 07:48:55 +01:00 committed by GitHub
parent 06e93c13e2 3a4f690a3e
commit ef1d736184
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 61 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
samples/console/Elsa.Samples.ConsoleApp.HelloWorld/Program.cs
View file

@ -1,4 +1,4 @@
using Elsa.Extensions;
using Elsa.Extensions;
using Elsa.Workflows.Activities;
using Elsa.Workflows.Contracts;
using Microsoft.Extensions.DependencyInjection;

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

@ -67,11 +67,11 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime
}
/// <inheritdoc />
public async Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options)
public async Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default)
{
var versionOptions = options.VersionOptions;
var correlationId = options.CorrelationId;
var input = options.Input;
var versionOptions = options?.VersionOptions;
var correlationId = options?.CorrelationId;
var input = options?.Input;
var workflowInstanceId = _identityGenerator.GenerateId();
var request = new StartWorkflowRequest
@ -81,21 +81,22 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime
VersionOptions = versionOptions.ToString(),
CorrelationId = correlationId.EmptyIfNull(),
Input = input?.SerializeInput(),
Properties = options.Properties?.SerializeProperties(),
TriggerActivityId = options.TriggerActivityId.EmptyIfNull(),
Properties = options?.Properties?.SerializeProperties(),
TriggerActivityId = options?.TriggerActivityId.EmptyIfNull(),
};
var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId);
var response = await client.CanStart(request, options.CancellationTokens.SystemCancellationToken);
var response = await client.CanStart(request, options?.CancellationTokens.SystemCancellationToken ?? default);
return new CanStartWorkflowResult(workflowInstanceId, response!.CanStart);
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options)
public async Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default)
{
// Load the workflow definition.
var workflowDefinition = await _workflowDefinitionService.FindAsync(definitionId, options.VersionOptions, options.CancellationTokens.SystemCancellationToken);
var versionOptions = options?.VersionOptions ?? VersionOptions.Published;
var workflowDefinition = await _workflowDefinitionService.FindAsync(definitionId, versionOptions, options?.CancellationTokens.SystemCancellationToken ?? default);
if (workflowDefinition == null)
return null;
@ -104,12 +105,12 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options)
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default)
{
var versionOptions = options.VersionOptions;
var correlationId = options.CorrelationId;
var workflowInstanceId = options.InstanceId ?? _identityGenerator.GenerateId();
var input = options.Input;
var versionOptions = options?.VersionOptions;
var correlationId = options?.CorrelationId;
var workflowInstanceId = options?.InstanceId ?? _identityGenerator.GenerateId();
var input = options?.Input;
var request = new StartWorkflowRequest
{
@ -118,22 +119,22 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime
VersionOptions = versionOptions.ToString(),
CorrelationId = correlationId.WithDefault(""),
Input = input?.SerializeInput(),
Properties = options.Properties?.SerializeProperties(),
TriggerActivityId = options.TriggerActivityId.WithDefault("")
Properties = options?.Properties?.SerializeProperties(),
TriggerActivityId = options?.TriggerActivityId.WithDefault("")
};
var client = _cluster.GetNamedWorkflowGrain(workflowInstanceId);
var response = await client.Start(request, options.CancellationTokens.SystemCancellationToken);
var response = await client.Start(request, options?.CancellationTokens.SystemCancellationToken ?? default);
return _workflowExecutionResultMapper.Map(response!);
}
/// <inheritdoc />
public async Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions options)
public async Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default)
{
var hash = _hasher.Hash(activityTypeName, bookmarkPayload);
var filter = new TriggerFilter { Hash = hash };
var systemCancellationToken = options.CancellationTokens.SystemCancellationToken;
var systemCancellationToken = options?.CancellationTokens.SystemCancellationToken ?? default;
var triggers = await _triggerStore.FindManyAsync(filter, systemCancellationToken);
var results = new List<WorkflowExecutionResult>();
@ -143,13 +144,13 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime
var startOptions = new StartWorkflowRuntimeOptions
{
CorrelationId = options.CorrelationId,
Input = options.Input,
Properties = options.Properties,
CorrelationId = options?.CorrelationId,
Input = options?.Input,
Properties = options?.Properties,
VersionOptions = VersionOptions.Published,
TriggerActivityId = trigger.ActivityId,
InstanceId = options.WorkflowInstanceId,
CancellationTokens = options.CancellationTokens
InstanceId = options?.WorkflowInstanceId,
CancellationTokens = options?.CancellationTokens ?? default
};
var canStartResult = await CanStartWorkflowAsync(definitionId, startOptions);
@ -166,47 +167,47 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options)
public async Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions? 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(),
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);
var response = await client.Resume(request, options?.CancellationTokens.SystemCancellationToken ?? default);
return _workflowExecutionResultMapper.Map(response!);
}
/// <inheritdoc />
public async Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions options)
public async Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default)
{
var hash = _hasher.Hash(activityTypeName, bookmarkPayload, options.ActivityInstanceId);
var correlationId = options.CorrelationId;
var workflowInstanceId = options.WorkflowInstanceId;
var 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);
var bookmarks = await _bookmarkStore.FindManyAsync(filter, options?.CancellationTokens.SystemCancellationToken ?? default);
return await ResumeWorkflowsAsync(
bookmarks,
new ResumeWorkflowRuntimeOptions
{
CorrelationId = correlationId,
Input = options.Input,
Properties = options.Properties,
CancellationTokens = options.CancellationTokens
Input = options?.Input,
Properties = options?.Properties,
CancellationTokens = options?.CancellationTokens ?? default
}
);
}
/// <inheritdoc />
public async Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions options)
public async Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default)
{
var startedWorkflows = await StartWorkflowsAsync(activityTypeName, bookmarkPayload, options);
var resumedWorkflows = await ResumeWorkflowsAsync(activityTypeName, bookmarkPayload, options);
@ -337,22 +338,6 @@ internal class ProtoActorWorkflowRuntime : IWorkflowRuntime
return resumedWorkflows;
}
private async Task StoreBookmarksAsync(string workflowInstanceId, ICollection<Bookmark> bookmarks, string? correlationId, CancellationToken cancellationToken = default)
{
foreach (var bookmark in bookmarks)
{
var storedBookmark = new StoredBookmark(bookmark.Id, bookmark.Name, bookmark.Hash, workflowInstanceId, bookmark.CreatedAt, bookmark.ActivityInstanceId, correlationId, bookmark.Payload);
await _bookmarkStore.SaveAsync(storedBookmark, cancellationToken);
}
}
private async Task RemoveBookmarksAsync(string workflowInstanceId, IEnumerable<Bookmark> bookmarks, CancellationToken cancellationToken = default)
{
var matchingHashes = bookmarks.Select(x => x.Hash).ToList();
var filter = new BookmarkFilter { Hashes = matchingHashes, WorkflowInstanceId = workflowInstanceId };
await _bookmarkManager.DeleteManyAsync(filter, cancellationToken);
}
private async Task<IEnumerable<WorkflowMatch>> FindStartableWorkflowsAsync(WorkflowsFilter workflowsFilter, CancellationToken cancellationToken)
{
var hash = _hasher.Hash(workflowsFilter.ActivityTypeName, workflowsFilter.BookmarkPayload);

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

@ -16,42 +16,42 @@ public interface IWorkflowRuntime
/// <summary>
/// Returns a value whether or not the specified workflow definition can create a new instance.
/// </summary>
Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options);
Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default);
/// <summary>
/// Creates a new workflow instance of the specified definition ID and executes it.
/// </summary>
/// <param name="definitionId">The workflow definition ID to run.</param>
/// <param name="options">Options for starting the workflow.</param>
Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options);
Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default);
/// <summary>
/// Starts all workflows with triggers matching the specified activity type and bookmark payload.
/// </summary>
/// <returns></returns>
Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions options);
Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default);
/// <summary>
/// Tries to start a workflow and returns the result if successful.
/// </summary>
Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options);
Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default);
/// <summary>
/// Resumes an existing workflow instance.
/// </summary>
/// <param name="workflowInstanceId">The ID of the workflow instance to resume.</param>
/// <param name="options">Options for resuming the workflow.</param>
Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions options);
Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeOptions? options = default);
/// <summary>
/// Resumes all workflows that are bookmarked on the specified activity type.
/// </summary>
Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions options);
Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default);
/// <summary>
/// Starts all workflows and resumes existing workflow instances based on the specified activity type and bookmark payload.
/// </summary>
Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions options);
Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = default);
/// <summary>
/// Executes a pending workflow.

32
src/modules/Elsa.Workflows.Runtime/Services/DefaultWorkflowRuntime.cs
View file

@ -34,7 +34,6 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
private readonly WorkflowStateMapper _workflowStateMapper;
private readonly IIdentityGenerator _identityGenerator;
private readonly IWorkflowExecutionContextStore _workflowExecutionContextStore;
private readonly IWorkflowStateExtractor _workflowStateExtractor;
private readonly IServiceProvider _serviceProvider;
private readonly IBookmarksPersister _bookmarksPersister;
@ -54,7 +53,6 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
WorkflowStateMapper workflowStateMapper,
IIdentityGenerator identityGenerator,
IWorkflowExecutionContextStore workflowExecutionContextStore,
IWorkflowStateExtractor workflowStateExtractor,
IServiceProvider serviceProvider,
IBookmarksPersister bookmarksPersister)
{
@ -70,36 +68,35 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
_workflowStateMapper = workflowStateMapper;
_identityGenerator = identityGenerator;
_workflowExecutionContextStore = workflowExecutionContextStore;
_workflowStateExtractor = workflowStateExtractor;
_serviceProvider = serviceProvider;
_bookmarksPersister = bookmarksPersister;
}
/// <inheritdoc />
public async Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options)
public async Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default)
{
var input = options.Input;
var correlationId = options.CorrelationId;
var workflowHost = await CreateWorkflowHostAsync(definitionId, options, options.CancellationTokens.SystemCancellationToken);
var input = options?.Input;
var correlationId = options?.CorrelationId;
var workflowHost = await CreateWorkflowHostAsync(definitionId, options, options?.CancellationTokens.SystemCancellationToken ?? default);
var startWorkflowOptions = new StartWorkflowHostOptions
{
CorrelationId = correlationId,
Input = input,
TriggerActivityId = options.TriggerActivityId
TriggerActivityId = options?.TriggerActivityId
};
var canStart = await workflowHost.CanStartWorkflowAsync(startWorkflowOptions, options.CancellationTokens.SystemCancellationToken);
var canStart = await workflowHost.CanStartWorkflowAsync(startWorkflowOptions, options?.CancellationTokens.SystemCancellationToken ?? default);
return new CanStartWorkflowResult(null, canStart);
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options)
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default)
{
var workflowHost = await CreateWorkflowHostAsync(definitionId, options, options.CancellationTokens.SystemCancellationToken);
var workflowHost = await CreateWorkflowHostAsync(definitionId, options, options?.CancellationTokens.SystemCancellationToken ?? default);
return await StartWorkflowAsync(workflowHost, options);
}
/// <inheritdoc />
public async Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions options)
public async Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeOptions? options = default)
{
return await StartWorkflowAsync(definitionId, options);
}
@ -399,10 +396,10 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
}
}
private async Task<IWorkflowHost> CreateWorkflowHostAsync(string definitionId, StartWorkflowRuntimeOptions options, CancellationToken cancellationToken)
private async Task<IWorkflowHost> CreateWorkflowHostAsync(string definitionId, StartWorkflowRuntimeOptions? options, CancellationToken cancellationToken)
{
var versionOptions = options.VersionOptions;
var host = await CreateWorkflowHostAsync(definitionId, versionOptions, cancellationToken);
var versionOptions = options?.VersionOptions;
var host = await _workflowHostFactory.CreateAsync(definitionId, versionOptions ?? VersionOptions.Published, cancellationToken);
if (host == null)
throw new Exception("Specified workflow definition and version does not exist");
@ -410,11 +407,6 @@ public class DefaultWorkflowRuntime : IWorkflowRuntime
return host;
}
private Task<IWorkflowHost?> CreateWorkflowHostAsync(string definitionId, VersionOptions versionOptions, CancellationToken cancellationToken)
{
return _workflowHostFactory.CreateAsync(definitionId, versionOptions, cancellationToken);
}
private async Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(IEnumerable<StoredBookmark> bookmarks, ResumeWorkflowRuntimeOptions runtimeOptions)
{
var resumedWorkflows = new List<WorkflowExecutionResult>();