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

Incremental work on central obsolete workflow runtime for reuse

Browse source
This commit is contained in:
Sipke Schoorstra 2025-01-28 00:32:43 +01:00
parent 60323d0522
commit e27101032b
10 changed files with 508 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
src/modules/Elsa.Workflows.Runtime.Distributed/Elsa.Workflows.Runtime.Distributed.csproj
View file

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

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

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

237
src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.Obsolete.cs Normal file
View file

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

13
src/modules/Elsa.Workflows.Runtime.Distributed/Services/DistributedWorkflowRuntime.cs
View file

@ -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;
/// <summary>
/// Represents a distributed workflow runtime that can create <see cref="IWorkflowClient"/> instances connected to a workflow instance.
/// </summary>
public class DistributedWorkflowRuntime(IServiceProvider serviceProvider, IIdentityGenerator identityGenerator) : IWorkflowRuntime
public partial class DistributedWorkflowRuntime(
IServiceProvider serviceProvider,
IIdentityGenerator identityGenerator,
IWorkflowDefinitionService workflowDefinitionService,
IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator,
IStimulusSender stimulusSender,
IStimulusHasher stimulusHasher,
IBookmarkStore bookmarkStore,
IWorkflowInstanceStore workflowInstanceStore,
ITriggerBoundWorkflowService triggerBoundWorkflowService,
IBookmarkBoundWorkflowService bookmarkBoundWorkflowService) : IWorkflowRuntime
{
/// <inheritdoc />
public async ValueTask<IWorkflowClient> CreateClientAsync(CancellationToken cancellationToken = default)

247
src/modules/Elsa.Workflows.Runtime.Distributed/Services/ObsoleteWorkflowRuntime.cs Normal file
View file

@ -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<string?, CancellationToken, Task<IWorkflowClient>> createClientAsync,
IWorkflowDefinitionService workflowDefinitionService,
IWorkflowActivationStrategyEvaluator workflowActivationStrategyEvaluator,
IStimulusSender stimulusSender,
IStimulusHasher stimulusHasher,
IBookmarkStore bookmarkStore,
IWorkflowInstanceStore workflowInstanceStore,
ITriggerBoundWorkflowService triggerBoundWorkflowService,
IBookmarkBoundWorkflowService bookmarkBoundWorkflowService)
{
public async Task<CanStartWorkflowResult> CanStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var workflowGraph = await workflowDefinitionService.FindWorkflowGraphAsync(definitionId, options?.VersionOptions ?? VersionOptions.Published, cancellationToken);
var workflow = workflowGraph!.Workflow;
var canStart = await workflowActivationStrategyEvaluator.CanStartWorkflowAsync(new()
{
Workflow = workflow,
CorrelationId = options?.CorrelationId,
CancellationToken = cancellationToken
});
return new(null, canStart);
}
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var client = await createClientAsync(options?.InstanceId, cancellationToken);
var createRequest = new CreateAndRunWorkflowInstanceRequest
{
Properties = options?.Properties,
CorrelationId = options?.CorrelationId,
Input = options?.Input,
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definitionId, options?.VersionOptions ?? VersionOptions.Published),
ParentId = options?.ParentWorkflowInstanceId,
TriggerActivityId = options?.TriggerActivityId
};
var response = await client.CreateAndRunInstanceAsync(createRequest, cancellationToken);
return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents);
}
public async Task<ICollection<WorkflowExecutionResult>> StartWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var metadata = new StimulusMetadata
{
CorrelationId = options?.CorrelationId,
WorkflowInstanceId = options?.WorkflowInstanceId,
Properties = options?.Properties,
ActivityInstanceId = options?.ActivityInstanceId,
Input = options?.Input
};
var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken);
var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList();
return results;
}
public async Task<WorkflowExecutionResult?> TryStartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null)
{
return await StartWorkflowAsync(definitionId, options);
}
public async Task<WorkflowExecutionResult?> ResumeWorkflowAsync(string workflowInstanceId, ResumeWorkflowRuntimeParams? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var workflowClient = await createClientAsync(workflowInstanceId, cancellationToken);
var exists = await workflowClient.InstanceExistsAsync(cancellationToken);
if (!exists)
return null;
var runWorkflowRequest = new RunWorkflowInstanceRequest
{
Input = options?.Input,
Properties = options?.Properties,
ActivityHandle = options?.ActivityHandle,
BookmarkId = options?.BookmarkId
};
var response = await workflowClient.RunInstanceAsync(runWorkflowRequest, cancellationToken);
return new(response.WorkflowInstanceId, response.Status, response.SubStatus, response.Bookmarks, response.Incidents);
}
public async Task<ICollection<WorkflowExecutionResult>> ResumeWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var metadata = new StimulusMetadata
{
CorrelationId = options?.CorrelationId,
WorkflowInstanceId = options?.WorkflowInstanceId,
Properties = options?.Properties,
ActivityInstanceId = options?.ActivityInstanceId,
Input = options?.Input
};
var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken);
var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList();
return results;
}
public async Task<TriggerWorkflowsResult> TriggerWorkflowsAsync(string activityTypeName, object bookmarkPayload, TriggerWorkflowsOptions? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var metadata = new StimulusMetadata
{
CorrelationId = options?.CorrelationId,
WorkflowInstanceId = options?.WorkflowInstanceId,
Properties = options?.Properties,
ActivityInstanceId = options?.ActivityInstanceId,
Input = options?.Input
};
var result = await stimulusSender.SendAsync(activityTypeName, bookmarkPayload, metadata, cancellationToken);
var results = result.WorkflowInstanceResponses.Select(x => new WorkflowExecutionResult(x.WorkflowInstanceId, x.Status, x.SubStatus, x.Bookmarks, x.Incidents)).ToList();
return new(results);
}
public async Task<WorkflowExecutionResult> ExecuteWorkflowAsync(WorkflowMatch match, ExecuteWorkflowParams? options = default)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
if (match is StartableWorkflowMatch collectedStartableWorkflow)
{
var startOptions = new StartWorkflowRuntimeParams
{
CorrelationId = collectedStartableWorkflow.CorrelationId,
Input = options?.Input,
Properties = options?.Properties,
VersionOptions = VersionOptions.Published,
TriggerActivityId = collectedStartableWorkflow.ActivityId,
CancellationToken = cancellationToken
};
var startResult = await StartWorkflowAsync(collectedStartableWorkflow.DefinitionId!, startOptions);
return startResult with
{
TriggeredActivityId = collectedStartableWorkflow.ActivityId
};
}
var collectedResumableWorkflow = (match as ResumableWorkflowMatch)!;
var runtimeOptions = new ResumeWorkflowRuntimeParams
{
CorrelationId = collectedResumableWorkflow.CorrelationId,
BookmarkId = collectedResumableWorkflow.BookmarkId,
Input = options?.Input,
Properties = options?.Properties,
CancellationToken = cancellationToken,
};
return (await ResumeWorkflowAsync(collectedResumableWorkflow.WorkflowInstanceId, runtimeOptions))!;
}
public async Task<CancellationResult> CancelWorkflowAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
{
var client = await createClientAsync(workflowInstanceId, cancellationToken);
await client.CancelAsync(cancellationToken);
return new(true);
}
public async Task<IEnumerable<WorkflowMatch>> FindWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default)
{
var startableWorkflows = await FindStartableWorkflowsAsync(filter, cancellationToken);
var resumableWorkflows = await FindResumableWorkflowsAsync(filter, cancellationToken);
var results = startableWorkflows.Concat(resumableWorkflows).ToList();
return results;
}
public async Task<WorkflowState?> ExportWorkflowStateAsync(string workflowInstanceId, CancellationToken cancellationToken = default)
{
var client = await createClientAsync(workflowInstanceId, cancellationToken);
return await client.ExportStateAsync(cancellationToken);
}
public async Task ImportWorkflowStateAsync(WorkflowState workflowState, CancellationToken cancellationToken = default)
{
var client = await createClientAsync(workflowState.Id, cancellationToken);
await client.ImportStateAsync(workflowState, cancellationToken);
}
public async Task UpdateBookmarkAsync(StoredBookmark bookmark, CancellationToken cancellationToken = default)
{
await bookmarkStore.SaveAsync(bookmark, cancellationToken);
}
public async Task<long> CountRunningWorkflowsAsync(CountRunningWorkflowsRequest request, CancellationToken cancellationToken = default)
{
var filter = new WorkflowInstanceFilter
{
DefinitionId = request.DefinitionId,
Version = request.Version,
CorrelationId = request.CorrelationId,
WorkflowStatus = WorkflowStatus.Running
};
return await workflowInstanceStore.CountAsync(filter, cancellationToken);
}
private async Task<IEnumerable<WorkflowMatch>> FindStartableWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken = default)
{
var stimulusHash = stimulusHasher.Hash(filter.ActivityTypeName, filter.BookmarkPayload, filter.Options.ActivityInstanceId);
var triggerBoundWorkflows = await triggerBoundWorkflowService.FindManyAsync(stimulusHash, cancellationToken).ToList();
var correlationId = filter.Options.CorrelationId;
var query =
from triggerBoundWorkflow in triggerBoundWorkflows
from trigger in triggerBoundWorkflow.Triggers
select new StartableWorkflowMatch(correlationId, trigger.ActivityId, triggerBoundWorkflow.WorkflowGraph.Workflow.Identity.DefinitionId, filter.BookmarkPayload);
return query.ToList();
}
private async Task<IEnumerable<WorkflowMatch>> FindResumableWorkflowsAsync(WorkflowsFilter filter, CancellationToken cancellationToken)
{
var bookmarkOptions = new FindBookmarkOptions
{
CorrelationId = filter.Options.CorrelationId,
WorkflowInstanceId = filter.Options.WorkflowInstanceId,
ActivityInstanceId = filter.Options.ActivityInstanceId
};
var bookmarkBoundWorkflows = await bookmarkBoundWorkflowService.FindManyAsync(filter.ActivityTypeName, filter.BookmarkPayload, bookmarkOptions, cancellationToken).ToList();
return (
from bookmarkBoundWorkflow in bookmarkBoundWorkflows
from bookmark in bookmarkBoundWorkflow.Bookmarks
select new ResumableWorkflowMatch(bookmarkBoundWorkflow.WorkflowInstanceId, bookmark.CorrelationId, bookmark.Id, bookmark.Payload))
.ToList();
}
}

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

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

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

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

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

@ -30,11 +30,7 @@ public partial class ProtoActorWorkflowRuntime
CancellationToken = cancellationToken
});
return new CanStartWorkflowResult(
{
CanStart = canStart,
InstanceId = null
};
return new(null, canStart);
}
/// <inheritdoc />

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

@ -36,7 +36,7 @@ public partial class LocalWorkflowRuntime
public async Task<WorkflowExecutionResult> StartWorkflowAsync(string definitionId, StartWorkflowRuntimeParams? options = null)
{
var cancellationToken = options?.CancellationToken ?? CancellationToken.None;
var client = (LocalWorkflowClient)await CreateClientAsync(options?.InstanceId, cancellationToken);
var client = await CreateClientAsync(options?.InstanceId, cancellationToken);
var createRequest = new CreateAndRunWorkflowInstanceRequest
{
Properties = options?.Properties,

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

@ -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,