Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Feature] Add scope support for workflow #1232

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[Feature] Add scope support for workflow #1232

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cd81a25
[Feature] Add scope support for workflow
ling921 Jan 27, 2024
9f35323
Revert changes for file `WorkflowTest.cs`
ling921 Jan 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions WorkflowCore.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@

Microsoft Visual Studio Solution File, Format Version 12.00
# Visual Studio Version 16
VisualStudioVersion = 16.0.29509.3
# Visual Studio Version 17
VisualStudioVersion = 17.9.34518.117
MinimumVisualStudioVersion = 10.0.40219.1
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "src", "src", "{EF47161E-E399-451C-BDE8-E92AAD3BD761}"
EndProject
Expand Down Expand Up @@ -154,6 +154,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Sample19", "sr
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "WorkflowCore.Persistence.RavenDB", "src\providers\WorkflowCore.Persistence.RavenDB\WorkflowCore.Persistence.RavenDB.csproj", "{AF205715-C8B7-42EF-BF14-AFC9E7F27242}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "WorkflowCore.TestScope", "src\samples\WorkflowCore.TestScope\WorkflowCore.TestScope.csproj", "{5F9E0E2C-BA1E-4F8A-8A46-D52050402B8A}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Expand Down Expand Up @@ -376,6 +378,10 @@ Global
{AF205715-C8B7-42EF-BF14-AFC9E7F27242}.Debug|Any CPU.Build.0 = Debug|Any CPU
{AF205715-C8B7-42EF-BF14-AFC9E7F27242}.Release|Any CPU.ActiveCfg = Release|Any CPU
{AF205715-C8B7-42EF-BF14-AFC9E7F27242}.Release|Any CPU.Build.0 = Release|Any CPU
{5F9E0E2C-BA1E-4F8A-8A46-D52050402B8A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{5F9E0E2C-BA1E-4F8A-8A46-D52050402B8A}.Debug|Any CPU.Build.0 = Debug|Any CPU
{5F9E0E2C-BA1E-4F8A-8A46-D52050402B8A}.Release|Any CPU.ActiveCfg = Release|Any CPU
{5F9E0E2C-BA1E-4F8A-8A46-D52050402B8A}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(SolutionProperties) = preSolution
HideSolutionNode = FALSE
Expand Down Expand Up @@ -438,6 +444,7 @@ Global
{54DE20BA-EBA7-4BF0-9BD9-F03766849716} = {E6CEAD8D-F565-471E-A0DC-676F54EAEDEB}
{1223ED47-3E5E-4960-B70D-DFAF550F6666} = {5080DB09-CBE8-4C45-9957-C3BB7651755E}
{AF205715-C8B7-42EF-BF14-AFC9E7F27242} = {2EEE6ABD-EE9B-473F-AF2D-6DABB85D7BA2}
{5F9E0E2C-BA1E-4F8A-8A46-D52050402B8A} = {5080DB09-CBE8-4C45-9957-C3BB7651755E}
EndGlobalSection
GlobalSection(ExtensibilityGlobals) = postSolution
SolutionGuid = {DC0FA8D3-6449-4FDA-BB46-ECF58FAD23B4}
Expand Down
7 changes: 6 additions & 1 deletion src/WorkflowCore/Interface/IWorkflowController.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
using System;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Threading.Tasks;

namespace WorkflowCore.Interface
Expand All @@ -9,6 +10,10 @@ public interface IWorkflowController
Task<string> StartWorkflow(string workflowId, int? version, object data = null, string reference=null);
Task<string> StartWorkflow<TData>(string workflowId, TData data = null, string reference=null) where TData : class, new();
Task<string> StartWorkflow<TData>(string workflowId, int? version, TData data = null, string reference=null) where TData : class, new();
Task<string> StartWorkflowWithScope(IServiceScope scope, string workflowId, object data = null, string reference = null);
Task<string> StartWorkflowWithScope(IServiceScope scope, string workflowId, int? version, object data = null, string reference = null);
Task<string> StartWorkflowWithScope<TData>(IServiceScope scope, string workflowId, TData data = null, string reference = null) where TData : class, new();
Task<string> StartWorkflowWithScope<TData>(IServiceScope scope, string workflowId, int? version, TData data = null, string reference = null) where TData : class, new();

Task PublishEvent(string eventName, string eventKey, object eventData, DateTime? effectiveDate = null);
void RegisterWorkflow<TWorkflow>() where TWorkflow : IWorkflow;
Expand Down
5 changes: 4 additions & 1 deletion src/WorkflowCore/Models/WorkflowInstance.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
using System;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Linq;

namespace WorkflowCore.Models
Expand Down Expand Up @@ -27,6 +28,8 @@ public class WorkflowInstance

public DateTime? CompleteTime { get; set; }

public IServiceScope CurrentServiceScope { get; set; }

public bool IsBranchComplete(string parentId)
{
return ExecutionPointers
Expand Down
36 changes: 32 additions & 4 deletions src/WorkflowCore/Services/WorkflowController.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,12 +37,12 @@ public WorkflowController(IPersistenceProvider persistenceStore, IDistributedLoc
_dateTimeProvider = dateTimeProvider;
}

public Task<string> StartWorkflow(string workflowId, object data = null, string reference=null)
public Task<string> StartWorkflow(string workflowId, object data = null, string reference = null)
{
return StartWorkflow(workflowId, null, data, reference);
}

public Task<string> StartWorkflow(string workflowId, int? version, object data = null, string reference=null)
public Task<string> StartWorkflow(string workflowId, int? version, object data = null, string reference = null)
{
return StartWorkflow<object>(workflowId, version, data, reference);
}
Expand All @@ -53,10 +53,37 @@ public Task<string> StartWorkflow<TData>(string workflowId, TData data = null, s
return StartWorkflow(workflowId, null, data, reference);
}

public async Task<string> StartWorkflow<TData>(string workflowId, int? version, TData data = null, string reference=null)
public Task<string> StartWorkflow<TData>(string workflowId, int? version, TData data = null, string reference = null)
where TData : class, new()
{
return StartWorkflowCore(workflowId, version, data, reference);
}

public Task<string> StartWorkflowWithScope(IServiceScope scope, string workflowId, object data = null, string reference = null)
{
return StartWorkflowWithScope(scope, workflowId, null, data, reference);
}

public Task<string> StartWorkflowWithScope(IServiceScope scope, string workflowId, int? version, object data = null, string reference = null)
{
return StartWorkflowWithScope<object>(scope, workflowId, version, data, reference);
}

public Task<string> StartWorkflowWithScope<TData>(IServiceScope scope, string workflowId, TData data = null, string reference = null)
where TData : class, new()
{
return StartWorkflowWithScope(scope, workflowId, null, data, reference);
}

public Task<string> StartWorkflowWithScope<TData>(IServiceScope scope, string workflowId, int? version, TData data = null, string reference = null)
where TData : class, new()
{
return StartWorkflowCore(workflowId, version, data, reference, scope);
}

private async Task<string> StartWorkflowCore<TData>(string workflowId, int? version, TData data, string reference, IServiceScope workflowScope = null)
where TData : class, new()
{
var def = _registry.GetDefinition(workflowId, version);
if (def == null)
{
Expand All @@ -72,7 +99,8 @@ public async Task<string> StartWorkflow<TData>(string workflowId, int? version,
NextExecution = 0,
CreateTime = _dateTimeProvider.UtcNow,
Status = WorkflowStatus.Runnable,
Reference = reference
Reference = reference,
CurrentServiceScope = workflowScope
};

if ((def.DataType != null) && (data == null))
Expand Down
80 changes: 45 additions & 35 deletions src/WorkflowCore/Services/WorkflowExecutor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,50 +156,60 @@ private async Task ExecuteStep(WorkflowInstance workflow, WorkflowStep step, Exe
CancellationToken = cancellationToken
};

using (var scope = _scopeProvider.CreateScope(context))
var shouldDisposeScope = false;
var scope = context.Workflow.CurrentServiceScope;
if (scope == null)
{
_logger.LogDebug("Starting step {StepName} on workflow {WorkflowId}", step.Name, workflow.Id);
scope = _scopeProvider.CreateScope(context);
shouldDisposeScope = true;
}

_logger.LogDebug("Starting step {StepName} on workflow {WorkflowId}", step.Name, workflow.Id);

IStepBody body = step.ConstructBody(scope.ServiceProvider);
var stepExecutor = scope.ServiceProvider.GetRequiredService<IStepExecutor>();
IStepBody body = step.ConstructBody(scope.ServiceProvider);
var stepExecutor = scope.ServiceProvider.GetRequiredService<IStepExecutor>();

if (body == null)
if (body == null)
{
_logger.LogError("Unable to construct step body {BodyType}", step.BodyType.ToString());
pointer.SleepUntil = _datetimeProvider.UtcNow.Add(_options.ErrorRetryInterval);
wfResult.Errors.Add(new ExecutionError
{
_logger.LogError("Unable to construct step body {BodyType}", step.BodyType.ToString());
pointer.SleepUntil = _datetimeProvider.UtcNow.Add(_options.ErrorRetryInterval);
wfResult.Errors.Add(new ExecutionError
{
WorkflowId = workflow.Id,
ExecutionPointerId = pointer.Id,
ErrorTime = _datetimeProvider.UtcNow,
Message = $"Unable to construct step body {step.BodyType}"
});
return;
}
WorkflowId = workflow.Id,
ExecutionPointerId = pointer.Id,
ErrorTime = _datetimeProvider.UtcNow,
Message = $"Unable to construct step body {step.BodyType}"
});
return;
}

foreach (var input in step.Inputs)
input.AssignInput(workflow.Data, body, context);
foreach (var input in step.Inputs)
input.AssignInput(workflow.Data, body, context);

switch (step.BeforeExecute(wfResult, context, pointer, body))
{
case ExecutionPipelineDirective.Defer:
return;
case ExecutionPipelineDirective.EndWorkflow:
workflow.Status = WorkflowStatus.Complete;
workflow.CompleteTime = _datetimeProvider.UtcNow;
return;
}
switch (step.BeforeExecute(wfResult, context, pointer, body))
{
case ExecutionPipelineDirective.Defer:
return;
case ExecutionPipelineDirective.EndWorkflow:
workflow.Status = WorkflowStatus.Complete;
workflow.CompleteTime = _datetimeProvider.UtcNow;
return;
}

var result = await stepExecutor.ExecuteStep(context, body);
var result = await stepExecutor.ExecuteStep(context, body);

if (result.Proceed)
{
foreach (var output in step.Outputs)
output.AssignOutput(workflow.Data, body, context);
}
if (result.Proceed)
{
foreach (var output in step.Outputs)
output.AssignOutput(workflow.Data, body, context);
}

_executionResultProcessor.ProcessExecutionResult(workflow, def, pointer, step, result, wfResult);
step.AfterExecute(wfResult, context, result, pointer);
_executionResultProcessor.ProcessExecutionResult(workflow, def, pointer, step, result, wfResult);
step.AfterExecute(wfResult, context, result, pointer);

if (shouldDisposeScope)
{
scope.Dispose();
}
}

Expand Down
31 changes: 27 additions & 4 deletions src/WorkflowCore/Services/WorkflowHost.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using OpenTelemetry.Trace;
using WorkflowCore.Interface;
Expand Down Expand Up @@ -49,7 +50,7 @@ public WorkflowHost(IPersistenceProvider persistenceStore, IQueueProvider queueP
_activityController = activityController;
_lifeCycleEventHub = lifeCycleEventHub;
}

public Task<string> StartWorkflow(string workflowId, object data = null, string reference=null)
{
return _workflowController.StartWorkflow(workflowId, data, reference);
Expand All @@ -65,13 +66,35 @@ public Task<string> StartWorkflow<TData>(string workflowId, TData data = null, s
{
return _workflowController.StartWorkflow<TData>(workflowId, null, data, reference);
}

public Task<string> StartWorkflow<TData>(string workflowId, int? version, TData data = null, string reference=null)
where TData : class, new()
{
return _workflowController.StartWorkflow(workflowId, version, data, reference);
}

public Task<string> StartWorkflowWithScope(IServiceScope scope, string workflowId, object data = null, string reference = null)
{
return _workflowController.StartWorkflowWithScope(scope, workflowId, data, reference);
}

public Task<string> StartWorkflowWithScope(IServiceScope scope, string workflowId, int? version, object data = null, string reference = null)
{
return _workflowController.StartWorkflowWithScope(scope, workflowId, version, data, reference);
}

public Task<string> StartWorkflowWithScope<TData>(IServiceScope scope, string workflowId, TData data = null, string reference = null)
where TData : class, new()
{
return _workflowController.StartWorkflowWithScope(scope, workflowId, null, data, reference);
}

public Task<string> StartWorkflowWithScope<TData>(IServiceScope scope, string workflowId, int? version, TData data = null, string reference = null)
where TData : class, new()
{
return _workflowController.StartWorkflowWithScope(scope, workflowId, version, data, reference);
}

public Task PublishEvent(string eventName, string eventKey, object eventData, DateTime? effectiveDate = null)
{
return _workflowController.PublishEvent(eventName, eventKey, eventData, effectiveDate);
Expand All @@ -81,7 +104,7 @@ public void Start()
{
StartAsync(CancellationToken.None).Wait();
}

public async Task StartAsync(CancellationToken cancellationToken)
{
var activity = WorkflowActivity.StartHost();
Expand Down Expand Up @@ -118,7 +141,7 @@ public void Stop()
{
StopAsync(CancellationToken.None).Wait();
}

public async Task StopAsync(CancellationToken cancellationToken)
{
_shutdown = true;
Expand Down
Loading