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

V3 StartAt Activity (#2969)

Browse source 
* Rename RunAt -> StartAt

* Update migrations

* Update Timer.cs

* Update schedulers

* Add samples
This commit is contained in:
Sipke Schoorstra 2022-04-27 12:07:23 +02:00 committed by GitHub
parent 53c60adc71
commit 57230948c2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 245 additions and 105 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
src/modules/Elsa.Modules.Scheduling/Activities/RunAt.cs
View file

@ -1,64 +0,0 @@
using System;
using System.Threading.Tasks;
using Elsa.Attributes;
using Elsa.Models;
using Elsa.Services;
using Microsoft.Extensions.Logging;
namespace Elsa.Modules.Scheduling.Activities;
[Activity("Elsa", "Scheduling", "Delay execution for the specified amount of time.")]
public class RunAt : Activity
{
public RunAt()
{
}
public RunAt(Input<DateTimeOffset> dateTime) => DateTime = dateTime;
public RunAt(Func<ExpressionExecutionContext, DateTimeOffset> dateTime) : this(new Input<DateTimeOffset>(dateTime))
{
}
public RunAt(Func<ExpressionExecutionContext, ValueTask<DateTimeOffset>> dateTime) : this(new Input<DateTimeOffset>(dateTime))
{
}
public RunAt(Func<ValueTask<DateTimeOffset>> dateTime) : this(new Input<DateTimeOffset>(dateTime))
{
}
public RunAt(Func<DateTimeOffset> dateTime) : this(new Input<DateTimeOffset>(dateTime))
{
}
public RunAt(DateTimeOffset dateTime) => DateTime = new Input<DateTimeOffset>(dateTime);
public RunAt(Variable<DateTimeOffset> dateTime) => DateTime = new Input<DateTimeOffset>(dateTime);
[Input] public Input<DateTimeOffset> DateTime { get; set; } = default!;
protected override void Execute(ActivityExecutionContext context)
{
// TODO: Update e.g. ScheduleWorkflows and other places to make sure this activity works correctly when suspending & resuming workflows.
var executeAt = context.ExpressionExecutionContext.Get(DateTime);
var clock = context.ExpressionExecutionContext.GetRequiredService<ISystemClock>();
var now = clock.UtcNow;
var logger = context.GetRequiredService<ILogger<RunAt>>();
if (executeAt <= now)
{
logger.LogDebug("Scheduled trigger time lies in the past ('{Delta}'). Skipping scheduling", now - executeAt);
context.JournalData.Add("Executed At", now);
return;
}
var payload = new RunAtPayload(executeAt);
context.CreateBookmark(payload);
}
public static RunAt From(DateTimeOffset value) => new(value);
}
public record RunAtPayload(DateTimeOffset ResumeAt);

69
src/modules/Elsa.Modules.Scheduling/Activities/StartAt.cs Normal file
View file

@ -0,0 +1,69 @@
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using Elsa.Attributes;
using Elsa.Models;
using Elsa.Services;
using Microsoft.Extensions.Logging;
namespace Elsa.Modules.Scheduling.Activities;
[Activity("Elsa", "Scheduling", "Trigger execution at a specific time in the future.")]
public class StartAt : Trigger
{
public StartAt()
{
}
public StartAt(Input<DateTimeOffset> dateTime) => DateTime = dateTime;
public StartAt(Func<ExpressionExecutionContext, DateTimeOffset> dateTime) : this(new Input<DateTimeOffset>(dateTime))
{
}
public StartAt(Func<ExpressionExecutionContext, ValueTask<DateTimeOffset>> dateTime) : this(new Input<DateTimeOffset>(dateTime))
{
}
public StartAt(Func<ValueTask<DateTimeOffset>> dateTime) : this(new Input<DateTimeOffset>(dateTime))
{
}
public StartAt(Func<DateTimeOffset> dateTime) : this(new Input<DateTimeOffset>(dateTime))
{
}
public StartAt(DateTimeOffset dateTime) => DateTime = new Input<DateTimeOffset>(dateTime);
public StartAt(Variable<DateTimeOffset> dateTime) => DateTime = new Input<DateTimeOffset>(dateTime);
[Input] public Input<DateTimeOffset> DateTime { get; set; } = default!;
protected override object GetTriggerDatum(TriggerIndexingContext context)
{
var executeAt = context.ExpressionExecutionContext.Get(DateTime);
return new StartAtPayload(executeAt);
}
protected override void Execute(ActivityExecutionContext context)
{
var executeAt = context.ExpressionExecutionContext.Get(DateTime);
var clock = context.ExpressionExecutionContext.GetRequiredService<ISystemClock>();
var now = clock.UtcNow;
var logger = context.GetRequiredService<ILogger<StartAt>>();
if (executeAt <= now)
{
logger.LogDebug("Scheduled trigger time lies in the past ('{Delta}'). Skipping scheduling", now - executeAt);
context.JournalData.Add("Executed At", now);
return;
}
var payload = new StartAtPayload(executeAt);
context.CreateBookmark(payload);
}
public static StartAt From(DateTimeOffset value) => new(value);
}
public record StartAtPayload(DateTimeOffset ExecuteAt);

4
src/modules/Elsa.Modules.Scheduling/Activities/Timer.cs
View file

@ -29,12 +29,12 @@ public class Timer : EventGenerator
[Input] public Input<TimeSpan> Interval { get; set; } = default!;
protected override IEnumerable<object> GetTriggerData(TriggerIndexingContext context)
protected override object GetTriggerDatum(TriggerIndexingContext context)
{
var interval = context.ExpressionExecutionContext.Get(Interval);
var clock = context.ExpressionExecutionContext.GetRequiredService<ISystemClock>();
var executeAt = clock.UtcNow.Add(interval);
yield return new TimerPayload(executeAt, interval);
return new TimerPayload(executeAt, interval);
}
public static Timer FromTimeSpan(TimeSpan value) => new(value);

22
src/modules/Elsa.Modules.Scheduling/Implementations/WorkflowBookmarkScheduler.cs
View file

@ -29,9 +29,13 @@ public class WorkflowBookmarkScheduler : IWorkflowBookmarkScheduler
// Select all Delay bookmarks.
var delayBookmarks = bookmarkList.Filter<Delay>().ToList();
// Select all StartAt bookmarks.
var startAtBookmarks = bookmarkList.Filter<StartAt>().ToList();
var groupKeys = new[] { RootGroupKey, workflowInstanceId };
// Schedule a trigger for each bookmark.
// Schedule a trigger for each Delay bookmark.
foreach (var bookmark in delayBookmarks)
{
var payload = JsonSerializer.Deserialize<DelayPayload>(bookmark.Data!)!;
@ -40,16 +44,26 @@ public class WorkflowBookmarkScheduler : IWorkflowBookmarkScheduler
var schedule = new SpecificInstantSchedule(resumeAt);
await _jobScheduler.ScheduleAsync(job, bookmark.Id, schedule, groupKeys, cancellationToken);
}
// Schedule a trigger for each StartAt bookmark.
foreach (var bookmark in startAtBookmarks)
{
var payload = JsonSerializer.Deserialize<StartAtPayload>(bookmark.Data!)!;
var executeAt = payload.ExecuteAt;
var job = new ResumeWorkflowJob(workflowInstanceId, bookmark.ToBookmark());
var schedule = new SpecificInstantSchedule(executeAt);
await _jobScheduler.ScheduleAsync(job, bookmark.Id, schedule, groupKeys, cancellationToken);
}
}
public async Task UnscheduleBookmarksAsync(string workflowInstanceId, IEnumerable<WorkflowBookmark> bookmarks, CancellationToken cancellationToken = default)
{
var bookmarkList = bookmarks.ToList();
var delayBookmarks = bookmarkList.Filter<Delay>().ToList();
var startAtBookmarks = bookmarkList.Filter<StartAt>().ToList();
var bookmarksToUnSchedule = delayBookmarks.Concat(startAtBookmarks).ToList();
foreach (var bookmark in delayBookmarks)
{
foreach (var bookmark in bookmarksToUnSchedule)
await _jobScheduler.UnscheduleAsync(bookmark.Id, cancellationToken);
}
}
}

31
src/modules/Elsa.Modules.Scheduling/Implementations/WorkflowTriggerScheduler.cs
View file

@ -27,26 +27,45 @@ public class WorkflowTriggerScheduler : IWorkflowTriggerScheduler
public async Task ScheduleTriggersAsync(IEnumerable<WorkflowTrigger> triggers, CancellationToken cancellationToken = default)
{
// Select all Timer triggers.
var timerTriggers = triggers.Filter<Timer>().ToList();
var triggerList = triggers.ToList();
// Schedule each trigger.
// Select all Timer triggers.
var timerTriggers = triggerList.Filter<Timer>().ToList();
var startAtTriggers = triggerList.Filter<StartAt>().ToList();
// Schedule each Timer trigger.
foreach (var trigger in timerTriggers)
{
// Schedule trigger.
var (dateTime, timeSpan) = JsonSerializer.Deserialize<TimerPayload>(trigger.Data!)!;
var groupKeys = new[] { RootGroupKey, trigger.WorkflowDefinitionId };
await _jobScheduler.ScheduleAsync(new RunWorkflowJob(trigger.WorkflowDefinitionId), trigger.WorkflowDefinitionId, new RecurringSchedule(dateTime, timeSpan), groupKeys, cancellationToken);
}
// Schedule each StartAt trigger.
foreach (var trigger in startAtTriggers)
{
var executeAt = JsonSerializer.Deserialize<StartAtPayload>(trigger.Data!)!.ExecuteAt;
var groupKeys = new[] { RootGroupKey, trigger.WorkflowDefinitionId };
await _jobScheduler.ScheduleAsync(new RunWorkflowJob(trigger.WorkflowDefinitionId), trigger.WorkflowDefinitionId, new SpecificInstantSchedule(executeAt), groupKeys, cancellationToken);
}
}
public async Task UnscheduleTriggersAsync(IEnumerable<WorkflowTrigger> triggers, CancellationToken cancellationToken = default)
{
var triggerList = triggers.ToList();
// Select all Timer triggers.
var timerTriggers = triggers.Filter<Timer>().ToList();
var timerTriggers = triggerList.Filter<Timer>().ToList();
// Select all StartAt triggers.
var startAtTriggers = triggerList.Filter<Timer>().ToList();
// Unschedule all triggers for the distinct set of affected workflows.
var workflowDefinitionIds = timerTriggers.Select(x => x.WorkflowDefinitionId).Distinct().ToList();
var workflowDefinitionIds = timerTriggers
.Select(x => x.WorkflowDefinitionId)
.Concat(startAtTriggers.Select(x => x.WorkflowDefinitionId))
.Distinct()
.ToList();
foreach (var workflowDefinitionId in workflowDefinitionIds)
{

33
src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220309231832_Initial.Designer.cs → src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220427095656_Initial.Designer.cs generated
View file

@ -11,13 +11,13 @@ using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
{
[DbContext(typeof(ElsaDbContext))]
[Migration("20220309231832_Initial")]
[Migration("20220427095656_Initial")]
partial class Initial
{
protected override void BuildTargetModel(ModelBuilder modelBuilder)
{
#pragma warning disable 612, 618
modelBuilder.HasAnnotation("ProductVersion", "6.0.1");
modelBuilder.HasAnnotation("ProductVersion", "6.0.3");
modelBuilder.Entity("Elsa.Persistence.Entities.WorkflowBookmark", b =>
{
@ -192,7 +192,6 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
.HasColumnType("TEXT");
b.Property<string>("CorrelationId")
.IsRequired()
.HasColumnType("TEXT");
b.Property<DateTimeOffset>("CreatedAt")
@ -221,10 +220,13 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
b.Property<string>("Name")
.HasColumnType("TEXT");
b.Property<int>("Version")
b.Property<int>("Status")
.HasColumnType("INTEGER");
b.Property<int>("WorkflowStatus")
b.Property<int>("SubStatus")
.HasColumnType("INTEGER");
b.Property<int>("Version")
.HasColumnType("INTEGER");
b.HasKey("Id");
@ -250,14 +252,23 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
b.HasIndex("Name")
.HasDatabaseName("IX_WorkflowInstance_Name");
b.HasIndex("WorkflowStatus")
.HasDatabaseName("IX_WorkflowInstance_WorkflowStatus");
b.HasIndex("Status")
.HasDatabaseName("IX_WorkflowInstance_Status");
b.HasIndex("WorkflowStatus", "DefinitionId")
.HasDatabaseName("IX_WorkflowInstance_WorkflowStatus_DefinitionId");
b.HasIndex("SubStatus")
.HasDatabaseName("IX_WorkflowInstance_SubStatus");
b.HasIndex("WorkflowStatus", "DefinitionId", "Version")
.HasDatabaseName("IX_WorkflowInstance_WorkflowStatus_DefinitionId_Version");
b.HasIndex("Status", "DefinitionId")
.HasDatabaseName("IX_WorkflowInstance_Status_DefinitionId");
b.HasIndex("Status", "SubStatus")
.HasDatabaseName("IX_WorkflowInstance_Status_SubStatus");
b.HasIndex("SubStatus", "DefinitionId")
.HasDatabaseName("IX_WorkflowInstance_SubStatus_DefinitionId");
b.HasIndex("Status", "SubStatus", "DefinitionId", "Version")
.HasDatabaseName("IX_WorkflowInstance_Status_SubStatus_DefinitionId_Version");
b.ToTable("WorkflowInstances");
});

32
src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220309231832_Initial.cs → src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220427095656_Initial.cs
View file

@ -75,8 +75,9 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
DefinitionId = table.Column<string>(type: "TEXT", nullable: false),
DefinitionVersionId = table.Column<string>(type: "TEXT", nullable: false),
Version = table.Column<int>(type: "INTEGER", nullable: false),
WorkflowStatus = table.Column<int>(type: "INTEGER", nullable: false),
CorrelationId = table.Column<string>(type: "TEXT", nullable: false),
Status = table.Column<int>(type: "INTEGER", nullable: false),
SubStatus = table.Column<int>(type: "INTEGER", nullable: false),
CorrelationId = table.Column<string>(type: "TEXT", nullable: true),
Name = table.Column<string>(type: "TEXT", nullable: true),
CreatedAt = table.Column<DateTimeOffset>(type: "TEXT", nullable: false),
LastExecutedAt = table.Column<DateTimeOffset>(type: "TEXT", nullable: true),
@ -222,19 +223,34 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
column: "Name");
migrationBuilder.CreateIndex(
name: "IX_WorkflowInstance_WorkflowStatus",
name: "IX_WorkflowInstance_Status",
table: "WorkflowInstances",
column: "WorkflowStatus");
column: "Status");
migrationBuilder.CreateIndex(
name: "IX_WorkflowInstance_WorkflowStatus_DefinitionId",
name: "IX_WorkflowInstance_Status_DefinitionId",
table: "WorkflowInstances",
columns: new[] { "WorkflowStatus", "DefinitionId" });
columns: new[] { "Status", "DefinitionId" });
migrationBuilder.CreateIndex(
name: "IX_WorkflowInstance_WorkflowStatus_DefinitionId_Version",
name: "IX_WorkflowInstance_Status_SubStatus",
table: "WorkflowInstances",
columns: new[] { "WorkflowStatus", "DefinitionId", "Version" });
columns: new[] { "Status", "SubStatus" });
migrationBuilder.CreateIndex(
name: "IX_WorkflowInstance_Status_SubStatus_DefinitionId_Version",
table: "WorkflowInstances",
columns: new[] { "Status", "SubStatus", "DefinitionId", "Version" });
migrationBuilder.CreateIndex(
name: "IX_WorkflowInstance_SubStatus",
table: "WorkflowInstances",
column: "SubStatus");
migrationBuilder.CreateIndex(
name: "IX_WorkflowInstance_SubStatus_DefinitionId",
table: "WorkflowInstances",
columns: new[] { "SubStatus", "DefinitionId" });
migrationBuilder.CreateIndex(
name: "IX_WorkflowTrigger_Hash",

31
src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/ElsaDbContextModelSnapshot.cs
View file

@ -15,7 +15,7 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
protected override void BuildModel(ModelBuilder modelBuilder)
{
#pragma warning disable 612, 618
modelBuilder.HasAnnotation("ProductVersion", "6.0.1");
modelBuilder.HasAnnotation("ProductVersion", "6.0.3");
modelBuilder.Entity("Elsa.Persistence.Entities.WorkflowBookmark", b =>
{
@ -190,7 +190,6 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
.HasColumnType("TEXT");
b.Property<string>("CorrelationId")
.IsRequired()
.HasColumnType("TEXT");
b.Property<DateTimeOffset>("CreatedAt")
@ -219,10 +218,13 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
b.Property<string>("Name")
.HasColumnType("TEXT");
b.Property<int>("Version")
b.Property<int>("Status")
.HasColumnType("INTEGER");
b.Property<int>("WorkflowStatus")
b.Property<int>("SubStatus")
.HasColumnType("INTEGER");
b.Property<int>("Version")
.HasColumnType("INTEGER");
b.HasKey("Id");
@ -248,14 +250,23 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations
b.HasIndex("Name")
.HasDatabaseName("IX_WorkflowInstance_Name");
b.HasIndex("WorkflowStatus")
.HasDatabaseName("IX_WorkflowInstance_WorkflowStatus");
b.HasIndex("Status")
.HasDatabaseName("IX_WorkflowInstance_Status");
b.HasIndex("WorkflowStatus", "DefinitionId")
.HasDatabaseName("IX_WorkflowInstance_WorkflowStatus_DefinitionId");
b.HasIndex("SubStatus")
.HasDatabaseName("IX_WorkflowInstance_SubStatus");
b.HasIndex("WorkflowStatus", "DefinitionId", "Version")
.HasDatabaseName("IX_WorkflowInstance_WorkflowStatus_DefinitionId_Version");
b.HasIndex("Status", "DefinitionId")
.HasDatabaseName("IX_WorkflowInstance_Status_DefinitionId");
b.HasIndex("Status", "SubStatus")
.HasDatabaseName("IX_WorkflowInstance_Status_SubStatus");
b.HasIndex("SubStatus", "DefinitionId")
.HasDatabaseName("IX_WorkflowInstance_SubStatus_DefinitionId");
b.HasIndex("Status", "SubStatus", "DefinitionId", "Version")
.HasDatabaseName("IX_WorkflowInstance_Status_SubStatus_DefinitionId_Version");
b.ToTable("WorkflowInstances");
});

4
src/samples/aspnet/Elsa.Samples.Web1/Program.cs
View file

@ -20,6 +20,8 @@ using Elsa.Modules.Scheduling.Extensions;
using Elsa.Modules.WorkflowContexts.Extensions;
using Elsa.Persistence.EntityFrameworkCore.Extensions;
using Elsa.Persistence.EntityFrameworkCore.Sqlite;
using Elsa.Pipelines.ActivityExecution;
using Elsa.Pipelines.ActivityExecution.Components;
using Elsa.Pipelines.WorkflowExecution.Components;
using Elsa.Runtime.Extensions;
using Elsa.Runtime.ProtoActor.Extensions;
@ -70,6 +72,8 @@ services
options.Workflows.Add<SubmitJobWorkflow>();
options.Workflows.Add<DelayWorkflow>();
options.Workflows.Add<OrderProcessingWorkflow>();
options.Workflows.Add<StartAtTriggerWorkflow>();
options.Workflows.Add<StartAtBookmarkWorkflow>();
});
// Testing only: allow client app to connect from anywhere.

29
src/samples/aspnet/Elsa.Samples.Web1/Workflows/StartAtBookmarkWorkflow.cs Normal file
View file

@ -0,0 +1,29 @@
using Elsa.Activities;
using Elsa.Modules.Activities.Console;
using Elsa.Modules.Scheduling.Activities;
using Elsa.Services;
namespace Elsa.Samples.Web1.Workflows;
public class StartAtBookmarkWorkflow : IWorkflow
{
private readonly ISystemClock _systemClock;
public StartAtBookmarkWorkflow(ISystemClock systemClock)
{
_systemClock = systemClock;
}
public void Build(IWorkflowDefinitionBuilder workflow)
{
workflow.WithRoot(new Sequence
{
Activities =
{
new WriteLine("Waiting for 5 seconds..."),
new StartAt(() => _systemClock.UtcNow.AddSeconds(5)),
new WriteLine(() => $"Executed at {_systemClock.UtcNow}")
}
});
}
}

31
src/samples/aspnet/Elsa.Samples.Web1/Workflows/StartAtTriggerWorkflow.cs Normal file
View file

@ -0,0 +1,31 @@
using System;
using Elsa.Activities;
using Elsa.Modules.Activities.Console;
using Elsa.Modules.Scheduling.Activities;
using Elsa.Services;
namespace Elsa.Samples.Web1.Workflows;
public class StartAtTriggerWorkflow : IWorkflow
{
private readonly ISystemClock _systemClock;
private readonly DateTimeOffset _executeAt;
public StartAtTriggerWorkflow(ISystemClock systemClock)
{
_systemClock = systemClock;
_executeAt = systemClock.UtcNow.AddSeconds(10);
}
public void Build(IWorkflowDefinitionBuilder workflow)
{
workflow.WithRoot(new Sequence
{
Activities =
{
new StartAt(_executeAt) { CanStartWorkflow = true },
new WriteLine(() => $"Executed at {_systemClock.UtcNow}")
}
});
}
}