using Elsa.Common.Models;
using Elsa.Extensions;
using Elsa.Scheduling.Activities;
using Elsa.Scheduling.Contracts;
using Elsa.Workflows.Runtime.Entities;
using Elsa.Workflows.Runtime.Models.Requests;
namespace Elsa.Scheduling.Services;
///
/// A default implementation of that schedules triggers using .
///
public class DefaultTriggerScheduler : ITriggerScheduler
{
private readonly IWorkflowScheduler _workflowScheduler;
///
/// Initializes a new instance of the class.
///
public DefaultTriggerScheduler(IWorkflowScheduler workflowScheduler)
{
_workflowScheduler = workflowScheduler;
}
///
public async Task ScheduleAsync(IEnumerable triggers, CancellationToken cancellationToken = default)
{
var triggerList = triggers.ToList();
// Select Timer, StartAt and Cron triggers.
var timerTriggers = triggerList.Filter().ToList();
var startAtTriggers = triggerList.Filter().ToList();
var cronTriggers = triggerList.Filter().ToList();
// Schedule each Timer trigger.
foreach (var trigger in timerTriggers)
{
var (startAt, interval) = trigger.GetPayload();
var input = new { StartAt = startAt, Interval = interval }.ToDictionary();
var request = new DispatchWorkflowDefinitionRequest
{
DefinitionId = trigger.WorkflowDefinitionId,
VersionOptions = VersionOptions.Published,
TriggerActivityId = trigger.ActivityId,
Input = input
};
await _workflowScheduler.ScheduleRecurringAsync(trigger.Id, request, startAt, interval, cancellationToken);
}
// Schedule each StartAt trigger.
foreach (var trigger in startAtTriggers)
{
var executeAt = trigger.GetPayload().ExecuteAt;
var input = new { ExecuteAt = executeAt }.ToDictionary();
var request = new DispatchWorkflowDefinitionRequest
{
DefinitionId = trigger.WorkflowDefinitionId,
VersionOptions = VersionOptions.Published,
TriggerActivityId = trigger.ActivityId,
Input = input
};
await _workflowScheduler.ScheduleAtAsync(trigger.Id, request, executeAt, cancellationToken);
}
// Schedule each Cron trigger.
foreach (var trigger in cronTriggers)
{
var payload = trigger.GetPayload();
var cronExpression = payload.CronExpression;
var input = new { CronExpression = cronExpression }.ToDictionary();
var request = new DispatchWorkflowDefinitionRequest
{
DefinitionId = trigger.WorkflowDefinitionId,
VersionOptions = VersionOptions.Published,
TriggerActivityId = trigger.ActivityId,
Input = input
};
await _workflowScheduler.ScheduleCronAsync(trigger.Id, request, cronExpression, cancellationToken);
}
}
///
public async Task UnscheduleAsync(IEnumerable triggers, CancellationToken cancellationToken = default)
{
var triggerList = triggers.ToList();
// Select all Timer triggers.
var timerTriggers = triggerList.Filter().ToList();
// Select all StartAt triggers.
var startAtTriggers = triggerList.Filter().ToList();
// Concatenate the filtered triggers.
var filteredTriggers = timerTriggers.Concat(startAtTriggers).ToList();
// Unschedule each trigger.
foreach (var trigger in filteredTriggers)
await _workflowScheduler.UnscheduleAsync(trigger.Id, cancellationToken);
}
}