diff --git a/src/activities/Elsa.Activities.Scheduling/Activities/Delay.cs b/src/activities/Elsa.Activities.Scheduling/Activities/Delay.cs new file mode 100644 index 000000000..7859e881d --- /dev/null +++ b/src/activities/Elsa.Activities.Scheduling/Activities/Delay.cs @@ -0,0 +1,23 @@ +using System; +using Elsa.Attributes; +using Elsa.Contracts; +using Elsa.Models; + +namespace Elsa.Activities.Scheduling; + +public class Delay : Activity +{ + [Input] public Input TimeSpan { get; set; } = default!; + + protected override void Execute(ActivityExecutionContext context) + { + var timeSpan = context.ExpressionExecutionContext.Get(TimeSpan); + var clock = context.ExpressionExecutionContext.GetRequiredService(); + var resumeAt = clock.UtcNow.Add(timeSpan); + var payload = new DelayPayload(resumeAt); + + context.SetBookmark(payload); + } +} + +public record DelayPayload(DateTime ResumeAt); \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Scheduling/Contracts/IJobScheduler.cs b/src/activities/Elsa.Activities.Scheduling/Contracts/IJobScheduler.cs index d9c35349e..7d6e791a5 100644 --- a/src/activities/Elsa.Activities.Scheduling/Contracts/IJobScheduler.cs +++ b/src/activities/Elsa.Activities.Scheduling/Contracts/IJobScheduler.cs @@ -6,4 +6,5 @@ namespace Elsa.Activities.Scheduling.Contracts; public interface IJobScheduler { Task ScheduleAsync(IJob job, ISchedule schedule, CancellationToken cancellationToken = default); + Task ClearAsync(CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Scheduling/Elsa.Activities.Scheduling.csproj.DotSettings b/src/activities/Elsa.Activities.Scheduling/Elsa.Activities.Scheduling.csproj.DotSettings index 93fae5e65..7867b65ab 100644 --- a/src/activities/Elsa.Activities.Scheduling/Elsa.Activities.Scheduling.csproj.DotSettings +++ b/src/activities/Elsa.Activities.Scheduling/Elsa.Activities.Scheduling.csproj.DotSettings @@ -1,2 +1,3 @@  + True True \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Scheduling/Handlers/ScheduleWorkflowsHandler.cs b/src/activities/Elsa.Activities.Scheduling/Handlers/ScheduleWorkflowsHandler.cs index d7e9522ae..ac20798b9 100644 --- a/src/activities/Elsa.Activities.Scheduling/Handlers/ScheduleWorkflowsHandler.cs +++ b/src/activities/Elsa.Activities.Scheduling/Handlers/ScheduleWorkflowsHandler.cs @@ -1,3 +1,4 @@ +using System.Linq; using System.Text.Json; using System.Threading; using System.Threading.Tasks; @@ -5,6 +6,7 @@ using Elsa.Activities.Scheduling.Contracts; using Elsa.Activities.Scheduling.Jobs; using Elsa.Activities.Scheduling.Schedules; using Elsa.Mediator.Contracts; +using Elsa.Persistence.Extensions; using Elsa.Persistence.Requests; using Elsa.Runtime.Notifications; @@ -23,7 +25,11 @@ public class ScheduleWorkflowsHandler : INotificationHandler(), cancellationToken); + // Unschedule everything. + await _jobScheduler.ClearAsync(cancellationToken); + + // Select all Timer triggers. + var timerTriggers = notification.Triggers.Filter().ToList(); foreach (var trigger in timerTriggers) { diff --git a/src/activities/Elsa.Activities.Scheduling/Triggers/Timer.cs b/src/activities/Elsa.Activities.Scheduling/Triggers/Timer.cs index ce9800540..ac8c0e354 100644 --- a/src/activities/Elsa.Activities.Scheduling/Triggers/Timer.cs +++ b/src/activities/Elsa.Activities.Scheduling/Triggers/Timer.cs @@ -6,6 +6,9 @@ using Elsa.Models; namespace Elsa.Activities.Scheduling; +/// +/// Represents a timer to periodically trigger the workflow it is associated with. +/// public class Timer : Trigger { [Input] public Input Interval { get; set; } = default!; diff --git a/src/core/Elsa.Core/Contracts/IHasher.cs b/src/core/Elsa.Core/Contracts/IHasher.cs index 79c2a3432..1cae3d2bf 100644 --- a/src/core/Elsa.Core/Contracts/IHasher.cs +++ b/src/core/Elsa.Core/Contracts/IHasher.cs @@ -6,4 +6,5 @@ namespace Elsa.Contracts; public interface IHasher { public string Hash(object value); + public string Hash(string value); } \ No newline at end of file diff --git a/src/core/Elsa.Core/Contracts/IPayloadSerializer.cs b/src/core/Elsa.Core/Contracts/IPayloadSerializer.cs new file mode 100644 index 000000000..2b722d239 --- /dev/null +++ b/src/core/Elsa.Core/Contracts/IPayloadSerializer.cs @@ -0,0 +1,7 @@ +namespace Elsa.Contracts; + +public interface IPayloadSerializer +{ + T Deserialize(string json) where T : notnull; + string Serialize(T payload) where T : notnull; +} \ No newline at end of file diff --git a/src/core/Elsa.Core/Models/ActivityExecutionContext.cs b/src/core/Elsa.Core/Models/ActivityExecutionContext.cs index f76b52844..bdda0a21e 100644 --- a/src/core/Elsa.Core/Models/ActivityExecutionContext.cs +++ b/src/core/Elsa.Core/Models/ActivityExecutionContext.cs @@ -47,33 +47,31 @@ public class ActivityExecutionContext ScheduleActivity(activity, completionCallback); } - public void SetBookmarks(IEnumerable hashInputs, IDictionary? data = default, ExecuteActivityDelegate? callback = default) + public void SetBookmarks(IEnumerable payloads, ExecuteActivityDelegate? callback = default) { - foreach (var hashInput in hashInputs) - SetBookmark(hashInput, data, callback); + foreach (var hashInput in payloads) + SetBookmark(payloads, callback); } public void SetBookmarks(IEnumerable bookmarks) => _bookmarks.AddRange(bookmarks); public void SetBookmark(Bookmark bookmark) => _bookmarks.Add(bookmark); - public void SetBookmark(object? hashInput, IDictionary? data = default, ExecuteActivityDelegate? callback = default) + public void SetBookmark(object? payload, ExecuteActivityDelegate? callback = default) { var hasher = GetRequiredService(); - var hash = hashInput != null ? hasher.Hash(hashInput) : default; - SetBookmark(hash, data, callback); - } - - public void SetBookmark(string? hash, IDictionary? data = default, ExecuteActivityDelegate? callback = default) - { + var identityGenerator = GetRequiredService(); + var payloadSerializer = GetRequiredService(); + var payloadJson = payload != null ? payloadSerializer.Serialize(payload) : default; + var hash = payloadJson != null ? hasher.Hash(payloadJson) : default; SetBookmark(new Bookmark( identityGenerator.GenerateId(), Activity.NodeType, hash, + payloadJson, Activity.Id, Id, - data ?? new Dictionary(), callback?.Method.Name)); } @@ -112,7 +110,7 @@ public class ActivityExecutionContext var locationReference = input.LocationReference; var value = await evaluator.EvaluateAsync(input, ExpressionExecutionContext); locationReference.Set(this, value); - return (T?)value; + return value; } private RegisterLocation? GetLocation(RegisterLocationReference locationReference) => diff --git a/src/core/Elsa.Core/Models/Bookmark.cs b/src/core/Elsa.Core/Models/Bookmark.cs index 41b16909e..0e9a194b7 100644 --- a/src/core/Elsa.Core/Models/Bookmark.cs +++ b/src/core/Elsa.Core/Models/Bookmark.cs @@ -4,8 +4,8 @@ public record Bookmark( string Id, string Name, string? Hash, + string? Payload, string ActivityId, string ActivityInstanceId, - IDictionary? Data = default, string? CallbackMethodName = default ); \ No newline at end of file diff --git a/src/core/Elsa.Core/Services/Hasher.cs b/src/core/Elsa.Core/Services/Hasher.cs index 4a726c291..36b56c29d 100644 --- a/src/core/Elsa.Core/Services/Hasher.cs +++ b/src/core/Elsa.Core/Services/Hasher.cs @@ -7,24 +7,25 @@ namespace Elsa.Services; public class Hasher : IHasher { + private readonly IPayloadSerializer _payloadSerializer; + + public Hasher(IPayloadSerializer payloadSerializer) + { + _payloadSerializer = payloadSerializer; + } + public string Hash(object value) { - var options = new JsonSerializerOptions - { - // Enables serialization of ValueTuples, which use fields instead of properties. - IncludeFields = true - }; - - var json = JsonSerializer.Serialize(value, options); + var json = _payloadSerializer.Serialize(value); return Hash(json); } - - private static string Hash(string input) + + public string Hash(string value) { using var sha = HashAlgorithm.Create(HashAlgorithmName.SHA256.ToString())!; - return Hash(sha, input); + return Hash(sha, value); } - + private static string Hash(HashAlgorithm hashAlgorithm, string input) { var data = hashAlgorithm.ComputeHash(Encoding.UTF8.GetBytes(input)); diff --git a/src/core/Elsa.Core/Services/PayloadSerializer.cs b/src/core/Elsa.Core/Services/PayloadSerializer.cs new file mode 100644 index 000000000..9be60c233 --- /dev/null +++ b/src/core/Elsa.Core/Services/PayloadSerializer.cs @@ -0,0 +1,21 @@ +using System.Text.Json; +using Elsa.Contracts; + +namespace Elsa.Services; + +public class PayloadSerializer : IPayloadSerializer +{ + private readonly JsonSerializerOptions _settings; + + public PayloadSerializer() + { + _settings = new JsonSerializerOptions + { + // Enables serialization of ValueTuples, which use fields instead of properties. + IncludeFields = true + }; + } + + public T Deserialize(string json) where T : notnull => JsonSerializer.Deserialize(json, _settings)!; + public string Serialize(T payload) where T : notnull => JsonSerializer.Serialize(payload, payload.GetType(), _settings); +} \ No newline at end of file diff --git a/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs b/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs index 2cbc48d3a..d9981dda9 100644 --- a/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs +++ b/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs @@ -43,6 +43,7 @@ public static class ServiceCollectionExtensions .AddSingleton() .AddSingleton() .AddSingleton() + .AddSingleton() // DSL. .AddDsl() diff --git a/src/modules/Elsa.Modules.Quartz/Services/QuartzJobScheduler.cs b/src/modules/Elsa.Modules.Quartz/Services/QuartzJobScheduler.cs index 0cef31dec..a64c33c97 100644 --- a/src/modules/Elsa.Modules.Quartz/Services/QuartzJobScheduler.cs +++ b/src/modules/Elsa.Modules.Quartz/Services/QuartzJobScheduler.cs @@ -3,6 +3,7 @@ using Elsa.Modules.Quartz.Contracts; using Elsa.Modules.Quartz.Jobs; using Microsoft.Extensions.Logging; using Quartz; +using Quartz.Impl.Matchers; using IElsaJobScheduler = Elsa.Activities.Scheduling.Contracts.IJobScheduler; using IElsaJob = Elsa.Activities.Scheduling.Contracts.IJob; using IElsaSchedule = Elsa.Activities.Scheduling.Contracts.ISchedule; @@ -12,6 +13,7 @@ namespace Elsa.Modules.Quartz.Services; public class QuartzJobScheduler : IElsaJobScheduler { public const string JobDataKey = "ElsaJob"; + public const string GroupKey = "ElsaJobs"; private readonly IElsaJobSerializer _elsaJobSerializer; private readonly ISchedulerFactory _schedulerFactory; private readonly ILogger _logger; @@ -28,7 +30,14 @@ public class QuartzJobScheduler : IElsaJobScheduler var quartzTrigger = CreateTrigger(job, schedule); await ScheduleJob(quartzTrigger, cancellationToken); } - + + public async Task ClearAsync(CancellationToken cancellationToken = default) + { + var scheduler = await _schedulerFactory.GetScheduler(cancellationToken); + var jobKeys = await scheduler.GetJobKeys(GroupMatcher.GroupEquals(GroupKey), cancellationToken); + await scheduler.DeleteJobs(jobKeys, cancellationToken); + } + private async Task ScheduleJob(ITrigger trigger, CancellationToken cancellationToken) { var scheduler = await _schedulerFactory.GetScheduler(cancellationToken); @@ -47,8 +56,9 @@ public class QuartzJobScheduler : IElsaJobScheduler private ITrigger CreateTrigger(IElsaJob job, IElsaSchedule schedule) { var jobName = job.GetType().Name; + var triggerKey = new TriggerKey(job.JobId, GroupKey); var json = _elsaJobSerializer.Serialize(job); - var builder = TriggerBuilder.Create().ForJob(jobName).WithIdentity(job.JobId).UsingJobData(JobDataKey, json); + var builder = TriggerBuilder.Create().ForJob(jobName).WithIdentity(triggerKey).UsingJobData(JobDataKey, json); switch (schedule) { diff --git a/src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs b/src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs index cf6595c70..813d6ee92 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs +++ b/src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs @@ -3,4 +3,4 @@ using Elsa.Persistence.Entities; namespace Elsa.Persistence.Commands; -public record ReplaceWorkflowTriggers(string WorkflowId, IEnumerable WorkflowTriggers) : ICommand; \ No newline at end of file +public record ReplaceWorkflowTriggers(ICollection WorkflowTriggers) : ICommand; \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Elsa.Persistence.Abstractions.csproj b/src/persistence/Elsa.Persistence.Abstractions/Elsa.Persistence.Abstractions.csproj index 12475cdd4..e255b73f1 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Elsa.Persistence.Abstractions.csproj +++ b/src/persistence/Elsa.Persistence.Abstractions/Elsa.Persistence.Abstractions.csproj @@ -15,5 +15,9 @@ + + + + diff --git a/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs b/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs index 18e74446c..0a41e6e7a 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs +++ b/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs @@ -4,11 +4,11 @@ public class WorkflowBookmark : Entity { public string Name { get; init; } = default!; public string? Hash { get; set; } + public string? Payload { get; set; } public string WorkflowDefinitionId { get; init; } = default!; public string WorkflowInstanceId { get; init; } = default!; public string CorrelationId { get; init; } = default!; public string ActivityId { get; init; } = default!; public string ActivityInstanceId { get; init; } = default!; - public IDictionary? Data { get; set; } public string? CallbackMethodName { get; set; } } \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Extensions/WorkflowTriggerExtensions.cs b/src/persistence/Elsa.Persistence.Abstractions/Extensions/WorkflowTriggerExtensions.cs new file mode 100644 index 000000000..4818dc582 --- /dev/null +++ b/src/persistence/Elsa.Persistence.Abstractions/Extensions/WorkflowTriggerExtensions.cs @@ -0,0 +1,14 @@ +using Elsa.Contracts; +using Elsa.Helpers; +using Elsa.Persistence.Entities; + +namespace Elsa.Persistence.Extensions; + +public static class WorkflowTriggerExtensions +{ + public static IEnumerable Filter(this IEnumerable triggers) where T : ITrigger + { + var triggerName = TypeNameHelper.GenerateTypeName(); + return triggers.Where(x => x.Name == triggerName); + } +} \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowInstanceMiddleware.cs b/src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowInstanceMiddleware.cs index a055a25c5..0d01a3dc1 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowInstanceMiddleware.cs +++ b/src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowInstanceMiddleware.cs @@ -22,6 +22,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware private readonly IRequestSender _requestSender; private readonly ICommandSender _commandSender; private readonly IWorkflowStateSerializer _workflowStateSerializer; + private readonly IPayloadSerializer _payloadSerializer; private readonly IIdentityGenerator _identityGenerator; private readonly ISystemClock _clock; @@ -30,6 +31,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware IRequestSender requestSender, ICommandSender commandSender, IWorkflowStateSerializer workflowStateSerializer, + IPayloadSerializer payloadSerializer, IIdentityGenerator identityGenerator, ISystemClock clock) { @@ -37,6 +39,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware _requestSender = requestSender; _commandSender = commandSender; _workflowStateSerializer = workflowStateSerializer; + _payloadSerializer = payloadSerializer; _identityGenerator = identityGenerator; _clock = clock; } @@ -47,7 +50,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware var workflow = context.Workflow; var (definitionId, version, definitionVersionId) = workflow.Identity; var existingWorkflowInstance = await _requestSender.RequestAsync(new FindWorkflowInstance(context.Id), cancellationToken); - + // Setup a new workflow instance if no existing instance was found. var workflowInstance = existingWorkflowInstance ?? new WorkflowInstance { @@ -63,7 +66,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware // Get a copy of current bookmarks. var existingBookmarks = await _requestSender.RequestAsync(new FindWorkflowBookmarks(workflowInstance.Id), cancellationToken); - var bookmarksSnapshot = existingBookmarks.Select(x => new Bookmark(x.Id, x.Name, x.Hash, x.ActivityId, x.ActivityInstanceId, x.Data, x.CallbackMethodName)).ToList(); + var bookmarksSnapshot = existingBookmarks.Select(x => new Bookmark(x.Id, x.Name, x.Hash, x.Payload, x.ActivityId, x.ActivityInstanceId, x.CallbackMethodName)).ToList(); // Exclude the current bookmark that initiated the creation of the workflow context, if any. var invokedBookmark = context.Bookmark; @@ -103,7 +106,7 @@ public class PersistWorkflowInstanceMiddleware : IWorkflowExecutionMiddleware WorkflowInstanceId = workflowInstance.Id, CorrelationId = workflowInstance.CorrelationId, Hash = x.Hash, - Data = x.Data, + Payload = x.Payload, Name = x.Name, ActivityId = x.ActivityId, ActivityInstanceId = x.ActivityInstanceId, diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220114122659_Initial.Designer.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220117205849_Initial.Designer.cs similarity index 99% rename from src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220114122659_Initial.Designer.cs rename to src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220117205849_Initial.Designer.cs index 87bd83347..ff433b558 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220114122659_Initial.Designer.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220117205849_Initial.Designer.cs @@ -11,7 +11,7 @@ using Microsoft.EntityFrameworkCore.Storage.ValueConversion; namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations { [DbContext(typeof(ElsaDbContext))] - [Migration("20220114122659_Initial")] + [Migration("20220117205849_Initial")] partial class Initial { protected override void BuildTargetModel(ModelBuilder modelBuilder) @@ -42,13 +42,13 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations b.Property("Hash") .HasColumnType("TEXT"); - b.Property("JsonData") - .HasColumnType("TEXT"); - b.Property("Name") .IsRequired() .HasColumnType("TEXT"); + b.Property("Payload") + .HasColumnType("TEXT"); + b.Property("WorkflowDefinitionId") .IsRequired() .HasColumnType("TEXT"); diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220114122659_Initial.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220117205849_Initial.cs similarity index 99% rename from src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220114122659_Initial.cs rename to src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220117205849_Initial.cs index 3343ff2d3..ec91c2971 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220114122659_Initial.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/20220117205849_Initial.cs @@ -16,13 +16,13 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations Id = table.Column(type: "TEXT", nullable: false), Name = table.Column(type: "TEXT", nullable: false), Hash = table.Column(type: "TEXT", nullable: true), + Payload = table.Column(type: "TEXT", nullable: true), WorkflowDefinitionId = table.Column(type: "TEXT", nullable: false), WorkflowInstanceId = table.Column(type: "TEXT", nullable: false), CorrelationId = table.Column(type: "TEXT", nullable: false), ActivityId = table.Column(type: "TEXT", nullable: false), ActivityInstanceId = table.Column(type: "TEXT", nullable: false), - CallbackMethodName = table.Column(type: "TEXT", nullable: true), - JsonData = table.Column(type: "TEXT", nullable: true) + CallbackMethodName = table.Column(type: "TEXT", nullable: true) }, constraints: table => { diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/ElsaDbContextModelSnapshot.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/ElsaDbContextModelSnapshot.cs index 60dc9e1c3..f84321fbc 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/ElsaDbContextModelSnapshot.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore.Sqlite/Migrations/ElsaDbContextModelSnapshot.cs @@ -40,13 +40,13 @@ namespace Elsa.Persistence.EntityFrameworkCore.Sqlite.Migrations b.Property("Hash") .HasColumnType("TEXT"); - b.Property("JsonData") - .HasColumnType("TEXT"); - b.Property("Name") .IsRequired() .HasColumnType("TEXT"); + b.Property("Payload") + .HasColumnType("TEXT"); + b.Property("WorkflowDefinitionId") .IsRequired() .HasColumnType("TEXT"); diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Configuration/WorkflowBookmarkConfiguration.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Configuration/WorkflowBookmarkConfiguration.cs index 854746a9d..e595a3fd4 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Configuration/WorkflowBookmarkConfiguration.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Configuration/WorkflowBookmarkConfiguration.cs @@ -8,9 +8,6 @@ namespace Elsa.Persistence.EntityFrameworkCore.Configuration { public void Configure(EntityTypeBuilder builder) { - builder.Ignore(x => x.Data); - builder.Property("JsonData"); - builder.HasIndex(x => x.Name).HasDatabaseName($"IX_{nameof(WorkflowBookmark)}_{nameof(WorkflowBookmark.Name)}"); builder.HasIndex(x => x.Hash).HasDatabaseName($"IX_{nameof(WorkflowBookmark)}_{nameof(WorkflowBookmark.Hash)}"); builder.HasIndex(x => x.ActivityId).HasDatabaseName($"IX_{nameof(WorkflowBookmark)}_{nameof(WorkflowBookmark.ActivityId)}"); diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs index 3155d4303..f074a283f 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs @@ -12,7 +12,8 @@ public class ReplaceWorkflowTriggersHandler : ICommandHandler HandleAsync(ReplaceWorkflowTriggers command, CancellationToken cancellationToken) { - await _store.DeleteWhereAsync(x => x.WorkflowDefinitionId == command.WorkflowId, cancellationToken); + var ids = command.WorkflowTriggers.Select(x => x.Id).ToList(); + await _store.DeleteWhereAsync(x => ids.Contains(x.Id), cancellationToken); await _store.SaveManyAsync(command.WorkflowTriggers, cancellationToken); return Unit.Instance; diff --git a/src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs b/src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs index 71c03acde..756aaf405 100644 --- a/src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs +++ b/src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs @@ -1,3 +1,4 @@ +using System.Linq; using System.Threading; using System.Threading.Tasks; using Elsa.Mediator.Contracts; @@ -18,9 +19,10 @@ public class ReplaceWorkflowTriggersHandler : ICommandHandler HandleAsync(ReplaceWorkflowTriggers command, CancellationToken cancellationToken) { - _store.DeleteWhere(x => x.WorkflowDefinitionId == command.WorkflowId); + var ids = command.WorkflowTriggers.Select(x => x.Id).ToList(); + _store.DeleteWhere(x => ids.Contains(x.Id)); _store.SaveMany(command.WorkflowTriggers); - + return Task.FromResult(Unit.Instance); } } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowOperatorActor.cs b/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowOperatorActor.cs index c5829aed5..ecb4c8c64 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowOperatorActor.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Actors/WorkflowOperatorActor.cs @@ -65,20 +65,15 @@ public class WorkflowOperatorActor : IActor private ExecuteWorkflowResponse MapResult(ExecuteWorkflowResult result) { - var bookmarks = result.Bookmarks.Select(x => + var bookmarks = result.Bookmarks.Select(x => new Bookmark { - var data = x.Data?.ToDictionary(y => y.Key, y => new Json - { - Text = JsonSerializer.Serialize(y.Value) - }) ?? new Dictionary(); - - var bookmark = new Bookmark - { - Id = x.Id - }; - - bookmark.Data.Add(data); - return bookmark; + Id = x.Id, + Hash = x.Hash, + Payload = x.Payload, + Name = x.Name, + ActivityId = x.ActivityId, + ActivityInstanceId = x.ActivityInstanceId, + CallbackMethodName = x.CallbackMethodName }); var response = new ExecuteWorkflowResponse @@ -104,9 +99,9 @@ public class WorkflowOperatorActor : IActor bookmarkMessage.Id, bookmarkMessage.Name, bookmarkMessage.Hash, + bookmarkMessage.Payload, bookmarkMessage.ActivityId, bookmarkMessage.ActivityInstanceId, - null, bookmarkMessage.CallbackMethodName); return await _workflowEngine.ExecuteAsync(workflow, workflowState, bookmark, cancellationToken); diff --git a/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs b/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs index f5dfa13df..411d6f713 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs +++ b/src/runtime/Elsa.Runtime.ProtoActor/Services/ProtoActorWorkflowInvoker.cs @@ -13,7 +13,7 @@ namespace Elsa.Runtime.ProtoActor.Services; using Bookmark = Elsa.Models.Bookmark; using BookmarkMessage = Messages.Bookmark; - + public class ProtoActorWorkflowInvoker : IWorkflowInvoker { private readonly Cluster _cluster; @@ -56,7 +56,7 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker return new ExecuteWorkflowResult(workflowState, bookmarks); } - + public async Task DispatchAsync(DispatchWorkflowInstanceRequest request, CancellationToken cancellationToken = default) { var (instanceId, bookmark) = request; @@ -92,7 +92,7 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker return new ExecuteWorkflowResult(workflowState, bookmarks); } - + private BookmarkMessage? MapBookmark(Bookmark? bookmark) { if (bookmark == null) @@ -103,10 +103,10 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker Id = bookmark.Id, Name = bookmark.Name, Hash = bookmark.Hash, + Payload = bookmark.Payload, ActivityId = bookmark.ActivityId, ActivityInstanceId = bookmark.ActivityInstanceId, CallbackMethodName = bookmark.CallbackMethodName, - Data = { } }; } @@ -116,9 +116,9 @@ public class ProtoActorWorkflowInvoker : IWorkflowInvoker bookmarkMessage.Id, bookmarkMessage.Name, bookmarkMessage.Hash, + bookmarkMessage.Payload, bookmarkMessage.ActivityId, bookmarkMessage.ActivityInstanceId, - bookmarkMessage.Data.ToDictionary(x => x.Key, x => JsonSerializer.Deserialize(x.Value.Text)), bookmarkMessage.CallbackMethodName ); } diff --git a/src/runtime/Elsa.Runtime.ProtoActor/messages.proto b/src/runtime/Elsa.Runtime.ProtoActor/messages.proto index 0d49b83e3..936b1f742 100644 --- a/src/runtime/Elsa.Runtime.ProtoActor/messages.proto +++ b/src/runtime/Elsa.Runtime.ProtoActor/messages.proto @@ -43,9 +43,9 @@ message Bookmark { string id = 1; string name = 2; optional string hash = 3; - string activityId = 4; - string activityInstanceId = 5; - map data = 6; + optional string payload = 4; + string activityId = 5; + string activityInstanceId = 6; optional string callbackMethodName = 7; } diff --git a/src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs b/src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs index 65f66f77a..9bae1fa90 100644 --- a/src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs +++ b/src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs @@ -1,4 +1,5 @@ using Elsa.Models; +using Elsa.Persistence.Entities; namespace Elsa.Runtime.Contracts; @@ -12,5 +13,5 @@ public interface ITriggerIndexer /// /// Indexes triggers of the specified workflow. /// - Task IndexTriggersAsync(Workflow workflow, CancellationToken cancellationToken = default); + Task> IndexTriggersAsync(Workflow workflow, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs b/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs index 6504c5b3c..37657e053 100644 --- a/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs +++ b/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs @@ -50,7 +50,7 @@ public class ResumeWorkflowInstructionInterpreter : WorkflowInstructionInterpret } // Resume workflow instance. - var bookmark = new Bookmark(workflowBookmark.Id, workflowBookmark.Name, workflowBookmark.Hash, workflowBookmark.ActivityId, workflowBookmark.ActivityInstanceId, workflowBookmark.Data, workflowBookmark.CallbackMethodName); + var bookmark = new Bookmark(workflowBookmark.Id, workflowBookmark.Name, workflowBookmark.Hash, workflowBookmark.Payload, workflowBookmark.ActivityId, workflowBookmark.ActivityInstanceId, workflowBookmark.CallbackMethodName); var workflowState = workflowInstance.WorkflowState; var workflowExecutionResult = await _workflowEngine.ExecuteAsync(workflow, workflowState, bookmark, cancellationToken); diff --git a/src/runtime/Elsa.Runtime/Notifications/TriggerIndexingFinished.cs b/src/runtime/Elsa.Runtime/Notifications/TriggerIndexingFinished.cs index 7187eaf3f..a762c7f59 100644 --- a/src/runtime/Elsa.Runtime/Notifications/TriggerIndexingFinished.cs +++ b/src/runtime/Elsa.Runtime/Notifications/TriggerIndexingFinished.cs @@ -1,7 +1,6 @@ using Elsa.Mediator.Contracts; +using Elsa.Persistence.Entities; namespace Elsa.Runtime.Notifications; -public class TriggerIndexingFinished : INotification -{ -} \ No newline at end of file +public record TriggerIndexingFinished(ICollection Triggers) : INotification; \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Services/TriggerIndexer.cs b/src/runtime/Elsa.Runtime/Services/TriggerIndexer.cs index 295bda71f..50fda59ce 100644 --- a/src/runtime/Elsa.Runtime/Services/TriggerIndexer.cs +++ b/src/runtime/Elsa.Runtime/Services/TriggerIndexer.cs @@ -58,24 +58,37 @@ public class TriggerIndexer : ITriggerIndexer _logger.LogInformation("Indexing workflow triggers"); stopwatch.Start(); + // Only stream workflows from providers that are not "dynamic" (such as DatabaseWorkflowProvider). var workflows = _workflowRegistry.StreamAllAsync(WorkflowRegistry.SkipDynamicProviders, cancellationToken); + var collectedTriggers = new List(); await foreach (var workflow in workflows.WithCancellation(cancellationToken)) - await IndexTriggersAsync(workflow, cancellationToken); + { + var triggers = await GetTriggersAsync(workflow, cancellationToken).ToListAsync(cancellationToken); + collectedTriggers.AddRange(triggers); + } + + // Replace triggers for the specified workflow. + await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(collectedTriggers), cancellationToken); stopwatch.Stop(); _logger.LogInformation("Finished indexing workflow triggers in {ElapsedTime}", stopwatch.Elapsed); - await _eventPublisher.PublishAsync(new TriggerIndexingFinished(), cancellationToken); + + // Publish event. + await _eventPublisher.PublishAsync(new TriggerIndexingFinished(collectedTriggers), cancellationToken); } - public async Task IndexTriggersAsync(Workflow workflow, CancellationToken cancellationToken = default) + public async Task> IndexTriggersAsync(Workflow workflow, CancellationToken cancellationToken = default) { // Collect new triggers. var triggers = await GetTriggersAsync(workflow, cancellationToken).ToListAsync(cancellationToken); // Replace triggers for the specified workflow. - var definitionId = workflow.Identity.DefinitionId; - await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(definitionId, triggers), cancellationToken); + await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(triggers), cancellationToken); + + // Publish event. + await _eventPublisher.PublishAsync(new TriggerIndexingFinished(triggers), cancellationToken); + return triggers; } private async IAsyncEnumerable GetTriggersAsync(Workflow workflow, [EnumeratorCancellation] CancellationToken cancellationToken = default) @@ -132,11 +145,11 @@ public class TriggerIndexer : ITriggerIndexer return triggers; } - private async Task> TryGetPayloadsAsync(ITrigger trigger, TriggerIndexingContext context, CancellationToken cancellationToken) + private async Task> TryGetPayloadsAsync(ITrigger trigger, TriggerIndexingContext context, CancellationToken cancellationToken) { try { - return await trigger.GetPayloadsAsync(context, cancellationToken); + return (await trigger.GetPayloadsAsync(context, cancellationToken)).ToList(); } catch (Exception e) {