Refactor Delay activity

Enables the workflow to execute without suspending
This commit is contained in:
Sipke Schoorstra 2022-11-10 16:00:03 +01:00
parent a781463f7e
commit 3be2e128d8
2 changed files with 69 additions and 3 deletions

View file

@ -1,5 +1,5 @@
using System;
using Elsa.Common.Services;
using Elsa.Common.Services;
using Elsa.Scheduling.Models;
using Elsa.Workflows.Core;
using Elsa.Workflows.Core.Attributes;
using Elsa.Workflows.Core.Models;
@ -18,16 +18,58 @@ public class Delay : Activity
public Delay(Variable<TimeSpan> timeSpan) => TimeSpan = new Input<TimeSpan>(timeSpan);
[Input] public Input<TimeSpan> TimeSpan { get; set; } = default!;
[Input] public Input<DelayBlockingStrategy> Strategy { get; set; } = default!;
/// <summary>
/// The threshold used by the <see cref="DelayBlockingStrategy.Auto"/>
/// </summary>
[Input] public Input<TimeSpan> AutoBlockingThreshold { get; set; } = new(System.TimeSpan.FromSeconds(5));
protected override void Execute(ActivityExecutionContext context)
/// <inheritdoc />
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var timeSpan = context.ExpressionExecutionContext.Get(TimeSpan);
var blockingMode = context.Get(Strategy);
switch (blockingMode)
{
case DelayBlockingStrategy.Blocking:
await BlockingStrategy(timeSpan);
break;
case DelayBlockingStrategy.NonBlocking:
await NonBlockingStrategy(timeSpan, context);
break;
case DelayBlockingStrategy.Auto:
await AutoBlockingStrategy(timeSpan, context);
break;
default:
throw new ArgumentOutOfRangeException();
}
}
private async ValueTask BlockingStrategy(TimeSpan timeSpan) => await Task.Delay(timeSpan);
private ValueTask NonBlockingStrategy(TimeSpan timeSpan, ActivityExecutionContext context)
{
var clock = context.ExpressionExecutionContext.GetRequiredService<ISystemClock>();
var resumeAt = clock.UtcNow.Add(timeSpan);
var payload = new DelayPayload(resumeAt);
context.JournalData.Add("ResumeAt", resumeAt);
context.CreateBookmark(payload);
return ValueTask.CompletedTask;
}
private async ValueTask AutoBlockingStrategy(TimeSpan timeSpan, ActivityExecutionContext context)
{
var threshold = context.Get(AutoBlockingThreshold);
if (timeSpan <= threshold)
await BlockingStrategy(timeSpan);
else
await NonBlockingStrategy(timeSpan, context);
}
public static Delay FromMilliseconds(double value) => new(System.TimeSpan.FromMilliseconds(value));

View file

@ -0,0 +1,24 @@
using Elsa.Scheduling.Activities;
namespace Elsa.Scheduling.Models;
/// <summary>
/// Represents an execution mode for <see cref="Delay"/>.
/// </summary>
public enum DelayBlockingStrategy
{
/// <summary>
/// Do not suspend the workflow, but instead wait
/// </summary>
NonBlocking,
/// <summary>
/// Suspend the workflow and schedule a background timer to resume the workflow.
/// </summary>
Blocking,
/// <summary>
/// If the delay is less than a configurable amount of time, behaves as <see cref="Blocking"/>, otherwise as <see cref="NonBlocking"/>.
/// </summary>
Auto
}