From 5cb45c154da8dc9968f24d1bb337958293ac0a4a Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Tue, 16 Oct 2018 15:00:25 +0200 Subject: [PATCH] Incremental work on persistence --- Flowsharp.sln | 28 ++++++++ .../Activities/Activity.cs | 1 + .../Activities/IActivity.cs | 1 + src/Flowsharp.Abstractions/Models/Workflow.cs | 5 +- .../Models/WorkflowExecutionContext.cs | 4 +- .../Tokenizers/WorkflowTokenizer.cs | 4 +- src/Flowsharp.Core/WorkflowInvoker.cs | 2 +- .../Extensions/SpecificationExtensions.cs | 26 +++++++ .../Flowsharp.Persistence.Abstractions.csproj | 12 ++++ .../IWorkflowDefinitionStore.cs | 15 ++++ .../IWorkflowInstanceStore.cs | 15 ++++ .../Models/WorkflowDefinition.cs | 10 +++ .../Models/WorkflowInstance.cs | 11 +++ .../Primitives/AndSpecification.cs | 19 +++++ .../Primitives/NotSpecification.cs | 17 +++++ .../Primitives/OrSpecification.cs | 19 +++++ .../Specifications/ISpecification.cs | 12 ++++ .../Specifications/ISpecificationVisitor.cs | 11 +++ ...IWorkflowDefinitionSpecificationVisitor.cs | 10 +++ .../IWorkflowInstanceSpecificationVisitor.cs | 9 +++ .../WorkflowIsBlockedOnActivity.cs | 30 ++++++++ .../WorkflowStartsWithActivity.cs | 32 +++++++++ .../Flowsharp.Persistence.InMemory.csproj | 11 +++ .../InMemoryWorkflowDefinitionStore.cs | 32 +++++++++ .../InMemoryWorkflowInstanceStore.cs | 37 ++++++++++ .../Flowsharp.Runtime.Abstractions.csproj | 11 +++ .../IWorkflowHost.cs | 17 +++++ .../Flowsharp.Runtime.csproj | 12 ++++ src/Flowsharp.Runtime/WorkflowHost.cs | 71 +++++++++++++++++++ .../Flowsharp.Samples.Console.csproj | 4 ++ src/Flowsharp.Samples.Console/Program.cs | 17 ++++- .../AdditionWorkflowProgramLongRunning.cs | 2 +- .../Programs/FileBasedWorkflowProgram.cs | 2 +- .../Programs/WorkflowHostProgram.cs | 51 +++++++++++++ 34 files changed, 547 insertions(+), 13 deletions(-) create mode 100644 src/Flowsharp.Persistence.Abstractions/Extensions/SpecificationExtensions.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Flowsharp.Persistence.Abstractions.csproj create mode 100644 src/Flowsharp.Persistence.Abstractions/IWorkflowDefinitionStore.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/IWorkflowInstanceStore.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Models/WorkflowDefinition.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Models/WorkflowInstance.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Primitives/AndSpecification.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Primitives/NotSpecification.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Primitives/OrSpecification.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Specifications/ISpecification.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Specifications/ISpecificationVisitor.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Specifications/IWorkflowDefinitionSpecificationVisitor.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Specifications/IWorkflowInstanceSpecificationVisitor.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Specifications/WorkflowIsBlockedOnActivity.cs create mode 100644 src/Flowsharp.Persistence.Abstractions/Specifications/WorkflowStartsWithActivity.cs create mode 100644 src/Flowsharp.Persistence.InMemory/Flowsharp.Persistence.InMemory.csproj create mode 100644 src/Flowsharp.Persistence.InMemory/InMemoryWorkflowDefinitionStore.cs create mode 100644 src/Flowsharp.Persistence.InMemory/InMemoryWorkflowInstanceStore.cs create mode 100644 src/Flowsharp.Runtime.Abstractions/Flowsharp.Runtime.Abstractions.csproj create mode 100644 src/Flowsharp.Runtime.Abstractions/IWorkflowHost.cs create mode 100644 src/Flowsharp.Runtime/Flowsharp.Runtime.csproj create mode 100644 src/Flowsharp.Runtime/WorkflowHost.cs create mode 100644 src/Flowsharp.Samples.Console/Programs/WorkflowHostProgram.cs diff --git a/Flowsharp.sln b/Flowsharp.sln index 7b88e75b5..7aa85cab3 100644 --- a/Flowsharp.sln +++ b/Flowsharp.sln @@ -19,6 +19,14 @@ EndProjectSection EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Flowsharp.Abstractions", "src\Flowsharp.Abstractions\Flowsharp.Abstractions.csproj", "{3130C574-557C-4DA2-933E-AED7A9EBE48A}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Flowsharp.Persistence.Abstractions", "src\Flowsharp.Persistence.Abstractions\Flowsharp.Persistence.Abstractions.csproj", "{A5829896-89C5-4C4E-ADC7-C806C8077316}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Flowsharp.Persistence.InMemory", "src\Flowsharp.Persistence.InMemory\Flowsharp.Persistence.InMemory.csproj", "{E93A1A6B-708F-4DCC-A399-1B8EC7682F92}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Flowsharp.Runtime", "src\Flowsharp.Runtime\Flowsharp.Runtime.csproj", "{3C1DCAA2-D77E-4F96-9899-DA8E0E0C757F}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Flowsharp.Runtime.Abstractions", "src\Flowsharp.Runtime.Abstractions\Flowsharp.Runtime.Abstractions.csproj", "{792362BF-D26D-46E9-8970-E5668941AA75}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -37,6 +45,22 @@ Global {3130C574-557C-4DA2-933E-AED7A9EBE48A}.Debug|Any CPU.Build.0 = Debug|Any CPU {3130C574-557C-4DA2-933E-AED7A9EBE48A}.Release|Any CPU.ActiveCfg = Release|Any CPU {3130C574-557C-4DA2-933E-AED7A9EBE48A}.Release|Any CPU.Build.0 = Release|Any CPU + {A5829896-89C5-4C4E-ADC7-C806C8077316}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {A5829896-89C5-4C4E-ADC7-C806C8077316}.Debug|Any CPU.Build.0 = Debug|Any CPU + {A5829896-89C5-4C4E-ADC7-C806C8077316}.Release|Any CPU.ActiveCfg = Release|Any CPU + {A5829896-89C5-4C4E-ADC7-C806C8077316}.Release|Any CPU.Build.0 = Release|Any CPU + {E93A1A6B-708F-4DCC-A399-1B8EC7682F92}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {E93A1A6B-708F-4DCC-A399-1B8EC7682F92}.Debug|Any CPU.Build.0 = Debug|Any CPU + {E93A1A6B-708F-4DCC-A399-1B8EC7682F92}.Release|Any CPU.ActiveCfg = Release|Any CPU + {E93A1A6B-708F-4DCC-A399-1B8EC7682F92}.Release|Any CPU.Build.0 = Release|Any CPU + {3C1DCAA2-D77E-4F96-9899-DA8E0E0C757F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {3C1DCAA2-D77E-4F96-9899-DA8E0E0C757F}.Debug|Any CPU.Build.0 = Debug|Any CPU + {3C1DCAA2-D77E-4F96-9899-DA8E0E0C757F}.Release|Any CPU.ActiveCfg = Release|Any CPU + {3C1DCAA2-D77E-4F96-9899-DA8E0E0C757F}.Release|Any CPU.Build.0 = Release|Any CPU + {792362BF-D26D-46E9-8970-E5668941AA75}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {792362BF-D26D-46E9-8970-E5668941AA75}.Debug|Any CPU.Build.0 = Debug|Any CPU + {792362BF-D26D-46E9-8970-E5668941AA75}.Release|Any CPU.ActiveCfg = Release|Any CPU + {792362BF-D26D-46E9-8970-E5668941AA75}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -48,5 +72,9 @@ Global {300EE2D5-54C5-46F2-AD03-BB43589EA074} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} {48EDB976-3227-4DFD-BBB3-3BC9AEA19A88} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} {3130C574-557C-4DA2-933E-AED7A9EBE48A} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} + {A5829896-89C5-4C4E-ADC7-C806C8077316} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} + {E93A1A6B-708F-4DCC-A399-1B8EC7682F92} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} + {3C1DCAA2-D77E-4F96-9899-DA8E0E0C757F} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} + {792362BF-D26D-46E9-8970-E5668941AA75} = {DA71CDAA-8DD3-4D5F-9FBD-8E4B37A2D925} EndGlobalSection EndGlobal diff --git a/src/Flowsharp.Abstractions/Activities/Activity.cs b/src/Flowsharp.Abstractions/Activities/Activity.cs index a49cfd9b7..053f36c31 100644 --- a/src/Flowsharp.Abstractions/Activities/Activity.cs +++ b/src/Flowsharp.Abstractions/Activities/Activity.cs @@ -2,5 +2,6 @@ { public abstract class Activity : IActivity { + public virtual string Name => GetType().Name; } } diff --git a/src/Flowsharp.Abstractions/Activities/IActivity.cs b/src/Flowsharp.Abstractions/Activities/IActivity.cs index 2179eea2b..92eebb81f 100644 --- a/src/Flowsharp.Abstractions/Activities/IActivity.cs +++ b/src/Flowsharp.Abstractions/Activities/IActivity.cs @@ -2,5 +2,6 @@ { public interface IActivity { + string Name { get; } } } diff --git a/src/Flowsharp.Abstractions/Models/Workflow.cs b/src/Flowsharp.Abstractions/Models/Workflow.cs index 8dbf4f795..ef7212790 100644 --- a/src/Flowsharp.Abstractions/Models/Workflow.cs +++ b/src/Flowsharp.Abstractions/Models/Workflow.cs @@ -19,16 +19,15 @@ namespace Flowsharp.Models CurrentScope = new WorkflowExecutionScope(); Scopes = new Stack(new[]{ CurrentScope }); Arguments = new Variables(); - HaltedActivities = new List(); + BlockingActivities = new List(); } - [JsonConverter(typeof(StringEnumConverter))] public WorkflowStatus Status { get; set; } public IList Activities { get; set; } = new List(); public IList Connections { get; set; } = new List(); public Stack Scopes { get; set; } public WorkflowExecutionScope CurrentScope { get; set; } public Variables Arguments { get; set; } - public IList HaltedActivities { get; set; } + public IList BlockingActivities { get; set; } } } \ No newline at end of file diff --git a/src/Flowsharp.Abstractions/Models/WorkflowExecutionContext.cs b/src/Flowsharp.Abstractions/Models/WorkflowExecutionContext.cs index 55482b5fc..ff98a1c14 100644 --- a/src/Flowsharp.Abstractions/Models/WorkflowExecutionContext.cs +++ b/src/Flowsharp.Abstractions/Models/WorkflowExecutionContext.cs @@ -63,9 +63,9 @@ namespace Flowsharp.Models public Task HaltAsync(CancellationToken cancellationToken) { var activity = CurrentActivity; - if (!Workflow.HaltedActivities.Contains(activity)) + if (!Workflow.BlockingActivities.Contains(activity)) { - Workflow.HaltedActivities.Add(activity); + Workflow.BlockingActivities.Add(activity); } Workflow.Status = WorkflowStatus.Halted; diff --git a/src/Flowsharp.Core/Serialization/Tokenizers/WorkflowTokenizer.cs b/src/Flowsharp.Core/Serialization/Tokenizers/WorkflowTokenizer.cs index 45c64fa72..a12b99124 100644 --- a/src/Flowsharp.Core/Serialization/Tokenizers/WorkflowTokenizer.cs +++ b/src/Flowsharp.Core/Serialization/Tokenizers/WorkflowTokenizer.cs @@ -61,7 +61,7 @@ namespace Flowsharp.Serialization.Tokenizers Status = (WorkflowStatus)Enum.Parse(typeof(WorkflowStatus), token["status"].Value()), Activities = activityDictionary.Values.ToList(), Connections = DeserializeConnections(token, activityDictionary).ToList(), - HaltedActivities = DeserializeHaltedActivities(token, activityDictionary).ToList(), + BlockingActivities = DeserializeHaltedActivities(token, activityDictionary).ToList(), Scopes = new Stack(scopeLookup.Values), CurrentScope = scopeLookup[currentScopeId] }; @@ -110,7 +110,7 @@ namespace Flowsharp.Serialization.Tokenizers { var haltedActivityModels = new JArray(); - foreach (var activity in workflow.HaltedActivities) + foreach (var activity in workflow.BlockingActivities) { var activityId = context.ActivityIdLookup[activity]; haltedActivityModels.Add(activityId); diff --git a/src/Flowsharp.Core/WorkflowInvoker.cs b/src/Flowsharp.Core/WorkflowInvoker.cs index ae54b391e..db391cd6b 100644 --- a/src/Flowsharp.Core/WorkflowInvoker.cs +++ b/src/Flowsharp.Core/WorkflowInvoker.cs @@ -27,7 +27,7 @@ namespace Flowsharp var isResuming = workflowExecutionContext.Workflow.Status == WorkflowStatus.Resuming; if (startActivity != null) - workflow.HaltedActivities.Remove(startActivity); + workflow.BlockingActivities.Remove(startActivity); else startActivity = workflow.Activities.First(); diff --git a/src/Flowsharp.Persistence.Abstractions/Extensions/SpecificationExtensions.cs b/src/Flowsharp.Persistence.Abstractions/Extensions/SpecificationExtensions.cs new file mode 100644 index 000000000..46f64a2ad --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Extensions/SpecificationExtensions.cs @@ -0,0 +1,26 @@ +using Flowsharp.Persistence.Primitives; +using Flowsharp.Persistence.Specifications; + +namespace Flowsharp.Persistence.Extensions +{ + public static class SpecificationExtensions + { + public static ISpecification And(this ISpecification left, ISpecification right) + where TVisitor : ISpecificationVisitor + { + return new AndSpecification(left, right); + } + + public static ISpecification Or(this ISpecification left, ISpecification right) + where TVisitor : ISpecificationVisitor + { + return new OrSpecification(left, right); + } + + public static ISpecification Not(this ISpecification specification) + where TVisitor : ISpecificationVisitor + { + return new NotSpecification(specification); + } + } +} diff --git a/src/Flowsharp.Persistence.Abstractions/Flowsharp.Persistence.Abstractions.csproj b/src/Flowsharp.Persistence.Abstractions/Flowsharp.Persistence.Abstractions.csproj new file mode 100644 index 000000000..3f2cddfb2 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Flowsharp.Persistence.Abstractions.csproj @@ -0,0 +1,12 @@ + + + + netstandard2.0 + Flowsharp.Persistence + + + + + + + diff --git a/src/Flowsharp.Persistence.Abstractions/IWorkflowDefinitionStore.cs b/src/Flowsharp.Persistence.Abstractions/IWorkflowDefinitionStore.cs new file mode 100644 index 000000000..5a9e42592 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/IWorkflowDefinitionStore.cs @@ -0,0 +1,15 @@ +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Flowsharp.Models; +using Flowsharp.Persistence.Models; +using Flowsharp.Persistence.Specifications; + +namespace Flowsharp.Persistence +{ + public interface IWorkflowDefinitionStore + { + Task> GetManyAsync(ISpecification specification, CancellationToken cancellationToken); + Task AddAsync(WorkflowDefinition value, CancellationToken cancellationToken); + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.Abstractions/IWorkflowInstanceStore.cs b/src/Flowsharp.Persistence.Abstractions/IWorkflowInstanceStore.cs new file mode 100644 index 000000000..d1d19be1d --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/IWorkflowInstanceStore.cs @@ -0,0 +1,15 @@ +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Flowsharp.Persistence.Models; +using Flowsharp.Persistence.Specifications; + +namespace Flowsharp.Persistence +{ + public interface IWorkflowInstanceStore + { + Task> GetManyAsync(ISpecification specification, CancellationToken cancellationToken); + Task AddAsync(WorkflowInstance value, CancellationToken cancellationToken); + Task UpdateAsync(WorkflowInstance value, CancellationToken cancellationToken); + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.Abstractions/Models/WorkflowDefinition.cs b/src/Flowsharp.Persistence.Abstractions/Models/WorkflowDefinition.cs new file mode 100644 index 000000000..869b848a6 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Models/WorkflowDefinition.cs @@ -0,0 +1,10 @@ +using Flowsharp.Models; + +namespace Flowsharp.Persistence.Models +{ + public class WorkflowDefinition + { + public string Id { get; set; } + public Workflow Workflow { get; set; } + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.Abstractions/Models/WorkflowInstance.cs b/src/Flowsharp.Persistence.Abstractions/Models/WorkflowInstance.cs new file mode 100644 index 000000000..e95f24ec2 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Models/WorkflowInstance.cs @@ -0,0 +1,11 @@ +using Flowsharp.Models; + +namespace Flowsharp.Persistence.Models +{ + public class WorkflowInstance + { + public string Id { get; set; } + public string WorkflowDefinitionId { get; set; } + public Workflow Workflow { get; set; } + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.Abstractions/Primitives/AndSpecification.cs b/src/Flowsharp.Persistence.Abstractions/Primitives/AndSpecification.cs new file mode 100644 index 000000000..d47d11f09 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Primitives/AndSpecification.cs @@ -0,0 +1,19 @@ +using Flowsharp.Persistence.Specifications; + +namespace Flowsharp.Persistence.Primitives +{ + public class AndSpecification : ISpecification where TVisitor : ISpecificationVisitor + { + public ISpecification Left { get; } + public ISpecification Right { get; } + + public AndSpecification(ISpecification left, ISpecification right) + { + Left = left; + Right = right; + } + + public void Accept(TVisitor visitor) => visitor.Visit(this); + public bool IsSatisfiedBy(T obj) => Left.IsSatisfiedBy(obj) && Right.IsSatisfiedBy(obj); + } +} diff --git a/src/Flowsharp.Persistence.Abstractions/Primitives/NotSpecification.cs b/src/Flowsharp.Persistence.Abstractions/Primitives/NotSpecification.cs new file mode 100644 index 000000000..c5662f8e9 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Primitives/NotSpecification.cs @@ -0,0 +1,17 @@ +using Flowsharp.Persistence.Specifications; + +namespace Flowsharp.Persistence.Primitives +{ + public class NotSpecification : ISpecification where TVisitor : ISpecificationVisitor + { + public ISpecification Specification { get; } + + public NotSpecification(ISpecification specification) + { + Specification = specification; + } + + public void Accept(TVisitor visitor) => visitor.Visit(this); + public bool IsSatisfiedBy(T obj) => !Specification.IsSatisfiedBy(obj); + } +} diff --git a/src/Flowsharp.Persistence.Abstractions/Primitives/OrSpecification.cs b/src/Flowsharp.Persistence.Abstractions/Primitives/OrSpecification.cs new file mode 100644 index 000000000..3c4840eda --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Primitives/OrSpecification.cs @@ -0,0 +1,19 @@ +using Flowsharp.Persistence.Specifications; + +namespace Flowsharp.Persistence.Primitives +{ + public class OrSpecification : ISpecification where TVisitor : ISpecificationVisitor + { + public ISpecification Left { get; } + public ISpecification Right { get; } + + public OrSpecification(ISpecification left, ISpecification right) + { + Left = left; + Right = right; + } + + public void Accept(TVisitor visitor) => visitor.Visit(this); + public bool IsSatisfiedBy(T obj) => Left.IsSatisfiedBy(obj) || Right.IsSatisfiedBy(obj); + } +} diff --git a/src/Flowsharp.Persistence.Abstractions/Specifications/ISpecification.cs b/src/Flowsharp.Persistence.Abstractions/Specifications/ISpecification.cs new file mode 100644 index 000000000..908952b48 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Specifications/ISpecification.cs @@ -0,0 +1,12 @@ +namespace Flowsharp.Persistence.Specifications +{ + public interface ISpecification + { + bool IsSatisfiedBy(T item); + } + + public interface ISpecification : ISpecification where TVisitor : ISpecificationVisitor + { + void Accept(TVisitor visitor); + } +} diff --git a/src/Flowsharp.Persistence.Abstractions/Specifications/ISpecificationVisitor.cs b/src/Flowsharp.Persistence.Abstractions/Specifications/ISpecificationVisitor.cs new file mode 100644 index 000000000..5b442fc52 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Specifications/ISpecificationVisitor.cs @@ -0,0 +1,11 @@ +using Flowsharp.Persistence.Primitives; + +namespace Flowsharp.Persistence.Specifications +{ + public interface ISpecificationVisitor where TVisitor : ISpecificationVisitor + { + void Visit(AndSpecification specification); + void Visit(OrSpecification specification); + void Visit(NotSpecification specification); + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.Abstractions/Specifications/IWorkflowDefinitionSpecificationVisitor.cs b/src/Flowsharp.Persistence.Abstractions/Specifications/IWorkflowDefinitionSpecificationVisitor.cs new file mode 100644 index 000000000..2256248fa --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Specifications/IWorkflowDefinitionSpecificationVisitor.cs @@ -0,0 +1,10 @@ +using Flowsharp.Models; +using Flowsharp.Persistence.Models; + +namespace Flowsharp.Persistence.Specifications +{ + public interface IWorkflowDefinitionSpecificationVisitor : ISpecificationVisitor + { + void Visit(WorkflowStartsWithActivity specification); + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.Abstractions/Specifications/IWorkflowInstanceSpecificationVisitor.cs b/src/Flowsharp.Persistence.Abstractions/Specifications/IWorkflowInstanceSpecificationVisitor.cs new file mode 100644 index 000000000..e16ec77f7 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Specifications/IWorkflowInstanceSpecificationVisitor.cs @@ -0,0 +1,9 @@ +using Flowsharp.Persistence.Models; + +namespace Flowsharp.Persistence.Specifications +{ + public interface IWorkflowInstanceSpecificationVisitor : ISpecificationVisitor + { + void Visit(WorkflowIsBlockedOnActivity specification); + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.Abstractions/Specifications/WorkflowIsBlockedOnActivity.cs b/src/Flowsharp.Persistence.Abstractions/Specifications/WorkflowIsBlockedOnActivity.cs new file mode 100644 index 000000000..773588ccf --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Specifications/WorkflowIsBlockedOnActivity.cs @@ -0,0 +1,30 @@ +using System.Linq; +using Flowsharp.Persistence.Models; + +namespace Flowsharp.Persistence.Specifications +{ + public class WorkflowIsBlockedOnActivity: ISpecification + { + public WorkflowIsBlockedOnActivity(string activityName) + { + ActivityName = activityName; + } + + public string ActivityName { get; } + + public bool IsSatisfiedBy(WorkflowInstance value) + { + var query = + from activity in value.Workflow.BlockingActivities + where activity.Name == ActivityName + select activity; + + return query.Any(); + } + + public void Accept(IWorkflowInstanceSpecificationVisitor visitor) + { + visitor.Visit(this); + } + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.Abstractions/Specifications/WorkflowStartsWithActivity.cs b/src/Flowsharp.Persistence.Abstractions/Specifications/WorkflowStartsWithActivity.cs new file mode 100644 index 000000000..3f06a72e9 --- /dev/null +++ b/src/Flowsharp.Persistence.Abstractions/Specifications/WorkflowStartsWithActivity.cs @@ -0,0 +1,32 @@ +using System.Linq; +using Flowsharp.Models; +using Flowsharp.Persistence.Models; + +namespace Flowsharp.Persistence.Specifications +{ + public class WorkflowStartsWithActivity : ISpecification + { + public WorkflowStartsWithActivity(string activityName) + { + ActivityName = activityName; + } + + public string ActivityName { get; } + + public bool IsSatisfiedBy(WorkflowDefinition value) + { + var query = + from activity in value.Workflow.Activities + where activity.Name == ActivityName + where !value.Workflow.Connections.Select(x => x.Target.Activity).Contains(activity) + select activity; + + return query.Any(); + } + + public void Accept(IWorkflowDefinitionSpecificationVisitor visitor) + { + visitor.Visit(this); + } + } +} diff --git a/src/Flowsharp.Persistence.InMemory/Flowsharp.Persistence.InMemory.csproj b/src/Flowsharp.Persistence.InMemory/Flowsharp.Persistence.InMemory.csproj new file mode 100644 index 000000000..06e900520 --- /dev/null +++ b/src/Flowsharp.Persistence.InMemory/Flowsharp.Persistence.InMemory.csproj @@ -0,0 +1,11 @@ + + + + netstandard2.0 + + + + + + + diff --git a/src/Flowsharp.Persistence.InMemory/InMemoryWorkflowDefinitionStore.cs b/src/Flowsharp.Persistence.InMemory/InMemoryWorkflowDefinitionStore.cs new file mode 100644 index 000000000..57c79fd97 --- /dev/null +++ b/src/Flowsharp.Persistence.InMemory/InMemoryWorkflowDefinitionStore.cs @@ -0,0 +1,32 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Flowsharp.Persistence.Models; +using Flowsharp.Persistence.Specifications; + +namespace Flowsharp.Persistence.InMemory +{ + public class InMemoryWorkflowDefinitionStore : IWorkflowDefinitionStore + { + public InMemoryWorkflowDefinitionStore() + { + WorkflowDefinitions = new List(); + } + + public IList WorkflowDefinitions { get; } + + public Task> GetManyAsync(ISpecification specification, CancellationToken cancellationToken) + { + var query = WorkflowDefinitions.AsQueryable().Where(x => specification.IsSatisfiedBy(x)); + var matches = query.Distinct().ToList(); + return Task.FromResult>(matches); + } + + public Task AddAsync(WorkflowDefinition value, CancellationToken cancellationToken) + { + WorkflowDefinitions.Add(value); + return Task.CompletedTask; + } + } +} \ No newline at end of file diff --git a/src/Flowsharp.Persistence.InMemory/InMemoryWorkflowInstanceStore.cs b/src/Flowsharp.Persistence.InMemory/InMemoryWorkflowInstanceStore.cs new file mode 100644 index 000000000..4858f6eca --- /dev/null +++ b/src/Flowsharp.Persistence.InMemory/InMemoryWorkflowInstanceStore.cs @@ -0,0 +1,37 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Flowsharp.Persistence.Models; +using Flowsharp.Persistence.Specifications; + +namespace Flowsharp.Persistence.InMemory +{ + public class InMemoryWorkflowInstanceStore : IWorkflowInstanceStore + { + public InMemoryWorkflowInstanceStore() + { + WorkflowInstances = new List(); + } + + public IList WorkflowInstances { get; } + + public Task> GetManyAsync(ISpecification specification, CancellationToken cancellationToken) + { + var query = WorkflowInstances.AsQueryable().Where(x => specification.IsSatisfiedBy(x)); + var matches = query.Distinct().ToList(); + return Task.FromResult>(matches); + } + + public Task AddAsync(WorkflowInstance value, CancellationToken cancellationToken) + { + WorkflowInstances.Add(value); + return Task.CompletedTask; + } + + public Task UpdateAsync(WorkflowInstance value, CancellationToken cancellationToken) + { + return Task.CompletedTask; + } + } +} \ No newline at end of file diff --git a/src/Flowsharp.Runtime.Abstractions/Flowsharp.Runtime.Abstractions.csproj b/src/Flowsharp.Runtime.Abstractions/Flowsharp.Runtime.Abstractions.csproj new file mode 100644 index 000000000..633d246d2 --- /dev/null +++ b/src/Flowsharp.Runtime.Abstractions/Flowsharp.Runtime.Abstractions.csproj @@ -0,0 +1,11 @@ + + + + netstandard2.0 + + + + + + + diff --git a/src/Flowsharp.Runtime.Abstractions/IWorkflowHost.cs b/src/Flowsharp.Runtime.Abstractions/IWorkflowHost.cs new file mode 100644 index 000000000..64eb203b2 --- /dev/null +++ b/src/Flowsharp.Runtime.Abstractions/IWorkflowHost.cs @@ -0,0 +1,17 @@ +using System.Threading; +using System.Threading.Tasks; +using Flowsharp.Models; + +namespace Flowsharp.Runtime.Abstractions +{ + /// + /// Provides higher-level workflow functionality, such as triggering new and halted workflows based on received stimuli. + /// + public interface IWorkflowHost + { + /// + /// Starts new workflows that start with the specified activity name and resumes halted workflows that are blocked on activities with the specified activity name. + /// + Task TriggerWorkflowAsync(string activityName, Variables arguments, CancellationToken cancellationToken); + } +} \ No newline at end of file diff --git a/src/Flowsharp.Runtime/Flowsharp.Runtime.csproj b/src/Flowsharp.Runtime/Flowsharp.Runtime.csproj new file mode 100644 index 000000000..d2323bb0c --- /dev/null +++ b/src/Flowsharp.Runtime/Flowsharp.Runtime.csproj @@ -0,0 +1,12 @@ + + + + netstandard2.0 + + + + + + + + diff --git a/src/Flowsharp.Runtime/WorkflowHost.cs b/src/Flowsharp.Runtime/WorkflowHost.cs new file mode 100644 index 000000000..0fe12f411 --- /dev/null +++ b/src/Flowsharp.Runtime/WorkflowHost.cs @@ -0,0 +1,71 @@ +using System; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Flowsharp.Models; +using Flowsharp.Persistence; +using Flowsharp.Persistence.Models; +using Flowsharp.Persistence.Specifications; +using Flowsharp.Runtime.Abstractions; + +namespace Flowsharp.Runtime +{ + public class WorkflowHost : IWorkflowHost + { + private readonly IWorkflowInvoker invoker; + private readonly IWorkflowDefinitionStore workflowDefinitionStore; + private readonly IWorkflowInstanceStore workflowInstanceStore; + + public WorkflowHost(IWorkflowInvoker invoker, IWorkflowDefinitionStore workflowDefinitionStore, IWorkflowInstanceStore workflowInstanceStore) + { + this.invoker = invoker; + this.workflowDefinitionStore = workflowDefinitionStore; + this.workflowInstanceStore = workflowInstanceStore; + } + + public async Task TriggerWorkflowAsync(string activityName, Variables arguments, CancellationToken cancellationToken) + { + await StartNewWorkflowsAsync(activityName, arguments, cancellationToken); + await ResumeExistingWorkflowsAsync(activityName, arguments, cancellationToken); + } + + private async Task StartNewWorkflowsAsync(string activityName, Variables arguments, CancellationToken cancellationToken) + { + var workflowDefinitions = await workflowDefinitionStore.GetManyAsync(new WorkflowStartsWithActivity(activityName), cancellationToken); + + foreach (var workflowDefinition in workflowDefinitions) + { + var startActivities = workflowDefinition.Workflow.Activities.Where(x => x.Name == activityName && !workflowDefinition.Workflow.Connections.Select(c => c.Target.Activity).Contains(x)); + + foreach (var activity in startActivities) + { + var workflowContext = await invoker.InvokeAsync(workflowDefinition.Workflow, activity, arguments, cancellationToken); + var workflowInstance = new WorkflowInstance + { + WorkflowDefinitionId = workflowDefinition.Id, + Id = Guid.NewGuid().ToString(), + Workflow = workflowContext.Workflow + }; + + await workflowInstanceStore.AddAsync(workflowInstance, cancellationToken); + } + } + } + + private async Task ResumeExistingWorkflowsAsync(string activityName, Variables arguments, CancellationToken cancellationToken) + { + var workflowInstances = await workflowInstanceStore.GetManyAsync(new WorkflowIsBlockedOnActivity(activityName), cancellationToken); + + foreach (var workflowInstance in workflowInstances) + { + var blockingActivities = workflowInstance.Workflow.BlockingActivities.Where(x => x.Name == activityName).ToList(); + + foreach (var activity in blockingActivities) + { + await invoker.ResumeAsync(workflowInstance.Workflow, activity, arguments, cancellationToken); + await workflowInstanceStore.UpdateAsync(workflowInstance, cancellationToken); + } + } + } + } +} \ No newline at end of file diff --git a/src/Flowsharp.Samples.Console/Flowsharp.Samples.Console.csproj b/src/Flowsharp.Samples.Console/Flowsharp.Samples.Console.csproj index 61641f168..e181ce60c 100644 --- a/src/Flowsharp.Samples.Console/Flowsharp.Samples.Console.csproj +++ b/src/Flowsharp.Samples.Console/Flowsharp.Samples.Console.csproj @@ -8,6 +8,10 @@ + + + + diff --git a/src/Flowsharp.Samples.Console/Program.cs b/src/Flowsharp.Samples.Console/Program.cs index 78cbfbe41..43c2d7980 100644 --- a/src/Flowsharp.Samples.Console/Program.cs +++ b/src/Flowsharp.Samples.Console/Program.cs @@ -2,6 +2,10 @@ using System.Threading.Tasks; using Flowsharp.Expressions; using Flowsharp.Handlers; +using Flowsharp.Persistence; +using Flowsharp.Persistence.InMemory; +using Flowsharp.Runtime; +using Flowsharp.Runtime.Abstractions; using Flowsharp.Samples.Console.Handlers; using Flowsharp.Samples.Console.Programs; using Flowsharp.Serialization; @@ -23,12 +27,13 @@ namespace Flowsharp.Samples.Console .AddSingleton() .AddSingleton() .AddSingleton() - .AddSingleton() + //.AddSingleton() .AddSingleton() .AddSingleton() .AddSingleton() .AddSingleton() .AddSingleton() + .AddSingleton() .AddLogging(logging => logging.AddConsole()); services @@ -37,14 +42,20 @@ namespace Flowsharp.Samples.Console .AddSingleton() .AddSingleton(); + services + .AddSingleton() + .AddSingleton(); + services .AddSingleton() .AddSingleton() - .AddSingleton(); + .AddSingleton() + .AddSingleton(); var serviceProvider = services.BuildServiceProvider(); //var program = serviceProvider.GetRequiredService(); - var program = serviceProvider.GetRequiredService(); + //var program = serviceProvider.GetRequiredService(); + var program = serviceProvider.GetRequiredService(); await program.RunAsync(CancellationToken.None); } diff --git a/src/Flowsharp.Samples.Console/Programs/AdditionWorkflowProgramLongRunning.cs b/src/Flowsharp.Samples.Console/Programs/AdditionWorkflowProgramLongRunning.cs index 57a9810e7..baa515050 100644 --- a/src/Flowsharp.Samples.Console/Programs/AdditionWorkflowProgramLongRunning.cs +++ b/src/Flowsharp.Samples.Console/Programs/AdditionWorkflowProgramLongRunning.cs @@ -39,7 +39,7 @@ namespace Flowsharp.Samples.Console.Programs { var json = serializer.Serialize(workflow); workflow = serializer.Deserialize(json); - var haltedActivity = (ReadLine)workflow.HaltedActivities.Single(); + var haltedActivity = (ReadLine)workflow.BlockingActivities.Single(); var args = new Variables {{ haltedActivity.ArgumentName, System.Console.ReadLine() }}; return await workflowInvoker.ResumeAsync(workflow, haltedActivity, args, cancellationToken); diff --git a/src/Flowsharp.Samples.Console/Programs/FileBasedWorkflowProgram.cs b/src/Flowsharp.Samples.Console/Programs/FileBasedWorkflowProgram.cs index b8e1abfc1..36bcd273c 100644 --- a/src/Flowsharp.Samples.Console/Programs/FileBasedWorkflowProgram.cs +++ b/src/Flowsharp.Samples.Console/Programs/FileBasedWorkflowProgram.cs @@ -40,7 +40,7 @@ namespace Flowsharp.Samples.Console.Programs private async Task ReadAndResumeAsync(Workflow workflow, CancellationToken cancellationToken) { - var haltedActivity = (ReadLine)workflow.HaltedActivities.Single(); + var haltedActivity = (ReadLine)workflow.BlockingActivities.Single(); var args = new Variables {{ haltedActivity.ArgumentName, System.Console.ReadLine() }}; return await workflowInvoker.ResumeAsync(workflow, haltedActivity, args, cancellationToken); } diff --git a/src/Flowsharp.Samples.Console/Programs/WorkflowHostProgram.cs b/src/Flowsharp.Samples.Console/Programs/WorkflowHostProgram.cs new file mode 100644 index 000000000..142285324 --- /dev/null +++ b/src/Flowsharp.Samples.Console/Programs/WorkflowHostProgram.cs @@ -0,0 +1,51 @@ +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Flowsharp.Models; +using Flowsharp.Persistence; +using Flowsharp.Persistence.Models; +using Flowsharp.Runtime.Abstractions; +using Flowsharp.Samples.Console.Activities; +using Flowsharp.Serialization; + +namespace Flowsharp.Samples.Console.Programs +{ + public class WorkflowHostProgram + { + private readonly IWorkflowHost workflowHost; + private readonly IWorkflowSerializer serializer; + private readonly IWorkflowDefinitionStore workflowStore; + + public WorkflowHostProgram(IWorkflowHost workflowHost, IWorkflowSerializer serializer, IWorkflowDefinitionStore workflowStore) + { + this.workflowHost = workflowHost; + this.serializer = serializer; + this.workflowStore = workflowStore; + } + + public async Task RunAsync(CancellationToken cancellationToken) + { + var assembly = typeof(FileBasedWorkflowProgramLongRunning).Assembly; + var resource = assembly.GetManifestResourceStream("Flowsharp.Samples.Console.SampleWorkflow.yaml"); + var resourceReader = new StreamReader(resource); + var data = await resourceReader.ReadToEndAsync(); + var workflow = serializer.Deserialize(data); + var workflowDefinition = new WorkflowDefinition { Id = "1", Workflow = workflow}; + + await workflowStore.AddAsync(workflowDefinition, cancellationToken); + await workflowHost.TriggerWorkflowAsync("WriteLine", Variables.Empty, cancellationToken); + await ReadAndResumeAsync("x", cancellationToken); + await ReadAndResumeAsync("y", cancellationToken); + await ReadAndResumeAsync("tryAgain", cancellationToken); + + System.Console.WriteLine(data); + } + + private async Task ReadAndResumeAsync(string argName, CancellationToken cancellationToken) + { + var args = new Variables { {argName, System.Console.ReadLine()} }; + await workflowHost.TriggerWorkflowAsync("ReadLine", args, cancellationToken); + } + } +} \ No newline at end of file