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

Incremental work on long running workflows

Browse source
This commit is contained in:
Sipke Schoorstra 2018-10-14 17:16:33 +02:00
parent 24fea8a176
commit 97a91c4235
16 changed files with 286 additions and 125 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
src/Flowsharp.Abstractions/Activities/Workflow.cs
View file

@ -1,26 +1,31 @@
using System.Collections.Generic;
using System.Linq;
using Flowsharp.ActivityResults;
using Flowsharp.Models;
namespace Flowsharp.Activities
{
public class Workflow
{
public Workflow()
{
}
public Workflow(IEnumerable<IActivity> activities, IEnumerable<Connection> connections)
public Workflow(IEnumerable<IActivity> activities, IEnumerable<Connection> connections) : this()
{
Activities = activities.ToList();
Connections = connections.ToList();
Scopes = new Stack<WorkflowExecutionScope>();
}
public Workflow()
{
CurrentScope = new WorkflowExecutionScope();
Scopes = new Stack<WorkflowExecutionScope>(new[]{ CurrentScope });
Arguments = new Variables();
HaltedActivities = new List<IActivity>();
}
public WorkflowStatus Status { get; set; }
public IList<IActivity> Activities { get; set; } = new List<IActivity>();
public IList<Connection> Connections { get; set; } = new List<Connection>();
public Stack<WorkflowExecutionScope> Scopes { get; set; }
public WorkflowExecutionScope CurrentScope { get; set; }
public Variables Arguments { get; set; }
public IList<IActivity> HaltedActivities { get; set; }
}
}

2
src/Flowsharp.Abstractions/ActivityResults/ReturnValueResult.cs
View file

@ -13,7 +13,7 @@ namespace Flowsharp.ActivityResults
protected override void Execute(WorkflowExecutionContext workflowContext)
{
workflowContext.SetReturnValue(value);
workflowContext.SetLastResult(value);
}
}
}

3
src/Flowsharp.Abstractions/Extensions/ConnectionActivityExtensions.cs
View file

@ -1,6 +1,3 @@
using Flowsharp.Activities;
using Flowsharp.Models;
namespace Flowsharp.Extensions
{
public static class ConnectionExtensions

1
src/Flowsharp.Abstractions/Extensions/InvokeExtensions.cs
View file

@ -1,7 +1,6 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Flowsharp.Activities;
using Microsoft.Extensions.Logging;
namespace Flowsharp.Extensions

22
src/Flowsharp.Abstractions/Models/WorkflowExecutionContext.cs
View file

@ -1,5 +1,4 @@
using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
@ -10,20 +9,16 @@ namespace Flowsharp.Models
{
public class WorkflowExecutionContext
{
public WorkflowExecutionContext(Workflow workflow, WorkflowStatus status = WorkflowStatus.Idle)
public WorkflowExecutionContext(Workflow workflow)
{
Workflow = workflow;
Status = status;
IsFirstPass = true;
scheduledActivities = new Stack<IActivity>();
BeginScope();
}
private readonly Stack<IActivity> scheduledActivities;
public Workflow Workflow { get; }
public WorkflowStatus Status { get; set; }
public bool HasScheduledActivities => scheduledActivities.Any();
public bool IsFirstPass { get; set; }
public IActivity CurrentActivity { get; private set; }
@ -55,9 +50,9 @@ namespace Flowsharp.Models
return CurrentActivity;
}
public void SetReturnValue(object value)
public void SetLastResult(object value)
{
CurrentScope.ReturnValue = value;
CurrentScope.LastResult = value;
}
public void Fault(Exception exception, IActivity activity)
@ -67,12 +62,19 @@ namespace Flowsharp.Models
public Task HaltAsync(CancellationToken cancellationToken)
{
throw new NotImplementedException();
var activity = CurrentActivity;
if (!Workflow.HaltedActivities.Contains(activity))
{
Workflow.HaltedActivities.Add(activity);
}
Workflow.Status = WorkflowStatus.Halted;
return Task.CompletedTask;
}
public void Finish()
{
Status = WorkflowStatus.Finished;
Workflow.Status = WorkflowStatus.Finished;
}
public virtual void ScheduleNextActivities(WorkflowExecutionContext workflowContext, SourceEndpoint endpoint)

13
src/Flowsharp.Abstractions/Models/WorkflowExecutionScope.cs
View file

@ -1,4 +1,5 @@
using System.Collections.Generic;
using Newtonsoft.Json;
namespace Flowsharp.Models
{
@ -6,11 +7,11 @@ namespace Flowsharp.Models
{
public WorkflowExecutionScope()
{
Variables = new Dictionary<string, object>();
Variables = new Variables();
}
public object ReturnValue { get; set; }
public IDictionary<string, object> Variables { get; }
public object LastResult { get; set; }
public Variables Variables { get; }
public void SetVariable(string variableName, object value)
{
@ -22,4 +23,10 @@ namespace Flowsharp.Models
return Variables.ContainsKey(name) ? (T)Variables[name] : default(T);
}
}
[JsonDictionary(ItemTypeNameHandling = TypeNameHandling.None)]
public class Variables : Dictionary<string, object>
{
}
}

30
src/Flowsharp.Core/Activities/ReadLine.cs
View file

@ -1,30 +0,0 @@
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Flowsharp.ActivityResults;
using Flowsharp.Models;
namespace Flowsharp.Activities
{
public class ReadLine : Activity
{
private readonly TextReader input;
public ReadLine() : this(Console.In)
{
}
public ReadLine(TextReader input)
{
this.input = input;
}
public override async Task<ActivityExecutionResult> ExecuteAsync(WorkflowExecutionContext workflowContext, ActivityExecutionContext activityContext, CancellationToken cancellationToken)
{
var value = await input.ReadLineAsync();
workflowContext.SetReturnValue(value);
return ActivateEndpoint();
}
}
}

17
src/Flowsharp.Core/Serialization/JsonWorkflowSerializer.cs
View file

@ -8,15 +8,22 @@ namespace Flowsharp.Serialization
{
public class JsonWorkflowSerializer : IWorkflowSerializer
{
private readonly JsonSerializerSettings settings = new JsonSerializerSettings
{
PreserveReferencesHandling = PreserveReferencesHandling.Objects,
TypeNameHandling = TypeNameHandling.Objects
};
public Task<string> SerializeAsync(Workflow workflow, CancellationToken cancellationToken)
{
var settings = new JsonSerializerSettings
{
PreserveReferencesHandling = PreserveReferencesHandling.Objects,
TypeNameHandling = TypeNameHandling.Objects
};
var json = JsonConvert.SerializeObject(workflow, settings);
return Task.FromResult(json);
}
public Task<Workflow> DeserializeAsync(string json, CancellationToken none)
{
var workflow = JsonConvert.DeserializeObject<Workflow>(json, settings);
return Task.FromResult(workflow);
}
}
}

14
src/Flowsharp.Core/Services/WorkflowInvoker.cs
View file

@ -1,5 +1,4 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
@ -23,12 +22,14 @@ namespace Flowsharp.Services
public async Task<WorkflowExecutionContext> InvokeAsync(Workflow workflow, IActivity startActivity = default, CancellationToken cancellationToken = default)
{
var workflowExecutionContext = new WorkflowExecutionContext(workflow);
var isResuming = workflowExecutionContext.Status == WorkflowStatus.Resuming;
var isResuming = workflowExecutionContext.Workflow.Status == WorkflowStatus.Resuming;
if (startActivity == null)
if (startActivity != null)
workflow.HaltedActivities.Remove(startActivity);
else
startActivity = workflow.Activities.First();
workflowExecutionContext.Status = WorkflowStatus.Executing;
workflowExecutionContext.Workflow.Status = WorkflowStatus.Executing;
workflowExecutionContext.ScheduleActivity(startActivity);
await InvokeActivitiesAsync(workflowExecutionContext, x => x.WorkflowStartingAsync(workflowExecutionContext, cancellationToken));
@ -46,7 +47,8 @@ namespace Flowsharp.Services
isResuming = false;
}
workflowExecutionContext.Status = WorkflowStatus.Finished;
if(workflowExecutionContext.Workflow.Status != WorkflowStatus.Halted)
workflowExecutionContext.Finish();
return workflowExecutionContext;
}
@ -59,7 +61,7 @@ namespace Flowsharp.Services
if (cancellationToken.IsCancellationRequested)
{
workflowContext.Status = WorkflowStatus.Aborted;
workflowContext.Workflow.Status = WorkflowStatus.Aborted;
return null;
}

49
src/Flowsharp.Samples.Console/Activities/ReadLine.cs Normal file
View file

@ -0,0 +1,49 @@
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Flowsharp.Activities;
using Flowsharp.ActivityResults;
using Flowsharp.Models;
namespace Flowsharp.Samples.Console.Activities
{
public class ReadLine : Activity
{
private readonly string argumentName;
private readonly TextReader input;
public ReadLine() : this(System.Console.In)
{
}
public ReadLine(TextReader input)
{
this.input = input;
}
public ReadLine(string argumentName)
{
this.argumentName = argumentName;
input = null;
}
public override async Task<ActivityExecutionResult> ExecuteAsync(WorkflowExecutionContext workflowContext, ActivityExecutionContext activityContext, CancellationToken cancellationToken)
{
if (input != null)
{
var value = await input.ReadLineAsync();
workflowContext.SetLastResult(value);
return ActivateEndpoint();
}
return Halt();
}
protected override ActivityExecutionResult Resume(WorkflowExecutionContext workflowContext, ActivityExecutionContext activityContext)
{
var receivedInput = workflowContext.Workflow.Arguments[argumentName];
workflowContext.SetLastResult(receivedInput);
return ActivateEndpoint();
}
}
}

10
src/Flowsharp.Core/Activities/WriteLine.cs → src/Flowsharp.Samples.Console/Activities/WriteLine.cs
View file

@ -1,12 +1,12 @@
using System;
using System.Diagnostics;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Flowsharp.ActivityResults;
using Flowsharp.Models;
using Activity = Flowsharp.Activities.Activity;
namespace Flowsharp.Activities
namespace Flowsharp.Samples.Console.Activities
{
/// <summary>
/// Writes a text string to the specified stream.
@ -16,15 +16,15 @@ namespace Flowsharp.Activities
private readonly TextWriter output;
private readonly Func<WorkflowExecutionContext, ActivityExecutionContext, string> textProvider;
public WriteLine() : this(Console.Out, null)
public WriteLine() : this(System.Console.Out, null)
{
}
public WriteLine(string text) : this(Console.Out, text)
public WriteLine(string text) : this(System.Console.Out, text)
{
}
public WriteLine(Func<WorkflowExecutionContext, ActivityExecutionContext, string> textProvider) : this(Console.Out, null)
public WriteLine(Func<WorkflowExecutionContext, ActivityExecutionContext, string> textProvider) : this(System.Console.Out, null)
{
this.textProvider = textProvider;
}

58
src/Flowsharp.Samples.Console/Program.cs
View file

@ -1,12 +1,6 @@
using System;
using System.Threading;
using System.Threading;
using System.Threading.Tasks;
using Flowsharp.Activities;
using Flowsharp.Models;
using Flowsharp.Serialization;
using Flowsharp.Services;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Flowsharp.Samples.Console.Programs;
namespace Flowsharp.Samples.Console
{
@ -14,52 +8,8 @@ namespace Flowsharp.Samples.Console
{
static async Task Main()
{
var invoker = new WorkflowInvoker(new Logger<WorkflowInvoker>(new NullLoggerFactory()));
var writeLine1 = new WriteLine("You have now transitioned into a networked workflow.");
var writeLine2 = new WriteLine("Let's run a program.");
var writeLine3 = new WriteLine("Enter first value:");
var readLine1 = new ReadLine();
var setVariable1 = new SetVariable("x", (w, a) => int.Parse((string) w.CurrentScope.ReturnValue));
var writeLine4 = new WriteLine("Enter second value:");
var readLine2 = new ReadLine();
var setVariable2 = new SetVariable("y", (w, a) => int.Parse((string) w.CurrentScope.ReturnValue));
var writeLine5 = new WriteLine((w, a) =>
{
var x = w.CurrentScope.GetVariable<int>("x");
var y = w.CurrentScope.GetVariable<int>("y");
var z = x + y;
return $"{x} + {y} = {z}";
});
var writeLine6 = new WriteLine("Try again? (Y/N)");
var readLine3 = new ReadLine();
var setVariable3 = new SetVariable("tryAgain", (w, a) => string.Equals("y", (string)w.CurrentScope.ReturnValue, StringComparison.CurrentCultureIgnoreCase));
var ifElse1 = new IfElse((w, a) => w.CurrentScope.GetVariable<bool>("tryAgain"));
var writeLine7 = new WriteLine("Bye!");
var activities = new IActivity[] { writeLine1, writeLine2, writeLine3, writeLine4, writeLine5, writeLine6, readLine1, readLine2, readLine3, setVariable1, setVariable2, setVariable3, ifElse1 };
var connections = new[]
{
new Connection(writeLine1, writeLine2),
new Connection(writeLine2, writeLine3),
new Connection(writeLine3, readLine1),
new Connection(readLine1, setVariable1),
new Connection(setVariable1, writeLine4),
new Connection(writeLine4, readLine2),
new Connection(readLine2, setVariable2),
new Connection(setVariable2, writeLine5),
new Connection(writeLine5, writeLine6),
new Connection(writeLine6, readLine3),
new Connection(readLine3, setVariable3),
new Connection(setVariable3, ifElse1),
new Connection(ifElse1, "True", writeLine3),
new Connection(ifElse1, "False", writeLine7),
};
var workflow = new Workflow(activities, connections);
var workflowContext = await invoker.InvokeAsync(workflow);
var serializer = new JsonWorkflowSerializer();
var json = await serializer.SerializeAsync(workflow, CancellationToken.None);
//await new AdditionWorkflowProgram().RunAsync(CancellationToken.None);
await new AdditionWorkflowProgramLongRunning().RunAsync(CancellationToken.None);
}
}
}

24
src/Flowsharp.Samples.Console/Programs/AdditionWorkflowProgram.cs Normal file
View file

@ -0,0 +1,24 @@
using System.Threading;
using System.Threading.Tasks;
using Flowsharp.Samples.Console.Workflows;
using Flowsharp.Serialization;
using Flowsharp.Services;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
namespace Flowsharp.Samples.Console.Programs
{
public class AdditionWorkflowProgram
{
public async Task RunAsync(CancellationToken cancellationToken)
{
var invoker = new WorkflowInvoker(new Logger<WorkflowInvoker>(new NullLoggerFactory()));
var workflow = new AdditionWorkflow();
var workflowContext = await invoker.InvokeAsync(workflow, null, cancellationToken);
var serializer = new JsonWorkflowSerializer();
var json = await serializer.SerializeAsync(workflow, cancellationToken);
System.Console.WriteLine(json);
}
}
}

43
src/Flowsharp.Samples.Console/Programs/AdditionWorkflowProgramLongRunning.cs Normal file
View file

@ -0,0 +1,43 @@
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Flowsharp.Models;
using Flowsharp.Samples.Console.Workflows;
using Flowsharp.Services;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
namespace Flowsharp.Samples.Console.Programs
{
public class AdditionWorkflowProgramLongRunning
{
private readonly WorkflowInvoker workflowInvoker;
public AdditionWorkflowProgramLongRunning()
{
workflowInvoker = new WorkflowInvoker(new Logger<WorkflowInvoker>(new NullLoggerFactory()));
}
public async Task RunAsync(CancellationToken cancellationToken)
{
var workflow = new AdditionWorkflowLongRunning();
var workflowContext = await workflowInvoker.InvokeAsync(workflow, null, cancellationToken);
while (workflowContext.Workflow.Status == WorkflowStatus.Halted)
{
workflowContext = await ReadAndResumeAsync(workflowContext, "x", cancellationToken);
workflowContext = await ReadAndResumeAsync(workflowContext, "y", cancellationToken);
workflowContext = await ReadAndResumeAsync(workflowContext, "tryAgain", cancellationToken);
}
}
private async Task<WorkflowExecutionContext> ReadAndResumeAsync(WorkflowExecutionContext workflowContext, string argumentName, CancellationToken cancellationToken)
{
var workflow = workflowContext.Workflow;
var haltedActivity = workflowContext.Workflow.HaltedActivities.Single();
workflow.Arguments[argumentName] = System.Console.ReadLine();
workflow.Status = WorkflowStatus.Resuming;
return await workflowInvoker.InvokeAsync(workflow, haltedActivity, cancellationToken);
}
}
}

53
src/Flowsharp.Samples.Console/Workflows/AdditionWorkflow.cs Normal file
View file

@ -0,0 +1,53 @@
using System;
using Flowsharp.Activities;
using Flowsharp.Models;
using Flowsharp.Samples.Console.Activities;
namespace Flowsharp.Samples.Console.Workflows
{
public class AdditionWorkflow : Workflow
{
public AdditionWorkflow()
{
var writeLine1 = new WriteLine("Welcome to Addition Workflow!");
var writeLine2 = new WriteLine("Let's run an interactive program.");
var writeLine3 = new WriteLine("Enter first value:");
var readLine1 = new ReadLine();
var setVariable1 = new SetVariable("x", (w, a) => int.Parse((string) w.CurrentScope.LastResult));
var writeLine4 = new WriteLine("Enter second value:");
var readLine2 = new ReadLine();
var setVariable2 = new SetVariable("y", (w, a) => int.Parse((string) w.CurrentScope.LastResult));
var writeLine5 = new WriteLine((w, a) =>
{
var x = w.CurrentScope.GetVariable<int>("x");
var y = w.CurrentScope.GetVariable<int>("y");
var z = x + y;
return $"{x} + {y} = {z}";
});
var writeLine6 = new WriteLine("Try again? (Y/N)");
var readLine3 = new ReadLine();
var setVariable3 = new SetVariable("tryAgain", (w, a) => string.Equals("y", (string)w.CurrentScope.LastResult, StringComparison.CurrentCultureIgnoreCase));
var ifElse1 = new IfElse((w, a) => w.CurrentScope.GetVariable<bool>("tryAgain"));
var writeLine7 = new WriteLine("Bye!");
Activities = new IActivity[] { writeLine1, writeLine2, writeLine3, writeLine4, writeLine5, writeLine6, readLine1, readLine2, readLine3, setVariable1, setVariable2, setVariable3, ifElse1 };
Connections = new[]
{
new Connection(writeLine1, writeLine2),
new Connection(writeLine2, writeLine3),
new Connection(writeLine3, readLine1),
new Connection(readLine1, setVariable1),
new Connection(setVariable1, writeLine4),
new Connection(writeLine4, readLine2),
new Connection(readLine2, setVariable2),
new Connection(setVariable2, writeLine5),
new Connection(writeLine5, writeLine6),
new Connection(writeLine6, readLine3),
new Connection(readLine3, setVariable3),
new Connection(setVariable3, ifElse1),
new Connection(ifElse1, "True", writeLine3),
new Connection(ifElse1, "False", writeLine7),
};
}
}
}

53
src/Flowsharp.Samples.Console/Workflows/AdditionWorkflowLongRunning.cs Normal file
View file

@ -0,0 +1,53 @@
using System;
using Flowsharp.Activities;
using Flowsharp.Models;
using Flowsharp.Samples.Console.Activities;
namespace Flowsharp.Samples.Console.Workflows
{
public class AdditionWorkflowLongRunning : Workflow
{
public AdditionWorkflowLongRunning()
{
var writeLine1 = new WriteLine("Welcome to Addition Workflow - Long Running edition!");
var writeLine2 = new WriteLine("Let's run a long-running program.");
var writeLine3 = new WriteLine("Enter first value:");
var readLine1 = new ReadLine("x");
var setVariable1 = new SetVariable("x", (w, a) => int.Parse((string) w.CurrentScope.LastResult));
var writeLine4 = new WriteLine("Enter second value:");
var readLine2 = new ReadLine("y");
var setVariable2 = new SetVariable("y", (w, a) => int.Parse((string) w.CurrentScope.LastResult));
var writeLine5 = new WriteLine((w, a) =>
{
var x = w.CurrentScope.GetVariable<int>("x");
var y = w.CurrentScope.GetVariable<int>("y");
var z = x + y;
return $"{x} + {y} = {z}";
});
var writeLine6 = new WriteLine("Try again? (Y/N)");
var readLine3 = new ReadLine("tryAgain");
var setVariable3 = new SetVariable("tryAgain", (w, a) => string.Equals("y", (string)w.CurrentScope.LastResult, StringComparison.CurrentCultureIgnoreCase));
var ifElse1 = new IfElse((w, a) => w.CurrentScope.GetVariable<bool>("tryAgain"));
var writeLine7 = new WriteLine("Bye!");
Activities = new IActivity[] { writeLine1, writeLine2, writeLine3, writeLine4, writeLine5, writeLine6, readLine1, readLine2, readLine3, setVariable1, setVariable2, setVariable3, ifElse1 };
Connections = new[]
{
new Connection(writeLine1, writeLine2),
new Connection(writeLine2, writeLine3),
new Connection(writeLine3, readLine1),
new Connection(readLine1, setVariable1),
new Connection(setVariable1, writeLine4),
new Connection(writeLine4, readLine2),
new Connection(readLine2, setVariable2),
new Connection(setVariable2, writeLine5),
new Connection(writeLine5, writeLine6),
new Connection(writeLine6, readLine3),
new Connection(readLine3, setVariable3),
new Connection(setVariable3, ifElse1),
new Connection(ifElse1, "True", writeLine3),
new Connection(ifElse1, "False", writeLine7),
};
}
}
}