diff --git a/src/api/Elsa.Api/Endpoints/Events/Trigger.cs b/src/api/Elsa.Api/Endpoints/Events/Trigger.cs index 1b1db978a..a9e1bab45 100644 --- a/src/api/Elsa.Api/Endpoints/Events/Trigger.cs +++ b/src/api/Elsa.Api/Endpoints/Events/Trigger.cs @@ -11,7 +11,7 @@ public static class Trigger public static IResult Handle(string eventName, IHasher hasher) { var hash = hasher.Hash(eventName); - var stimulus = Stimuli.Standard(nameof(Event), hash); + var stimulus = Stimulus.Standard(nameof(Event), hash); return new ProcessStimulusResult(stimulus); } } \ No newline at end of file diff --git a/src/core/Elsa.Core/Elsa.Core.csproj b/src/core/Elsa.Core/Elsa.Core.csproj index 006e4dfd0..1c4a8ecc5 100644 --- a/src/core/Elsa.Core/Elsa.Core.csproj +++ b/src/core/Elsa.Core/Elsa.Core.csproj @@ -15,8 +15,4 @@ - - - - diff --git a/src/core/Elsa.Core/Helpers/Diff.cs b/src/core/Elsa.Core/Helpers/Diff.cs new file mode 100644 index 000000000..3c736f644 --- /dev/null +++ b/src/core/Elsa.Core/Helpers/Diff.cs @@ -0,0 +1,30 @@ +namespace Elsa.Helpers; + +/// +/// A simple utility that compares two collections and returns a set of added, removed and unchanged items. +/// +public class Diff +{ + internal Diff(ICollection added, ICollection removed, ICollection unchanged) + { + Added = added; + Removed = removed; + Unchanged = unchanged; + } + + public ICollection Added { get; } + public ICollection Removed { get; } + public ICollection Unchanged { get; } +} + +public static class Diff +{ + public static Diff For(ICollection firstSet, ICollection secondSet) + { + var added = firstSet.Except(secondSet).ToList(); + var removed = secondSet.Except(firstSet).ToList(); + var unchanged = firstSet.Intersect(secondSet).ToList(); + + return new Diff(added, removed, unchanged); + } +} \ No newline at end of file diff --git a/src/core/Elsa.Core/Models/TriggerActivity.cs b/src/core/Elsa.Core/Models/TriggerActivity.cs index 45d32fadd..a1cd59272 100644 --- a/src/core/Elsa.Core/Models/TriggerActivity.cs +++ b/src/core/Elsa.Core/Models/TriggerActivity.cs @@ -18,11 +18,24 @@ public class TriggerActivity : Activity, ITrigger set => NodeType = value; } - public virtual ValueTask> GetPayloadsAsync(TriggerIndexingContext context, CancellationToken cancellationToken = default) + ValueTask> ITrigger.GetPayloadsAsync(TriggerIndexingContext context, CancellationToken cancellationToken) => GetPayloadsAsync(context, cancellationToken); + + /// + /// Override this method to return a list of trigger payloads. + /// + protected virtual ValueTask> GetPayloadsAsync(TriggerIndexingContext context, CancellationToken cancellationToken = default) { - var hashes = GetHashInputs(context); + var hashes = GetPayloads(context); return ValueTask.FromResult(hashes); } - protected virtual IEnumerable GetHashInputs(TriggerIndexingContext context) => Enumerable.Empty(); + /// + /// Override this method to return a list of trigger payloads. + /// + protected virtual IEnumerable GetPayloads(TriggerIndexingContext context) => new[]{ GetPayload(context) }; + + /// + /// Override this method to return a trigger payloads. + /// + protected virtual object GetPayload(TriggerIndexingContext context) => new(); } \ No newline at end of file diff --git a/src/core/Elsa.Core/Models/TriggerIndexingContext.cs b/src/core/Elsa.Core/Models/TriggerIndexingContext.cs index d42a27cef..de205248a 100644 --- a/src/core/Elsa.Core/Models/TriggerIndexingContext.cs +++ b/src/core/Elsa.Core/Models/TriggerIndexingContext.cs @@ -14,4 +14,6 @@ public class TriggerIndexingContext ExpressionExecutionContext = expressionExecutionContext; Trigger = trigger; } + + public T? Get(Input input) => ExpressionExecutionContext.Get(input); } \ No newline at end of file diff --git a/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs b/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs index 12738dec2..1e705eca1 100644 --- a/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs +++ b/src/core/Elsa/Extensions/ServiceCollectionExtensions.cs @@ -76,5 +76,5 @@ public static class ServiceCollectionExtensions public static IServiceCollection AddStimulusHandler(this IServiceCollection services) where T : class, IStimulusHandler => services.AddSingleton(); public static IServiceCollection AddInstructionInterpreter(this IServiceCollection services) where T : class, IWorkflowInstructionInterpreter => services.AddSingleton(); public static IServiceCollection ConfigureWorkflowRuntime(this IServiceCollection services, Action configure) => services.Configure(configure); - public static IServiceCollection IndexWorkflowTriggers(this IServiceCollection services) => services.AddHostedService(); + public static IServiceCollection IndexWorkflowTriggers(this IServiceCollection services) => services.AddHostedService(); } \ No newline at end of file diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Activities/MessageReceived.cs b/src/modules/Elsa.Modules.AzureServiceBus/Activities/MessageReceived.cs new file mode 100644 index 000000000..7384fc9cc --- /dev/null +++ b/src/modules/Elsa.Modules.AzureServiceBus/Activities/MessageReceived.cs @@ -0,0 +1,19 @@ +using Elsa.Attributes; +using Elsa.Models; +using Elsa.Modules.AzureServiceBus.Models; + +namespace Elsa.Modules.AzureServiceBus.Activities; + +[Activity("Azure.ServiceBus.MessageReceived", "Executes when a message is received from the configured queue or topic and subscription", "Azure Service Bus")] +public class MessageReceived : TriggerActivity +{ + public Input QueueOrTopic { get; set; } = default!; + public Input? Subscription { get; set; } = default!; + + protected override object GetPayload(TriggerIndexingContext context) + { + var queueOrTopic = context.Get(QueueOrTopic)!; + var subscription = context.Get(Subscription!)!; + return new MessageReceivedTriggerPayload(queueOrTopic, subscription); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Activities/Send.cs b/src/modules/Elsa.Modules.AzureServiceBus/Activities/SendMessage.cs similarity index 95% rename from src/modules/Elsa.Modules.AzureServiceBus/Activities/Send.cs rename to src/modules/Elsa.Modules.AzureServiceBus/Activities/SendMessage.cs index 8ebefcab4..406aa7dd5 100644 --- a/src/modules/Elsa.Modules.AzureServiceBus/Activities/Send.cs +++ b/src/modules/Elsa.Modules.AzureServiceBus/Activities/SendMessage.cs @@ -8,7 +8,7 @@ using Microsoft.Extensions.Logging; namespace Elsa.Modules.AzureServiceBus.Activities; [Activity("Azure.ServiceBus.Send", "Send a message to a queue or topic", "Azure Service Bus")] -public class Send : Activity +public class SendMessage : Activity { public Input MessageBody { get; set; } = default!; public Input QueueOrTopic { get; set; } = default!; @@ -55,7 +55,7 @@ public class Send : Activity private static bool ValidatePreconditions(ActivityExecutionContext context, string? queueOrTopic, object? messageBody) { - var logger = context.GetRequiredService>(); + var logger = context.GetRequiredService>(); if (string.IsNullOrWhiteSpace(queueOrTopic)) { diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Contracts/IWorkerManager.cs b/src/modules/Elsa.Modules.AzureServiceBus/Contracts/IWorkerManager.cs new file mode 100644 index 000000000..31491e700 --- /dev/null +++ b/src/modules/Elsa.Modules.AzureServiceBus/Contracts/IWorkerManager.cs @@ -0,0 +1,37 @@ +using Elsa.Modules.AzureServiceBus.Models; +using Elsa.Modules.AzureServiceBus.Services; + +namespace Elsa.Modules.AzureServiceBus.Contracts; + +/// +/// Manages message workers. +/// +public interface IWorkerManager +{ + IReadOnlyCollection Workers { get; } + + /// + /// Returns a value indicating whether a worker exists for the specified trigger. + /// + /// + /// + /// + /// + Worker? FindWorkerFor(string queueOrTopic, string? subscription); + + /// + /// Creates a worker for the specified queue or topic and subscription. + /// + Task StartWorkerAsync(string queueOrTopic, string? subscription, CancellationToken cancellationToken = default); + + /// + /// Decrements the ref count for the worker for the specified queue or topic and subscription. + /// If the worker's ref count reaches 0, it is removed. + /// + Task StopWorkerAsync(string queueOrTopic, string? subscription, CancellationToken cancellationToken = default); + + /// + /// Removes the specified worker. + /// + Task RemoveWorkerAsync(Worker worker); +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Elsa.Modules.AzureServiceBus.csproj b/src/modules/Elsa.Modules.AzureServiceBus/Elsa.Modules.AzureServiceBus.csproj index c56344153..1e5639ae1 100644 --- a/src/modules/Elsa.Modules.AzureServiceBus/Elsa.Modules.AzureServiceBus.csproj +++ b/src/modules/Elsa.Modules.AzureServiceBus/Elsa.Modules.AzureServiceBus.csproj @@ -18,6 +18,8 @@ + + diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs b/src/modules/Elsa.Modules.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs index 722830144..acb07014e 100644 --- a/src/modules/Elsa.Modules.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs +++ b/src/modules/Elsa.Modules.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs @@ -1,6 +1,8 @@ using Azure.Messaging.ServiceBus; using Azure.Messaging.ServiceBus.Administration; +using Elsa.Mediator.Extensions; using Elsa.Modules.AzureServiceBus.Contracts; +using Elsa.Modules.AzureServiceBus.Handlers; using Elsa.Modules.AzureServiceBus.HostedServices; using Elsa.Modules.AzureServiceBus.Options; using Elsa.Modules.AzureServiceBus.Providers; @@ -25,14 +27,22 @@ public static class ServiceCollectionExtensions .AddSingleton(CreateServiceBusManagementClient) .AddSingleton(CreateServiceBusClient) .AddSingleton() - .AddTransient() + .AddSingleton() + .AddTransient(); + + // Definition providers. + services .AddSingleton(sp => sp.GetRequiredService()) .AddSingleton(sp => sp.GetRequiredService()) .AddSingleton(sp => sp.GetRequiredService()); + // Hosted services. if (autoCreateQueuesTopicsAndSubscriptions) services.AddHostedService(); + // Handlers. + services.AddHandlersFrom(); + return services; } diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Handlers/UpdateWorkers.cs b/src/modules/Elsa.Modules.AzureServiceBus/Handlers/UpdateWorkers.cs new file mode 100644 index 000000000..14231b386 --- /dev/null +++ b/src/modules/Elsa.Modules.AzureServiceBus/Handlers/UpdateWorkers.cs @@ -0,0 +1,68 @@ +using Elsa.Contracts; +using Elsa.Mediator.Contracts; +using Elsa.Modules.AzureServiceBus.Activities; +using Elsa.Modules.AzureServiceBus.Contracts; +using Elsa.Modules.AzureServiceBus.Models; +using Elsa.Persistence.Extensions; +using Elsa.Runtime.Models; +using Elsa.Runtime.Notifications; + +namespace Elsa.Modules.AzureServiceBus.Handlers; + +/// +/// Creates workers for each trigger & bookmark in response to updated workflow trigger indexes and bookmarks. +/// +public class UpdateWorkers : INotificationHandler, INotificationHandler, INotificationHandler +{ + private readonly IPayloadSerializer _serializer; + private readonly IWorkerManager _workerManager; + + public UpdateWorkers(IPayloadSerializer serializer, IWorkerManager workerManager) + { + _serializer = serializer; + _workerManager = workerManager; + } + + /// + /// Adds, updates and removes workers based on added and removed triggers. + /// + public async Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken) + { + var added = notification.IndexedWorkflow.AddedTriggers.Filter().Select(x => DeserializePayload(x.Payload!)); + var removed = notification.IndexedWorkflow.RemovedTriggers.Filter().Select(x => DeserializePayload(x.Payload!)); + + await StopWorkersAsync(removed, cancellationToken); + await StartWorkersAsync(added, cancellationToken); + } + + /// + /// Removes workers based on removed bookmarks. + /// + public async Task HandleAsync(WorkflowBookmarksDeleted notification, CancellationToken cancellationToken) + { + var payloads = notification.Bookmarks.Filter().Select(x => DeserializePayload(x.Payload!)); + await StopWorkersAsync(payloads, cancellationToken); + } + + /// + /// Starts workers based on added bookmarks. + /// + public async Task HandleAsync(WorkflowBookmarksSaved notification, CancellationToken cancellationToken) + { + var payloads = notification.Bookmarks.Filter().Select(x => DeserializePayload(x.Payload!)); + await StartWorkersAsync(payloads, cancellationToken); + } + + private MessageReceivedTriggerPayload DeserializePayload(string payload) => _serializer.Deserialize(payload); + private ICollection DeserializePayloads(IEnumerable payloads) => payloads.Select(_serializer.Deserialize).ToList(); + + private async Task StartWorkersAsync(IEnumerable payloads, CancellationToken cancellationToken) + { + foreach (var payload in payloads) await _workerManager.StartWorkerAsync(payload.QueueOrTopic, payload.Subscription, cancellationToken); + } + + private async Task StopWorkersAsync(IEnumerable payloads, CancellationToken cancellationToken) + { + foreach (var payload in payloads) await _workerManager.StopWorkerAsync(payload.QueueOrTopic, payload.Subscription, cancellationToken); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Models/MessageReceivedTriggerPayload.cs b/src/modules/Elsa.Modules.AzureServiceBus/Models/MessageReceivedTriggerPayload.cs new file mode 100644 index 000000000..af2d3350b --- /dev/null +++ b/src/modules/Elsa.Modules.AzureServiceBus/Models/MessageReceivedTriggerPayload.cs @@ -0,0 +1,38 @@ +using System.Text.Json.Serialization; + +namespace Elsa.Modules.AzureServiceBus.Models; + +public record MessageReceivedTriggerPayload +{ + private readonly string _queueOrTopic = default!; + private readonly string? _subscription; + + public MessageReceivedTriggerPayload(string queueOrTopic, string? subscription) + { + QueueOrTopic = queueOrTopic; + Subscription = subscription; + } + + [JsonConstructor] + public MessageReceivedTriggerPayload() + { + } + + public string QueueOrTopic + { + get => _queueOrTopic; + init => _queueOrTopic = value.ToLowerInvariant(); + } + + public string? Subscription + { + get => _subscription; + init => _subscription = value?.ToLowerInvariant(); + } + + public void Deconstruct(out string queueOrTopic, out string? subscription) + { + queueOrTopic = QueueOrTopic; + subscription = Subscription; + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Services/Worker.cs b/src/modules/Elsa.Modules.AzureServiceBus/Services/Worker.cs new file mode 100644 index 000000000..088fa4a82 --- /dev/null +++ b/src/modules/Elsa.Modules.AzureServiceBus/Services/Worker.cs @@ -0,0 +1,80 @@ +using Azure.Messaging.ServiceBus; +using Elsa.Contracts; +using Elsa.Helpers; +using Elsa.Modules.AzureServiceBus.Activities; +using Elsa.Modules.AzureServiceBus.Models; +using Elsa.Runtime.Contracts; +using Elsa.Runtime.Models; +using Microsoft.Extensions.Logging; + +namespace Elsa.Modules.AzureServiceBus.Services; + +/// +/// Processes messages received via a queue specified through the . +/// When a message is received, the appropriate workflows are executed. +/// +public class Worker : IAsyncDisposable +{ + private static readonly string BookmarkName = TypeNameHelper.GenerateTypeName(); + private readonly ServiceBusProcessor _processor; + private readonly IHasher _hasher; + private readonly IWorkflowServer _workflowServer; + private readonly ILogger _logger; + private int _refCount = 1; + + public Worker(string queueOrTopic, string? subscription, ServiceBusClient client, IHasher hasher, IWorkflowServer workflowServer, ILogger logger) + { + QueueOrTopic = queueOrTopic; + Subscription = subscription; + _hasher = hasher; + _workflowServer = workflowServer; + _logger = logger; + + var options = new ServiceBusProcessorOptions(); + var processor = subscription == null ? client.CreateProcessor(queueOrTopic, options) : client.CreateProcessor(queueOrTopic, subscription, options); + + processor.ProcessMessageAsync += OnMessageReceivedAsync; + processor.ProcessErrorAsync += OnErrorAsync; + _processor = processor; + } + + public string QueueOrTopic { get; } + public string? Subscription { get; } + + /// + /// Maintains the number of workflows are relying on this worker. When it goes to zero, the worker will be removed. + /// + public int RefCount + { + get => _refCount; + private set + { + if (value < 0) + throw new ArgumentException("RefCount cannot be less than zero"); + + _refCount = value; + } + } + + public async Task StartAsync(CancellationToken cancellationToken = default) => await _processor.StartProcessingAsync(cancellationToken); + public void IncrementRefCount() => RefCount++; + public void DecrementRefCount() => RefCount--; + public async ValueTask DisposeAsync() => await _processor.DisposeAsync(); + private async Task OnMessageReceivedAsync(ProcessMessageEventArgs args) => await InvokeWorkflowsAsync(args.Message, args.CancellationToken); + + private Task OnErrorAsync(ProcessErrorEventArgs args) + { + _logger.LogError(args.Exception, "An error occurred while processing {EnrityPath}", args.EntityPath); + return Task.CompletedTask; + } + + private async Task InvokeWorkflowsAsync(ServiceBusReceivedMessage message, CancellationToken cancellationToken) + { + var payload = new MessageReceivedTriggerPayload(QueueOrTopic, Subscription); + var hash = _hasher.Hash(payload); + var stimulus = Stimulus.Standard(BookmarkName, hash); + var executionResults = (await _workflowServer.ExecuteStimulusAsync(stimulus, cancellationToken)).ToList(); + + _logger.LogInformation("Triggered {WorkflowCount} workflows", executionResults.Count); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.AzureServiceBus/Services/WorkerManager.cs b/src/modules/Elsa.Modules.AzureServiceBus/Services/WorkerManager.cs new file mode 100644 index 000000000..1c484738f --- /dev/null +++ b/src/modules/Elsa.Modules.AzureServiceBus/Services/WorkerManager.cs @@ -0,0 +1,57 @@ +using Elsa.Modules.AzureServiceBus.Contracts; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Modules.AzureServiceBus.Services; + +public class WorkerManager : IWorkerManager, IAsyncDisposable +{ + private readonly IServiceProvider _serviceProvider; + private readonly ICollection _workers = new List(); + + public WorkerManager(IServiceProvider serviceProvider) + { + _serviceProvider = serviceProvider; + } + + public IReadOnlyCollection Workers => _workers.ToList(); + public Worker? FindWorkerFor(string queueOrTopic, string? subscription) => _workers.FirstOrDefault(x => x.QueueOrTopic == queueOrTopic && x.Subscription == subscription); + + public async Task StartWorkerAsync(string queueOrTopic, string? subscription, CancellationToken cancellationToken = default) + { + var worker = FindWorkerFor(queueOrTopic, subscription); + + if (worker != null) + { + worker.IncrementRefCount(); + return; + } + + worker = ActivatorUtilities.CreateInstance(_serviceProvider, queueOrTopic, subscription!); + + _workers.Add(worker); + await worker.StartAsync(cancellationToken); + } + + public async Task StopWorkerAsync(string queueOrTopic, string? subscription, CancellationToken cancellationToken = default) + { + var worker = FindWorkerFor(queueOrTopic, subscription); + + if (worker == null) + return; + + worker.DecrementRefCount(); + + if (worker.RefCount == 0) await RemoveWorkerAsync(worker); + } + + public async Task RemoveWorkerAsync(Worker worker) + { + _workers.Remove(worker); + await worker.DisposeAsync(); + } + + public async ValueTask DisposeAsync() + { + foreach (var worker in Workers) await worker.DisposeAsync(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.Http/Activities/HttpTrigger.cs b/src/modules/Elsa.Modules.Http/Activities/HttpTrigger.cs index 57aeb0f60..400cef624 100644 --- a/src/modules/Elsa.Modules.Http/Activities/HttpTrigger.cs +++ b/src/modules/Elsa.Modules.Http/Activities/HttpTrigger.cs @@ -4,6 +4,7 @@ using System.Net.Http; using Elsa.Attributes; using Elsa.Management.Models; using Elsa.Models; +using Elsa.Modules.Http.Models; namespace Elsa.Modules.Http; @@ -19,7 +20,7 @@ public class HttpTrigger : TriggerActivity [Output] public Output? Result { get; set; } - protected override IEnumerable GetHashInputs(TriggerIndexingContext context) => GetHashInputs(context.ExpressionExecutionContext); + protected override IEnumerable GetPayloads(TriggerIndexingContext context) => GetHashInputs(context.ExpressionExecutionContext); protected override void Execute(ActivityExecutionContext context) => context.SetBookmarks(GetHashInputs(context.ExpressionExecutionContext)); private IEnumerable GetHashInputs(ExpressionExecutionContext context) @@ -27,6 +28,6 @@ public class HttpTrigger : TriggerActivity // Generate a bookmark hash for path and selected methods. var path = context.Get(Path); var methods = context.Get(SupportedMethods); - return methods!.Select(x => (path!.ToLowerInvariant(), x.ToLowerInvariant())).Cast().ToArray(); + return methods!.Select(x => new HttpTriggerPayload(path!, x.ToLowerInvariant())).Cast().ToArray(); } } \ No newline at end of file diff --git a/src/modules/Elsa.Modules.Http/Middleware/HttpTriggerMiddleware.cs b/src/modules/Elsa.Modules.Http/Middleware/HttpTriggerMiddleware.cs index d98dbe883..fbcf15ae0 100644 --- a/src/modules/Elsa.Modules.Http/Middleware/HttpTriggerMiddleware.cs +++ b/src/modules/Elsa.Modules.Http/Middleware/HttpTriggerMiddleware.cs @@ -3,6 +3,7 @@ using System.Text.Json; using System.Threading.Tasks; using Elsa.Contracts; using Elsa.Helpers; +using Elsa.Modules.Http.Models; using Elsa.Runtime.Contracts; using Elsa.Runtime.Models; using Microsoft.AspNetCore.Http; @@ -25,11 +26,11 @@ public class HttpTriggerMiddleware var path = GetPath(httpContext); var method = httpContext.Request.Method!.ToLowerInvariant(); var abortToken = httpContext.RequestAborted; - var hash = _hasher.Hash((path.ToLowerInvariant(), method.ToLowerInvariant())); + var hash = _hasher.Hash(new HttpTriggerPayload(path, method)); var activityTypeName = TypeNameHelper.GenerateTypeName(); - var stimulus = Stimuli.Standard(activityTypeName, hash); + var stimulus = Stimulus.Standard(activityTypeName, hash); var executionResults = (await workflowServer.ExecuteStimulusAsync(stimulus, abortToken)).ToList(); - + if (!executionResults.Any()) { await _next(httpContext); @@ -42,12 +43,12 @@ public class HttpTriggerMiddleware { response.ContentType = "application/json"; response.StatusCode = StatusCodes.Status200OK; - + var model = new { workflowInstanceIds = executionResults.Select(x => x.ExecuteWorkflowResult.WorkflowState.Id).ToArray() }; - + var json = JsonSerializer.Serialize(model); await response.WriteAsync(json, abortToken); } diff --git a/src/modules/Elsa.Modules.Http/Models/HttpTriggerPayload.cs b/src/modules/Elsa.Modules.Http/Models/HttpTriggerPayload.cs new file mode 100644 index 000000000..2d6e2c90a --- /dev/null +++ b/src/modules/Elsa.Modules.Http/Models/HttpTriggerPayload.cs @@ -0,0 +1,25 @@ +namespace Elsa.Modules.Http.Models; + +public record HttpTriggerPayload +{ + private readonly string _path = default!; + private readonly string _method = default!; + + public HttpTriggerPayload(string path, string method) + { + Path = path; + Method = method; + } + + public string Path + { + get => _path; + init => _path = value.ToLowerInvariant(); + } + + public string Method + { + get => _method; + init => _method = value.ToLowerInvariant(); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.Modules.Scheduling/Contracts/IWorkflowTriggerScheduler.cs b/src/modules/Elsa.Modules.Scheduling/Contracts/IWorkflowTriggerScheduler.cs index b45440784..5daf29a9b 100644 --- a/src/modules/Elsa.Modules.Scheduling/Contracts/IWorkflowTriggerScheduler.cs +++ b/src/modules/Elsa.Modules.Scheduling/Contracts/IWorkflowTriggerScheduler.cs @@ -11,4 +11,5 @@ namespace Elsa.Modules.Scheduling.Contracts; public interface IWorkflowTriggerScheduler { Task ScheduleTriggersAsync(IEnumerable triggers, CancellationToken cancellationToken = default); + Task UnscheduleTriggersAsync(IEnumerable triggers, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/modules/Elsa.Modules.Scheduling/Extensions/ServiceCollectionExtensions.cs b/src/modules/Elsa.Modules.Scheduling/Extensions/ServiceCollectionExtensions.cs index 57602ba0b..fe5e721e2 100644 --- a/src/modules/Elsa.Modules.Scheduling/Extensions/ServiceCollectionExtensions.cs +++ b/src/modules/Elsa.Modules.Scheduling/Extensions/ServiceCollectionExtensions.cs @@ -18,7 +18,7 @@ public static class ServiceCollectionExtensions .AddSingleton() .AddJobHandler() .AddJobHandler() - .AddNotificationHandlersFrom() + .AddNotificationHandlersFrom() .AddHostedService(); return services; diff --git a/src/modules/Elsa.Modules.Scheduling/Handlers/ScheduleWorkflowsHandler.cs b/src/modules/Elsa.Modules.Scheduling/Handlers/ScheduleWorkflows.cs similarity index 60% rename from src/modules/Elsa.Modules.Scheduling/Handlers/ScheduleWorkflowsHandler.cs rename to src/modules/Elsa.Modules.Scheduling/Handlers/ScheduleWorkflows.cs index b026c5d0d..5b74ae0e9 100644 --- a/src/modules/Elsa.Modules.Scheduling/Handlers/ScheduleWorkflowsHandler.cs +++ b/src/modules/Elsa.Modules.Scheduling/Handlers/ScheduleWorkflows.cs @@ -2,23 +2,28 @@ using System.Threading; using System.Threading.Tasks; using Elsa.Mediator.Contracts; using Elsa.Modules.Scheduling.Contracts; +using Elsa.Runtime.Models; using Elsa.Runtime.Notifications; namespace Elsa.Modules.Scheduling.Handlers; // Updates scheduled jobs based on the updated workflow triggers. -public class ScheduleWorkflowsHandler : INotificationHandler, INotificationHandler +public class ScheduleWorkflows : INotificationHandler, INotificationHandler { private readonly IWorkflowTriggerScheduler _workflowTriggerScheduler; private readonly IWorkflowBookmarkScheduler _workflowBookmarkScheduler; - public ScheduleWorkflowsHandler(IWorkflowTriggerScheduler workflowTriggerScheduler, IWorkflowBookmarkScheduler workflowBookmarkScheduler) + public ScheduleWorkflows(IWorkflowTriggerScheduler workflowTriggerScheduler, IWorkflowBookmarkScheduler workflowBookmarkScheduler) { _workflowTriggerScheduler = workflowTriggerScheduler; _workflowBookmarkScheduler = workflowBookmarkScheduler; } - public async Task HandleAsync(TriggerIndexingFinished notification, CancellationToken cancellationToken) => await _workflowTriggerScheduler.ScheduleTriggersAsync(notification.Triggers, cancellationToken); + public async Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken) + { + await _workflowTriggerScheduler.UnscheduleTriggersAsync(notification.IndexedWorkflow.RemovedTriggers, cancellationToken); + await _workflowTriggerScheduler.ScheduleTriggersAsync(notification.IndexedWorkflow.AddedTriggers, cancellationToken); + } public async Task HandleAsync(WorkflowExecuted notification, CancellationToken cancellationToken) { diff --git a/src/modules/Elsa.Modules.Scheduling/HostedServices/ScheduleWorkflowsHostedService.cs b/src/modules/Elsa.Modules.Scheduling/HostedServices/ScheduleWorkflowsHostedService.cs index f1f236d6a..d8979f61f 100644 --- a/src/modules/Elsa.Modules.Scheduling/HostedServices/ScheduleWorkflowsHostedService.cs +++ b/src/modules/Elsa.Modules.Scheduling/HostedServices/ScheduleWorkflowsHostedService.cs @@ -39,7 +39,7 @@ public class ScheduleWorkflowsHostedService : BackgroundService private async Task ScheduleTriggersAsync(CancellationToken cancellationToken) { - var timerTriggers = (await _requestSender.RequestAsync(FindWorkflowTriggers.ForTrigger(), cancellationToken)).ToImmutableList(); + var timerTriggers = (await _requestSender.RequestAsync(FindWorkflowTriggersByName.ForTrigger(), cancellationToken)).ToImmutableList(); await _workflowTriggerScheduler.ScheduleTriggersAsync(timerTriggers, cancellationToken); } diff --git a/src/modules/Elsa.Modules.Scheduling/Services/WorkflowTriggerScheduler.cs b/src/modules/Elsa.Modules.Scheduling/Services/WorkflowTriggerScheduler.cs index 564599f44..8c5e0be25 100644 --- a/src/modules/Elsa.Modules.Scheduling/Services/WorkflowTriggerScheduler.cs +++ b/src/modules/Elsa.Modules.Scheduling/Services/WorkflowTriggerScheduler.cs @@ -17,20 +17,33 @@ namespace Elsa.Modules.Scheduling.Services; public class WorkflowTriggerScheduler : IWorkflowTriggerScheduler { private const string RootGroupKey = "WorkflowDefinition"; - + private readonly IJobScheduler _jobScheduler; public WorkflowTriggerScheduler(IJobScheduler jobScheduler) { _jobScheduler = jobScheduler; } - + public async Task ScheduleTriggersAsync(IEnumerable triggers, CancellationToken cancellationToken = default) { - var triggerList = triggers.ToList(); - // Select all Timer triggers. - var timerTriggers = triggerList.Filter().ToList(); + var timerTriggers = triggers.Filter().ToList(); + + // Schedule each trigger. + foreach (var trigger in timerTriggers) + { + // Schedule trigger. + var (dateTime, timeSpan) = JsonSerializer.Deserialize(trigger.Payload!)!; + var groupKeys = new[] { RootGroupKey, trigger.WorkflowDefinitionId }; + await _jobScheduler.ScheduleAsync(new RunWorkflowJob(trigger.WorkflowDefinitionId), new RecurringSchedule(dateTime, timeSpan), groupKeys, cancellationToken); + } + } + + public async Task UnscheduleTriggersAsync(IEnumerable triggers, CancellationToken cancellationToken = default) + { + // Select all Timer triggers. + var timerTriggers = triggers.Filter().ToList(); // Unschedule all triggers for the distinct set of affected workflows. var workflowDefinitionIds = timerTriggers.Select(x => x.WorkflowDefinitionId).Distinct().ToList(); @@ -40,13 +53,5 @@ public class WorkflowTriggerScheduler : IWorkflowTriggerScheduler var groupKeys = new[] { RootGroupKey, workflowDefinitionId }; await _jobScheduler.ClearAsync(groupKeys, cancellationToken); } - - foreach (var trigger in timerTriggers) - { - // Schedule trigger. - var (dateTime, timeSpan) = JsonSerializer.Deserialize(trigger.Payload!)!; - var groupKeys = new[] { RootGroupKey, trigger.WorkflowDefinitionId }; - await _jobScheduler.ScheduleAsync(new RunWorkflowJob(trigger.WorkflowDefinitionId), new RecurringSchedule(dateTime, timeSpan), groupKeys, cancellationToken); - } } } \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs b/src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs index d081190c4..9f42c48b1 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs +++ b/src/persistence/Elsa.Persistence.Abstractions/Commands/ReplaceWorkflowTriggers.cs @@ -4,4 +4,4 @@ using Elsa.Persistence.Entities; namespace Elsa.Persistence.Commands; -public record ReplaceWorkflowTriggers(Workflow Workflow, ICollection WorkflowTriggers) : ICommand; \ No newline at end of file +public record ReplaceWorkflowTriggers(Workflow Workflow, ICollection Removed, ICollection Added) : ICommand; \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs b/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs index d987b2ce7..b9a693f28 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs +++ b/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowBookmark.cs @@ -15,4 +15,19 @@ public class WorkflowBookmark : Entity public string? CallbackMethodName { get; set; } public Bookmark ToBookmark() => new(Id, Name, Hash, Payload, ActivityId, ActivityInstanceId, CallbackMethodName); + + public static WorkflowBookmark FromBookmark(Bookmark bookmark, WorkflowInstance workflowInstance) => + new() + { + Id = bookmark.Id, + WorkflowDefinitionId = workflowInstance.DefinitionId, + WorkflowInstanceId = workflowInstance.Id, + CorrelationId = workflowInstance.CorrelationId, + Hash = bookmark.Hash, + Payload = bookmark.Payload, + Name = bookmark.Name, + ActivityId = bookmark.ActivityId, + ActivityInstanceId = bookmark.ActivityInstanceId, + CallbackMethodName = bookmark.CallbackMethodName + }; } \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowTrigger.cs b/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowTrigger.cs index b455dff50..c0a236ce3 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowTrigger.cs +++ b/src/persistence/Elsa.Persistence.Abstractions/Entities/WorkflowTrigger.cs @@ -1,5 +1,8 @@ namespace Elsa.Persistence.Entities; +/// +/// Represents a trigger associated with a workflow definition. +/// public class WorkflowTrigger : Entity { public string WorkflowDefinitionId { get; set; } = default!; diff --git a/src/persistence/Elsa.Persistence.Abstractions/Extensions/WorkflowBookmarkExtensions.cs b/src/persistence/Elsa.Persistence.Abstractions/Extensions/WorkflowBookmarkExtensions.cs new file mode 100644 index 000000000..37a696a64 --- /dev/null +++ b/src/persistence/Elsa.Persistence.Abstractions/Extensions/WorkflowBookmarkExtensions.cs @@ -0,0 +1,14 @@ +using Elsa.Contracts; +using Elsa.Helpers; +using Elsa.Persistence.Entities; + +namespace Elsa.Persistence.Extensions; + +public static class WorkflowBookmarkExtensions +{ + public static IEnumerable Filter(this IEnumerable bookmarks) where T : ITrigger + { + var bookmarkName = TypeNameHelper.GenerateTypeName(); + return bookmarks.Where(x => x.Name == bookmarkName); + } +} \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggers.cs b/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggers.cs deleted file mode 100644 index c0ec68e64..000000000 --- a/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggers.cs +++ /dev/null @@ -1,20 +0,0 @@ -using Elsa.Contracts; -using Elsa.Helpers; -using Elsa.Mediator.Contracts; -using Elsa.Persistence.Entities; - -namespace Elsa.Persistence.Requests; - -public record FindWorkflowTriggers : IRequest> -{ - public FindWorkflowTriggers(string name, string? hash = default) - { - Name = name; - Hash = hash; - } - - public string Name { get; } - public string? Hash { get; } - - public static FindWorkflowTriggers ForTrigger(string? hash = default) where T : ITrigger => new(TypeNameHelper.GenerateTypeName(), hash); -} \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggersByName.cs b/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggersByName.cs new file mode 100644 index 000000000..bcb806d61 --- /dev/null +++ b/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggersByName.cs @@ -0,0 +1,11 @@ +using Elsa.Contracts; +using Elsa.Helpers; +using Elsa.Mediator.Contracts; +using Elsa.Persistence.Entities; + +namespace Elsa.Persistence.Requests; + +public record FindWorkflowTriggersByName(string Name, string? Hash = default) : IRequest> +{ + public static FindWorkflowTriggersByName ForTrigger(string? hash = default) where T : ITrigger => new(TypeNameHelper.GenerateTypeName(), hash); +} \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggersByWorkflowDefinition.cs b/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggersByWorkflowDefinition.cs new file mode 100644 index 000000000..df3bee2cd --- /dev/null +++ b/src/persistence/Elsa.Persistence.Abstractions/Requests/FindWorkflowTriggersByWorkflowDefinition.cs @@ -0,0 +1,6 @@ +using Elsa.Mediator.Contracts; +using Elsa.Persistence.Entities; + +namespace Elsa.Persistence.Requests; + +public record FindWorkflowTriggersByWorkflowDefinition(string WorkflowDefinitionId) : IRequest>; \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Contracts/IStore.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Contracts/IStore.cs index 2856b314a..4c716dc82 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Contracts/IStore.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Contracts/IStore.cs @@ -11,6 +11,7 @@ public interface IStore where TEntity : Entity Task FindAsync(Expression> predicate, CancellationToken cancellationToken = default); Task> FindManyAsync(Expression> predicate, CancellationToken cancellationToken = default); Task DeleteAsync(TEntity entity, CancellationToken cancellationToken = default); + Task DeleteManyAsync(IEnumerable entities, CancellationToken cancellationToken = default); Task DeleteWhereAsync(Expression> predicate, CancellationToken cancellationToken = default); IEnumerable Load(ElsaDbContext dbContext, IEnumerable entities); Task> QueryAsync(Func, IQueryable> query, CancellationToken cancellationToken = default); diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs index 0bf7a9e2a..00f97a367 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs @@ -12,9 +12,8 @@ public class ReplaceWorkflowTriggersHandler : ICommandHandler HandleAsync(ReplaceWorkflowTriggers command, CancellationToken cancellationToken) { - var definitionId = command.Workflow.Identity.DefinitionId; - await _store.DeleteWhereAsync(x => x.WorkflowDefinitionId == definitionId, cancellationToken); - await _store.SaveManyAsync(command.WorkflowTriggers, cancellationToken); + await _store.DeleteManyAsync(command.Removed, cancellationToken); + await _store.SaveManyAsync(command.Added, cancellationToken); return Unit.Instance; } diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Requests/FindWorkflowTriggersHandler.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Requests/FindWorkflowTriggersByNameHandler.cs similarity index 53% rename from src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Requests/FindWorkflowTriggersHandler.cs rename to src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Requests/FindWorkflowTriggersByNameHandler.cs index 28c03da27..ad720bfb2 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Requests/FindWorkflowTriggersHandler.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Handlers/Requests/FindWorkflowTriggersByNameHandler.cs @@ -5,14 +5,14 @@ using Elsa.Persistence.Requests; namespace Elsa.Persistence.EntityFrameworkCore.Handlers.Requests; -public class FindWorkflowTriggersHandler : IRequestHandler> +public class FindWorkflowTriggersByNameHandler : IRequestHandler> { private readonly IStore _store; - public FindWorkflowTriggersHandler(IStore store) => _store = store; + public FindWorkflowTriggersByNameHandler(IStore store) => _store = store; - public async Task> HandleAsync(FindWorkflowTriggers request, CancellationToken cancellationToken) + public async Task> HandleAsync(FindWorkflowTriggersByName request, CancellationToken cancellationToken) { - return await _store.QueryAsync(query => + var triggers = await _store.QueryAsync(query => { query = query.Where(x => x.Name == request.Name); @@ -21,5 +21,7 @@ public class FindWorkflowTriggersHandler : IRequestHandler> +{ + private readonly IStore _store; + public FindWorkflowTriggersByWorkflowDefinitionHandler(IStore store) => _store = store; + + public async Task> HandleAsync(FindWorkflowTriggersByWorkflowDefinition request, CancellationToken cancellationToken) + { + var triggers = await _store.QueryAsync(query => query.Where(x => x.WorkflowDefinitionId == request.WorkflowDefinitionId), cancellationToken); + return triggers.ToList(); + } +} \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Services/EFCoreStore.cs b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Services/EFCoreStore.cs index 05e39c885..5c8f15e7a 100644 --- a/src/persistence/Elsa.Persistence.EntityFrameworkCore/Services/EFCoreStore.cs +++ b/src/persistence/Elsa.Persistence.EntityFrameworkCore/Services/EFCoreStore.cs @@ -60,13 +60,21 @@ public class EFCoreStore : IStore where TEntity : Entity return await dbContext.SaveChangesAsync(cancellationToken) == 1; } + public async Task DeleteManyAsync(IEnumerable entities, CancellationToken cancellationToken = default) + { + await using var dbContext = await CreateDbContextAsync(cancellationToken); + var list = entities.ToList(); + await dbContext.BulkDeleteAsync(list, cancellationToken: cancellationToken); + return list.Count; + } + public async Task DeleteWhereAsync(Expression> predicate, CancellationToken cancellationToken = default) { await using var dbContext = await CreateDbContextAsync(cancellationToken); var set = dbContext.Set(); return await set.Where(predicate).BatchDeleteWithWorkAroundAsync(dbContext, cancellationToken); } - + public async Task> QueryAsync(Func, IQueryable> query, CancellationToken cancellationToken = default) { await using var dbContext = await CreateDbContextAsync(cancellationToken); @@ -74,12 +82,12 @@ public class EFCoreStore : IStore where TEntity : Entity var queryable = query(set.AsQueryable()); queryable = query(queryable); - + return Load(dbContext, queryable).ToList(); } public IEnumerable Load(ElsaDbContext dbContext, IEnumerable entities) => OnLoading(dbContext, entities.ToList()); - + private void OnSaving(ElsaDbContext dbContext, TEntity entity) { var handler = _serviceProvider.GetService>(); diff --git a/src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs b/src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs index 756aaf405..9e4363de6 100644 --- a/src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs +++ b/src/persistence/Elsa.Persistence.InMemory/Handlers/Commands/ReplaceWorkflowTriggersHandler.cs @@ -1,4 +1,3 @@ -using System.Linq; using System.Threading; using System.Threading.Tasks; using Elsa.Mediator.Contracts; @@ -19,9 +18,8 @@ public class ReplaceWorkflowTriggersHandler : ICommandHandler HandleAsync(ReplaceWorkflowTriggers command, CancellationToken cancellationToken) { - var ids = command.WorkflowTriggers.Select(x => x.Id).ToList(); - _store.DeleteWhere(x => ids.Contains(x.Id)); - _store.SaveMany(command.WorkflowTriggers); + _store.DeleteMany(command.Removed); + _store.SaveMany(command.Added); return Task.FromResult(Unit.Instance); } diff --git a/src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersHandler.cs b/src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersByNameHandler.cs similarity index 59% rename from src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersHandler.cs rename to src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersByNameHandler.cs index 4454289d1..638a6aced 100644 --- a/src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersHandler.cs +++ b/src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersByNameHandler.cs @@ -9,12 +9,12 @@ using Elsa.Persistence.Requests; namespace Elsa.Persistence.InMemory.Handlers.Requests; -public class FindWorkflowTriggersHandler : IRequestHandler> +public class FindWorkflowTriggersByNameHandler : IRequestHandler> { private readonly InMemoryStore _store; - public FindWorkflowTriggersHandler(InMemoryStore store) => _store = store; + public FindWorkflowTriggersByNameHandler(InMemoryStore store) => _store = store; - public Task> HandleAsync(FindWorkflowTriggers request, CancellationToken cancellationToken) + public Task> HandleAsync(FindWorkflowTriggersByName request, CancellationToken cancellationToken) { var triggers = _store.Query(query => { @@ -26,6 +26,6 @@ public class FindWorkflowTriggersHandler : IRequestHandler>(triggers.ToList()); } } \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersByWorkflowDefinitionHandler.cs b/src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersByWorkflowDefinitionHandler.cs new file mode 100644 index 000000000..08c695694 --- /dev/null +++ b/src/persistence/Elsa.Persistence.InMemory/Handlers/Requests/FindWorkflowTriggersByWorkflowDefinitionHandler.cs @@ -0,0 +1,23 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Elsa.Mediator.Contracts; +using Elsa.Persistence.Entities; +using Elsa.Persistence.InMemory.Services; +using Elsa.Persistence.Requests; + +namespace Elsa.Persistence.InMemory.Handlers.Requests; + +public class FindWorkflowTriggersByWorkflowDefinitionHandler : IRequestHandler> +{ + private readonly InMemoryStore _store; + public FindWorkflowTriggersByWorkflowDefinitionHandler(InMemoryStore store) => _store = store; + + public Task> HandleAsync(FindWorkflowTriggersByWorkflowDefinition request, CancellationToken cancellationToken) + { + var triggers = _store.Query(query => query.Where(x => x.WorkflowDefinitionId == request.WorkflowDefinitionId)); + + return Task.FromResult>(triggers.ToList()); + } +} \ No newline at end of file diff --git a/src/persistence/Elsa.Persistence.InMemory/Services/InMemoryStore.cs b/src/persistence/Elsa.Persistence.InMemory/Services/InMemoryStore.cs index 7327acc69..be9036fc2 100644 --- a/src/persistence/Elsa.Persistence.InMemory/Services/InMemoryStore.cs +++ b/src/persistence/Elsa.Persistence.InMemory/Services/InMemoryStore.cs @@ -12,7 +12,7 @@ public class InMemoryStore where TEntity : Entity public void SaveMany(IEnumerable entities) { - foreach (var entity in entities) + foreach (var entity in entities) Save(entity); } @@ -48,6 +48,20 @@ public class InMemoryStore where TEntity : Entity return count; } + public int DeleteMany(IEnumerable entities) + { + var count = 0; + var list = entities.ToList(); + + foreach (var entity in list) + { + count++; + Entities.Remove(entity.Id); + } + + return count; + } + public IEnumerable Query(Func, IQueryable> query) { var queryable = Entities.Values.AsQueryable(); diff --git a/src/runtime/Elsa.Runtime/Contracts/IBookmarkManager.cs b/src/runtime/Elsa.Runtime/Contracts/IBookmarkManager.cs new file mode 100644 index 000000000..34b8b24f2 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Contracts/IBookmarkManager.cs @@ -0,0 +1,12 @@ +using Elsa.Persistence.Entities; + +namespace Elsa.Runtime.Contracts; + +/// +/// Provides methods to add and remove bookmarks. +/// +public interface IBookmarkManager +{ + Task DeleteBookmarksAsync(IEnumerable bookmarks, CancellationToken cancellationToken = default); + Task SaveBookmarksAsync(IEnumerable bookmarks, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs b/src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs index 9bae1fa90..b2479316e 100644 --- a/src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs +++ b/src/runtime/Elsa.Runtime/Contracts/ITriggerIndexer.cs @@ -1,5 +1,7 @@ using Elsa.Models; using Elsa.Persistence.Entities; +using Elsa.Runtime.Models; +using Elsa.Runtime.Notifications; namespace Elsa.Runtime.Contracts; @@ -8,10 +10,10 @@ public interface ITriggerIndexer /// /// Indexes triggers from all workflows. /// - Task IndexTriggersAsync(CancellationToken cancellationToken = default); + Task> IndexTriggersAsync(CancellationToken cancellationToken = default); /// /// 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/Elsa.Runtime.csproj b/src/runtime/Elsa.Runtime/Elsa.Runtime.csproj index 54a2d6068..5732618fb 100644 --- a/src/runtime/Elsa.Runtime/Elsa.Runtime.csproj +++ b/src/runtime/Elsa.Runtime/Elsa.Runtime.csproj @@ -16,7 +16,7 @@ - + diff --git a/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs b/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs index 6c7786c9e..65150a759 100644 --- a/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs +++ b/src/runtime/Elsa.Runtime/Extensions/ServiceCollectionExtensions.cs @@ -24,6 +24,7 @@ public static class ServiceCollectionExtensions .AddSingleton() .AddSingleton() .AddSingleton() + .AddSingleton() // Stimulus handlers. .AddStimulusHandler() @@ -44,8 +45,8 @@ public static class ServiceCollectionExtensions .AddNotificationHandlersFrom(typeof(ServiceCollectionExtensions)) // Hosted Services. - .AddHostedService() - .AddHostedService(); + .AddHostedService() + .AddHostedService(); ; } @@ -60,5 +61,5 @@ public static class ServiceCollectionExtensions public static IServiceCollection AddStimulusHandler(this IServiceCollection services) where T : class, IStimulusHandler => services.AddSingleton(); public static IServiceCollection AddInstructionInterpreter(this IServiceCollection services) where T : class, IWorkflowInstructionInterpreter => services.AddSingleton(); public static IServiceCollection ConfigureWorkflowRuntime(this IServiceCollection services, Action configure) => services.Configure(configure); - public static IServiceCollection IndexWorkflowTriggers(this IServiceCollection services) => services.AddHostedService(); + public static IServiceCollection IndexWorkflowTriggers(this IServiceCollection services) => services.AddHostedService(); } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Extensions/WorkflowExecutionBuilderExtensions.cs b/src/runtime/Elsa.Runtime/Extensions/WorkflowExecutionBuilderExtensions.cs new file mode 100644 index 000000000..f02ff1a17 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Extensions/WorkflowExecutionBuilderExtensions.cs @@ -0,0 +1,12 @@ +using Elsa.Contracts; +using Elsa.Pipelines.WorkflowExecution; +using Elsa.Runtime.Middleware; + +namespace Elsa.Runtime.Extensions; + +public static class WorkflowExecutionBuilderExtensions +{ + public static IWorkflowExecutionBuilder UsePersistence(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware(); + public static IWorkflowExecutionBuilder UseWorkflowExecutionEvents(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware(); + public static IWorkflowExecutionBuilder UseWorkflowExecutionLogPersistence(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware(); +} \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/HostedServices/IndexWorkflowTriggersHostedService.cs b/src/runtime/Elsa.Runtime/HostedServices/IndexWorkflowTriggers.cs similarity index 81% rename from src/runtime/Elsa.Runtime/HostedServices/IndexWorkflowTriggersHostedService.cs rename to src/runtime/Elsa.Runtime/HostedServices/IndexWorkflowTriggers.cs index 238989bf5..8cdc76834 100644 --- a/src/runtime/Elsa.Runtime/HostedServices/IndexWorkflowTriggersHostedService.cs +++ b/src/runtime/Elsa.Runtime/HostedServices/IndexWorkflowTriggers.cs @@ -7,11 +7,11 @@ namespace Elsa.Runtime.HostedServices; /// /// Asynchronously indexes all workflows in the background. /// -public class IndexWorkflowTriggersHostedService : BackgroundService +public class IndexWorkflowTriggers : BackgroundService { private readonly IServiceScopeFactory _serviceScopeFactory; - public IndexWorkflowTriggersHostedService(IServiceScopeFactory serviceScopeFactory) + public IndexWorkflowTriggers(IServiceScopeFactory serviceScopeFactory) { _serviceScopeFactory = serviceScopeFactory; } diff --git a/src/runtime/Elsa.Runtime/HostedServices/RegisterDescriptorsHostedService.cs b/src/runtime/Elsa.Runtime/HostedServices/RegisterDescriptors.cs similarity index 80% rename from src/runtime/Elsa.Runtime/HostedServices/RegisterDescriptorsHostedService.cs rename to src/runtime/Elsa.Runtime/HostedServices/RegisterDescriptors.cs index abcca46a1..da558a48c 100644 --- a/src/runtime/Elsa.Runtime/HostedServices/RegisterDescriptorsHostedService.cs +++ b/src/runtime/Elsa.Runtime/HostedServices/RegisterDescriptors.cs @@ -3,12 +3,12 @@ using Microsoft.Extensions.Hosting; namespace Elsa.Runtime.HostedServices; -public class RegisterDescriptorsHostedService : IHostedService +public class RegisterDescriptors : IHostedService { private readonly IActivityRegistryPopulator _activityRegistryPopulator; private readonly ITriggerRegistryPopulator _triggerRegistryPopulator; - public RegisterDescriptorsHostedService(IActivityRegistryPopulator activityRegistryPopulator, ITriggerRegistryPopulator triggerRegistryPopulator) + public RegisterDescriptors(IActivityRegistryPopulator activityRegistryPopulator, ITriggerRegistryPopulator triggerRegistryPopulator) { _activityRegistryPopulator = activityRegistryPopulator; _triggerRegistryPopulator = triggerRegistryPopulator; diff --git a/src/runtime/Elsa.Runtime/HostedServices/RegisterExpressionSyntaxDescriptorsHostedService.cs b/src/runtime/Elsa.Runtime/HostedServices/RegisterExpressionSyntaxDescriptors.cs similarity index 61% rename from src/runtime/Elsa.Runtime/HostedServices/RegisterExpressionSyntaxDescriptorsHostedService.cs rename to src/runtime/Elsa.Runtime/HostedServices/RegisterExpressionSyntaxDescriptors.cs index 68c23b5e9..5568b0b2d 100644 --- a/src/runtime/Elsa.Runtime/HostedServices/RegisterExpressionSyntaxDescriptorsHostedService.cs +++ b/src/runtime/Elsa.Runtime/HostedServices/RegisterExpressionSyntaxDescriptors.cs @@ -3,10 +3,10 @@ using Microsoft.Extensions.Hosting; namespace Elsa.Runtime.HostedServices; -public class RegisterExpressionSyntaxDescriptorsHostedService : IHostedService +public class RegisterExpressionSyntaxDescriptors : IHostedService { private readonly IExpressionSyntaxRegistryPopulator _expressionSyntaxRegistryPopulator; - public RegisterExpressionSyntaxDescriptorsHostedService(IExpressionSyntaxRegistryPopulator expressionSyntaxRegistryPopulator) => _expressionSyntaxRegistryPopulator = expressionSyntaxRegistryPopulator; + public RegisterExpressionSyntaxDescriptors(IExpressionSyntaxRegistryPopulator expressionSyntaxRegistryPopulator) => _expressionSyntaxRegistryPopulator = expressionSyntaxRegistryPopulator; public async Task StartAsync(CancellationToken cancellationToken) => await _expressionSyntaxRegistryPopulator.PopulateRegistryAsync(cancellationToken); public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; } \ 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 37657e053..15dda3540 100644 --- a/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs +++ b/src/runtime/Elsa.Runtime/Interpreters/ResumeWorkflowInstructionInterpreter.cs @@ -36,7 +36,7 @@ public class ResumeWorkflowInstructionInterpreter : WorkflowInstructionInterpret { _logger .LogWarning( - "Workflow bookmark {WorkflowBookmarkId} for workflow definition {WorkflowDefinitionId} points to workflow instance ID {WorkflowInstanceId}, but no such workflow instance was found", workflowBookmark.Id, workflowBookmark.WorkflowDefinitionId, workflowBookmark.WorkflowInstanceId); + "Workflow bookmark {WorkflowBookmarkId} for workflow definition {WorkflowDefinitionId} references workflow instance ID {WorkflowInstanceId}, but no such workflow instance was found", workflowBookmark.Id, workflowBookmark.WorkflowDefinitionId, workflowBookmark.WorkflowInstanceId); return null; } @@ -45,7 +45,7 @@ public class ResumeWorkflowInstructionInterpreter : WorkflowInstructionInterpret if (workflow == null) { - _logger.LogWarning("Workflow bookmark {WorkflowBookmarkId} points to workflow definition ID {WorkflowDefinitionId}, but no such workflow definition was found", workflowBookmark.Id, workflowBookmark.WorkflowDefinitionId); + _logger.LogWarning("Workflow bookmark {WorkflowBookmarkId} references workflow definition ID {WorkflowDefinitionId}, but no such workflow definition was found", workflowBookmark.Id, workflowBookmark.WorkflowDefinitionId); return null; } diff --git a/src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowExecutionLogMiddleware.cs b/src/runtime/Elsa.Runtime/Middleware/PersistWorkflowExecutionLogMiddleware.cs similarity index 83% rename from src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowExecutionLogMiddleware.cs rename to src/runtime/Elsa.Runtime/Middleware/PersistWorkflowExecutionLogMiddleware.cs index 8b85940da..a2bc34424 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowExecutionLogMiddleware.cs +++ b/src/runtime/Elsa.Runtime/Middleware/PersistWorkflowExecutionLogMiddleware.cs @@ -6,12 +6,7 @@ using Elsa.Persistence.Entities; using Elsa.Pipelines.WorkflowExecution; using Elsa.Pipelines.WorkflowExecution.Components; -namespace Elsa.Persistence.Middleware.WorkflowExecution; - -public static class PersistWorkflowExecutionLogMiddlewareExtensions -{ - public static IWorkflowExecutionBuilder UseWorkflowExecutionLogPersistence(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware(); -} +namespace Elsa.Runtime.Middleware; /// /// Takes care of persisting a workflow instance after workflow execution. diff --git a/src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowInstanceMiddleware.cs b/src/runtime/Elsa.Runtime/Middleware/PersistWorkflowInstanceMiddleware.cs similarity index 65% rename from src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowInstanceMiddleware.cs rename to src/runtime/Elsa.Runtime/Middleware/PersistWorkflowInstanceMiddleware.cs index d02defb38..c7d14fb6b 100644 --- a/src/persistence/Elsa.Persistence.Abstractions/Middleware/WorkflowExecution/PersistWorkflowInstanceMiddleware.cs +++ b/src/runtime/Elsa.Runtime/Middleware/PersistWorkflowInstanceMiddleware.cs @@ -1,4 +1,5 @@ using Elsa.Contracts; +using Elsa.Helpers; using Elsa.Mediator.Contracts; using Elsa.Models; using Elsa.Persistence.Commands; @@ -6,13 +7,9 @@ using Elsa.Persistence.Entities; using Elsa.Persistence.Requests; using Elsa.Pipelines.WorkflowExecution; using Elsa.Pipelines.WorkflowExecution.Components; +using Elsa.Runtime.Contracts; -namespace Elsa.Persistence.Middleware.WorkflowExecution; - -public static class PersistWorkflowInstanceMiddlewareExtensions -{ - public static IWorkflowExecutionBuilder UsePersistence(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware(); -} +namespace Elsa.Runtime.Middleware; /// /// Takes care of persisting a workflow instance after workflow execution. @@ -22,7 +19,7 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware private readonly IRequestSender _requestSender; private readonly ICommandSender _commandSender; private readonly IWorkflowStateSerializer _workflowStateSerializer; - private readonly IPayloadSerializer _payloadSerializer; + private readonly IBookmarkManager _bookmarkManager; private readonly IIdentityGenerator _identityGenerator; private readonly ISystemClock _clock; @@ -30,15 +27,15 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware WorkflowMiddlewareDelegate next, IRequestSender requestSender, ICommandSender commandSender, + IBookmarkManager bookmarkManager, IWorkflowStateSerializer workflowStateSerializer, - IPayloadSerializer payloadSerializer, IIdentityGenerator identityGenerator, ISystemClock clock) : base(next) { _requestSender = requestSender; _commandSender = commandSender; + _bookmarkManager = bookmarkManager; _workflowStateSerializer = workflowStateSerializer; - _payloadSerializer = payloadSerializer; _identityGenerator = identityGenerator; _clock = clock; } @@ -66,19 +63,14 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware // Get a copy of current bookmarks. var existingBookmarks = await _requestSender.RequestAsync(new FindWorkflowBookmarks(workflowInstance.Id), cancellationToken); var bookmarksSnapshot = existingBookmarks.Select(x => x.ToBookmark()).ToList(); + var bookmarks = bookmarksSnapshot.ToList(); // Exclude the current bookmark that initiated the creation of the workflow context, if any. var invokedBookmark = context.Bookmark; - var removedBookmarkIds = new List(); - - if (invokedBookmark != null) - { - bookmarksSnapshot.RemoveAll(x => x.Id == invokedBookmark.Id); - removedBookmarkIds.Add(invokedBookmark.Id); - } + if (invokedBookmark != null) bookmarks.RemoveAll(x => x.Id == invokedBookmark.Id); // Apply bookmarks to workflow context. - context.RegisterBookmarks(bookmarksSnapshot); + context.RegisterBookmarks(bookmarks); // Persist workflow instance. await _commandSender.ExecuteAsync(new SaveWorkflowInstance(workflowInstance), cancellationToken); @@ -93,25 +85,15 @@ public class PersistWorkflowInstanceMiddleware : WorkflowExecutionMiddleware // Persist updated workflow instance. await _commandSender.ExecuteAsync(new SaveWorkflowInstance(workflowInstance), cancellationToken); - // Remove bookmarks that were in the snapshot but no longer present in context. - removedBookmarkIds.AddRange(bookmarksSnapshot.Except(context.Bookmarks).Select(x => x.Id)); - await _commandSender.ExecuteAsync(new DeleteWorkflowBookmarks(removedBookmarkIds), cancellationToken); + // Get a diff between bookmarks that were in the snapshot but no longer present in context. + var diff = Diff.For(bookmarksSnapshot, context.Bookmarks.ToList()); - // Persist bookmarks, if any. - var workflowBookmarks = context.Bookmarks.Select(x => new WorkflowBookmark - { - Id = x.Id, - WorkflowDefinitionId = definitionId, - WorkflowInstanceId = workflowInstance.Id, - CorrelationId = workflowInstance.CorrelationId, - Hash = x.Hash, - Payload = x.Payload, - Name = x.Name, - ActivityId = x.ActivityId, - ActivityInstanceId = x.ActivityInstanceId, - CallbackMethodName = x.CallbackMethodName - }).ToList(); + // Delete removed bookmarks. + var removedBookmarks = diff.Removed.Select(x => WorkflowBookmark.FromBookmark(x, workflowInstance)).ToList(); + await _bookmarkManager.DeleteBookmarksAsync(removedBookmarks, cancellationToken); - await _commandSender.ExecuteAsync(new SaveWorkflowBookmarks(workflowBookmarks), cancellationToken); + // Persist created bookmarks. + var createdBookmarks = diff.Added.Select(x => WorkflowBookmark.FromBookmark(x, workflowInstance)).ToList(); + await _bookmarkManager.SaveBookmarksAsync(createdBookmarks, cancellationToken); } } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Middleware/PublishWorkflowExecutionEventsMiddleware.cs b/src/runtime/Elsa.Runtime/Middleware/PublishWorkflowExecutionEventsMiddleware.cs index a303021c3..223179b49 100644 --- a/src/runtime/Elsa.Runtime/Middleware/PublishWorkflowExecutionEventsMiddleware.cs +++ b/src/runtime/Elsa.Runtime/Middleware/PublishWorkflowExecutionEventsMiddleware.cs @@ -7,11 +7,6 @@ using Elsa.Runtime.Notifications; namespace Elsa.Runtime.Middleware; -public static class PersistWorkflowExecutionLogMiddlewareExtensions -{ - public static IWorkflowExecutionBuilder UseWorkflowExecutionEvents(this IWorkflowExecutionBuilder builder) => builder.UseMiddleware(); -} - /// /// Processes collected bookmarks. /// diff --git a/src/runtime/Elsa.Runtime/Models/IndexedWorkflow.cs b/src/runtime/Elsa.Runtime/Models/IndexedWorkflow.cs new file mode 100644 index 000000000..1390cde1a --- /dev/null +++ b/src/runtime/Elsa.Runtime/Models/IndexedWorkflow.cs @@ -0,0 +1,6 @@ +using Elsa.Models; +using Elsa.Persistence.Entities; + +namespace Elsa.Runtime.Models; + +public record IndexedWorkflow(Workflow Workflow, ICollection AddedTriggers, ICollection RemovedTriggers); \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/Stimuli.cs b/src/runtime/Elsa.Runtime/Models/Stimulus.cs similarity index 89% rename from src/runtime/Elsa.Runtime/Models/Stimuli.cs rename to src/runtime/Elsa.Runtime/Models/Stimulus.cs index 0c78771a9..f3b2d7476 100644 --- a/src/runtime/Elsa.Runtime/Models/Stimuli.cs +++ b/src/runtime/Elsa.Runtime/Models/Stimulus.cs @@ -2,7 +2,7 @@ using Elsa.Runtime.Stimuli; namespace Elsa.Runtime.Models; -public static class Stimuli +public static class Stimulus { public static StandardStimulus Standard(string activityTypeName, string? hash = default, IDictionary? data = default) => new(activityTypeName, hash, data); } \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/WorkflowIndexingCompleted.cs b/src/runtime/Elsa.Runtime/Models/WorkflowIndexingCompleted.cs new file mode 100644 index 000000000..47f2b79b1 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Models/WorkflowIndexingCompleted.cs @@ -0,0 +1,10 @@ +using Elsa.Mediator.Contracts; +using Elsa.Runtime.Notifications; + +namespace Elsa.Runtime.Models; + +/// +/// Published when the all workflows' triggers have been indexed. +/// +/// Contains the workflows that were indexed and their resulting triggers. +public record WorkflowIndexingCompleted(ICollection IndexedWorkflows) : INotification; \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Models/WorkflowTriggersIndexed.cs b/src/runtime/Elsa.Runtime/Models/WorkflowTriggersIndexed.cs new file mode 100644 index 000000000..4cf8967ef --- /dev/null +++ b/src/runtime/Elsa.Runtime/Models/WorkflowTriggersIndexed.cs @@ -0,0 +1,10 @@ +using Elsa.Mediator.Contracts; +using Elsa.Runtime.Notifications; + +namespace Elsa.Runtime.Models; + +/// +/// Published when the specified workflow's triggers have been indexed. +/// +/// Contains the workflow that was indexed and the resulting triggers. +public record WorkflowTriggersIndexed(IndexedWorkflow IndexedWorkflow) : INotification; \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Notifications/TriggerIndexingFinished.cs b/src/runtime/Elsa.Runtime/Notifications/WorkflowBookmarksDeleted.cs similarity index 52% rename from src/runtime/Elsa.Runtime/Notifications/TriggerIndexingFinished.cs rename to src/runtime/Elsa.Runtime/Notifications/WorkflowBookmarksDeleted.cs index a762c7f59..1bdf8e9ba 100644 --- a/src/runtime/Elsa.Runtime/Notifications/TriggerIndexingFinished.cs +++ b/src/runtime/Elsa.Runtime/Notifications/WorkflowBookmarksDeleted.cs @@ -3,4 +3,4 @@ using Elsa.Persistence.Entities; namespace Elsa.Runtime.Notifications; -public record TriggerIndexingFinished(ICollection Triggers) : INotification; \ No newline at end of file +public record WorkflowBookmarksDeleted(List Bookmarks) : INotification; \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Notifications/WorkflowBookmarksSaved.cs b/src/runtime/Elsa.Runtime/Notifications/WorkflowBookmarksSaved.cs new file mode 100644 index 000000000..1a35b64a9 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Notifications/WorkflowBookmarksSaved.cs @@ -0,0 +1,6 @@ +using Elsa.Mediator.Contracts; +using Elsa.Persistence.Entities; + +namespace Elsa.Runtime.Notifications; + +public record WorkflowBookmarksSaved(List Bookmarks) : INotification; \ No newline at end of file diff --git a/src/runtime/Elsa.Runtime/Services/BookmarkManager.cs b/src/runtime/Elsa.Runtime/Services/BookmarkManager.cs new file mode 100644 index 000000000..21feefc13 --- /dev/null +++ b/src/runtime/Elsa.Runtime/Services/BookmarkManager.cs @@ -0,0 +1,34 @@ +using Elsa.Mediator.Contracts; +using Elsa.Persistence.Commands; +using Elsa.Persistence.Entities; +using Elsa.Runtime.Contracts; +using Elsa.Runtime.Notifications; + +namespace Elsa.Runtime.Services; + +public class BookmarkManager : IBookmarkManager +{ + private readonly ICommandSender _commandSender; + private readonly IEventPublisher _eventPublisher; + + public BookmarkManager(ICommandSender commandSender, IEventPublisher eventPublisher) + { + _commandSender = commandSender; + _eventPublisher = eventPublisher; + } + + public async Task DeleteBookmarksAsync(IEnumerable bookmarks, CancellationToken cancellationToken = default) + { + var list = bookmarks.ToList(); + var ids = list.Select(x => x.Id).ToList(); + await _commandSender.ExecuteAsync(new DeleteWorkflowBookmarks(ids), cancellationToken); + await _eventPublisher.PublishAsync(new WorkflowBookmarksDeleted(list), cancellationToken); + } + + public async Task SaveBookmarksAsync(IEnumerable bookmarks, CancellationToken cancellationToken = default) + { + var list = bookmarks.ToList(); + await _commandSender.ExecuteAsync(new SaveWorkflowBookmarks(list), cancellationToken); + await _eventPublisher.PublishAsync(new WorkflowBookmarksSaved(list), cancellationToken); + } +} \ 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 33c4a7d4d..f0609bfe1 100644 --- a/src/runtime/Elsa.Runtime/Services/TriggerIndexer.cs +++ b/src/runtime/Elsa.Runtime/Services/TriggerIndexer.cs @@ -1,3 +1,4 @@ +using System.Collections.ObjectModel; using System.Diagnostics; using System.Runtime.CompilerServices; using System.Text.Json; @@ -8,7 +9,9 @@ using Elsa.Mediator.Contracts; using Elsa.Models; using Elsa.Persistence.Commands; using Elsa.Persistence.Entities; +using Elsa.Persistence.Requests; using Elsa.Runtime.Contracts; +using Elsa.Runtime.Models; using Elsa.Runtime.Notifications; using Microsoft.Extensions.Logging; @@ -25,6 +28,7 @@ public class TriggerIndexer : ITriggerIndexer private readonly IWorkflowRegistry _workflowRegistry; private readonly IExpressionEvaluator _expressionEvaluator; private readonly IIdentityGenerator _identityGenerator; + private readonly IRequestSender _requestSender; private readonly ICommandSender _commandSender; private readonly IEventPublisher _eventPublisher; private readonly IServiceProvider _serviceProvider; @@ -35,6 +39,7 @@ public class TriggerIndexer : ITriggerIndexer IWorkflowRegistry workflowRegistry, IExpressionEvaluator expressionEvaluator, IIdentityGenerator identityGenerator, + IRequestSender requestSender, ICommandSender commandSender, IEventPublisher eventPublisher, IServiceProvider serviceProvider, @@ -44,6 +49,7 @@ public class TriggerIndexer : ITriggerIndexer _workflowRegistry = workflowRegistry; _expressionEvaluator = expressionEvaluator; _identityGenerator = identityGenerator; + _requestSender = requestSender; _commandSender = commandSender; _eventPublisher = eventPublisher; _serviceProvider = serviceProvider; @@ -51,7 +57,7 @@ public class TriggerIndexer : ITriggerIndexer _logger = logger; } - public async Task IndexTriggersAsync(CancellationToken cancellationToken = default) + public async Task> IndexTriggersAsync(CancellationToken cancellationToken = default) { var stopwatch = new Stopwatch(); @@ -60,37 +66,49 @@ public class TriggerIndexer : ITriggerIndexer // Only stream workflows from providers that are not "dynamic" (such as DatabaseWorkflowProvider). var workflows = _workflowRegistry.StreamAllAsync(WorkflowRegistry.SkipDynamicProviders, cancellationToken); - //var collectedTriggers = new List(); + // Index each workflow. + var indexedWorkflows = new Collection(); await foreach (var workflow in workflows.WithCancellation(cancellationToken)) { - //var triggers = await GetTriggersAsync(workflow, cancellationToken).ToListAsync(cancellationToken); - await IndexTriggersAsync(workflow, cancellationToken); + var indexedWorkflow = await IndexTriggersAsync(workflow, cancellationToken); + indexedWorkflows.Add(indexedWorkflow); } - // // Replace triggers for the specified workflow. - // await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(collectedTriggers), cancellationToken); + // Publish event. + await _eventPublisher.PublishAsync(new WorkflowIndexingCompleted(indexedWorkflows), cancellationToken); stopwatch.Stop(); _logger.LogInformation("Finished indexing workflow triggers in {ElapsedTime}", stopwatch.Elapsed); - - // // Publish event. - // await _eventPublisher.PublishAsync(new TriggerIndexingFinished(collectedTriggers), cancellationToken); + return indexedWorkflows; } - 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); + // Get current triggers + var currentTriggers = await GetCurrentTriggersAsync(workflow.Identity.DefinitionId, cancellationToken); + + // Collect new triggers **if workflow is published**. + var newTriggers = workflow.Publication.IsPublished + ? await GetTriggersAsync(workflow, cancellationToken).ToListAsync(cancellationToken) + : new List(0); + + // Diff triggers. + var diff = Diff.For(currentTriggers, newTriggers); // Replace triggers for the specified workflow. - await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(workflow, triggers), cancellationToken); + await _commandSender.ExecuteAsync(new ReplaceWorkflowTriggers(workflow, diff.Removed, diff.Added), cancellationToken); + + var indexedWorkflow = new IndexedWorkflow(workflow, diff.Added, diff.Removed); // Publish event. - await _eventPublisher.PublishAsync(new TriggerIndexingFinished(triggers), cancellationToken); - return triggers; + await _eventPublisher.PublishAsync(new WorkflowTriggersIndexed(indexedWorkflow), cancellationToken); + return indexedWorkflow; } + private async Task> GetCurrentTriggersAsync(string workflowDefinitionId, CancellationToken cancellationToken) => + await _requestSender.RequestAsync(new FindWorkflowTriggersByWorkflowDefinition(workflowDefinitionId), cancellationToken); + private async IAsyncEnumerable GetTriggersAsync(Workflow workflow, [EnumeratorCancellation] CancellationToken cancellationToken = default) { var context = new WorkflowIndexingContext(workflow); diff --git a/src/runtime/Elsa.Runtime/Services/WorkflowServer.cs b/src/runtime/Elsa.Runtime/Services/WorkflowServer.cs index fabd106d7..59c942552 100644 --- a/src/runtime/Elsa.Runtime/Services/WorkflowServer.cs +++ b/src/runtime/Elsa.Runtime/Services/WorkflowServer.cs @@ -7,7 +7,7 @@ namespace Elsa.Runtime.Services; /// /// A workflow server represents a logical unit of services to invoke workflows. -/// It's basically a thin wrapper around a service provider built from a given service collection using . +/// It's essentially a thin wrapper around a service provider built from a given service collection using . /// May be useful for scenarios where you have multiple tenants in a system or if you simply want to use different workflow engine configurations within the same application. /// public class WorkflowServer : IWorkflowServer diff --git a/src/runtime/Elsa.Runtime/Stimuli/Handlers/TriggerWorkflowsStimulusHandler.cs b/src/runtime/Elsa.Runtime/Stimuli/Handlers/TriggerWorkflowsStimulusHandler.cs index d5b943f35..48c7bed41 100644 --- a/src/runtime/Elsa.Runtime/Stimuli/Handlers/TriggerWorkflowsStimulusHandler.cs +++ b/src/runtime/Elsa.Runtime/Stimuli/Handlers/TriggerWorkflowsStimulusHandler.cs @@ -13,7 +13,7 @@ public class TriggerWorkflowsStimulusHandler : StimulusHandler protected override async ValueTask> GetInstructionsAsync(StandardStimulus stimulus, CancellationToken cancellationToken = default) { - var workflowTriggers = (await _mediator.RequestAsync(new FindWorkflowTriggers(stimulus.ActivityTypeName, stimulus.Hash), cancellationToken)).ToList(); + var workflowTriggers = (await _mediator.RequestAsync(new FindWorkflowTriggersByName(stimulus.ActivityTypeName, stimulus.Hash), cancellationToken)).ToList(); return workflowTriggers.Select(x => new TriggerWorkflowInstruction(x)); } } \ No newline at end of file diff --git a/src/samples/aspnet/Elsa.Samples.Web1/Program.cs b/src/samples/aspnet/Elsa.Samples.Web1/Program.cs index 2a5817ed3..166e6652a 100644 --- a/src/samples/aspnet/Elsa.Samples.Web1/Program.cs +++ b/src/samples/aspnet/Elsa.Samples.Web1/Program.cs @@ -16,7 +16,6 @@ using Elsa.Modules.Scheduling.Extensions; using Elsa.Modules.Scheduling.Triggers; using Elsa.Persistence.EntityFrameworkCore.Extensions; using Elsa.Persistence.EntityFrameworkCore.Sqlite; -using Elsa.Persistence.Middleware.WorkflowExecution; using Elsa.Pipelines.WorkflowExecution.Components; using Elsa.Runtime.Middleware; using Elsa.Runtime.ProtoActor.Extensions; @@ -29,6 +28,7 @@ using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using WorkflowExecutionBuilderExtensions = Elsa.Runtime.Extensions.WorkflowExecutionBuilderExtensions; var builder = WebApplication.CreateBuilder(args); @@ -49,11 +49,13 @@ services .AddAzureServiceBusServices(options => configuration.GetSection("AzureServiceBus").Bind(options)) .ConfigureWorkflowRuntime(options => { - options.Workflows.Add("HelloWorldWorkflow", new HelloWorldWorkflow()); - options.Workflows.Add("HttpWorkflow", new HttpWorkflow()); - options.Workflows.Add("ForkedHttpWorkflow", new ForkedHttpWorkflow()); - options.Workflows.Add("AzureServiceBusWorkflow", new AzureServiceBusWorkflow()); + // Register workflows. + options.Workflows.Add(nameof(HelloWorldWorkflow), new HelloWorldWorkflow()); + options.Workflows.Add(nameof(HttpWorkflow), new HttpWorkflow()); + options.Workflows.Add(nameof(ForkedHttpWorkflow), new ForkedHttpWorkflow()); options.Workflows.Add(nameof(CompositeActivitiesWorkflow), new CompositeActivitiesWorkflow()); + options.Workflows.Add(nameof(SendMessageWorkflow), new SendMessageWorkflow()); + options.Workflows.Add(nameof(ReceiveMessageWorkflow), new ReceiveMessageWorkflow()); }); // Testing only: allow client app to connect from anywhere. @@ -70,7 +72,7 @@ services .AddActivity() .AddActivity() .AddActivity() - .AddActivity() + .AddActivity() ; // Register available triggers. @@ -95,10 +97,7 @@ wellKnownTypeRegistry.RegisterType("boolean"); wellKnownTypeRegistry.RegisterType("string"); // Configure workflow engine execution pipeline. -serviceProvider.ConfigureDefaultWorkflowExecutionPipeline(pipeline => pipeline - .UsePersistence() - .UseWorkflowExecutionLogPersistence() - .UseWorkflowExecutionEvents() +serviceProvider.ConfigureDefaultWorkflowExecutionPipeline(pipeline => WorkflowExecutionBuilderExtensions.UseWorkflowExecutionEvents(WorkflowExecutionBuilderExtensions.UseWorkflowExecutionLogPersistence(WorkflowExecutionBuilderExtensions.UsePersistence(pipeline))) .UseActivityScheduler() ); diff --git a/src/samples/aspnet/Elsa.Samples.Web1/Workflows/ReceiveMessageWorkflow.cs b/src/samples/aspnet/Elsa.Samples.Web1/Workflows/ReceiveMessageWorkflow.cs new file mode 100644 index 000000000..32ba553e5 --- /dev/null +++ b/src/samples/aspnet/Elsa.Samples.Web1/Workflows/ReceiveMessageWorkflow.cs @@ -0,0 +1,28 @@ +using Elsa.Activities.Console; +using Elsa.Activities.Workflows; +using Elsa.Contracts; +using Elsa.Models; +using Elsa.Modules.AzureServiceBus.Activities; +using Elsa.Modules.AzureServiceBus.Models; +using Elsa.Runtime.Contracts; + +namespace Elsa.Samples.Web1.Workflows; + +public class ReceiveMessageWorkflow : IWorkflow +{ + public void Build(IWorkflowDefinitionBuilder workflow) + { + workflow.AddTrigger(new MessageReceived + { + QueueOrTopic = new Input("inbox") + }); + + workflow.WithRoot(new Sequence + { + Activities = + { + new WriteLine("Message received!") + } + }); + } +} \ No newline at end of file diff --git a/src/samples/aspnet/Elsa.Samples.Web1/Workflows/AzureServiceBusWorkflow.cs b/src/samples/aspnet/Elsa.Samples.Web1/Workflows/SendMessageWorkflow.cs similarity index 84% rename from src/samples/aspnet/Elsa.Samples.Web1/Workflows/AzureServiceBusWorkflow.cs rename to src/samples/aspnet/Elsa.Samples.Web1/Workflows/SendMessageWorkflow.cs index 356ee2633..27e6ec7ba 100644 --- a/src/samples/aspnet/Elsa.Samples.Web1/Workflows/AzureServiceBusWorkflow.cs +++ b/src/samples/aspnet/Elsa.Samples.Web1/Workflows/SendMessageWorkflow.cs @@ -5,11 +5,11 @@ using Elsa.Runtime.Contracts; namespace Elsa.Samples.Web1.Workflows; -public class AzureServiceBusWorkflow : IWorkflow +public class SendMessageWorkflow : IWorkflow { public void Build(IWorkflowDefinitionBuilder workflow) { - workflow.WithRoot(new Send + workflow.WithRoot(new SendMessage { QueueOrTopic = new Input("inbox"), MessageBody = new Input(new { Subject = "Greetings", Message = "Hello World!" }), diff --git a/src/samples/aspnet/Elsa.Samples.Web2/Program.cs b/src/samples/aspnet/Elsa.Samples.Web2/Program.cs index 117678603..0f536b602 100644 --- a/src/samples/aspnet/Elsa.Samples.Web2/Program.cs +++ b/src/samples/aspnet/Elsa.Samples.Web2/Program.cs @@ -2,13 +2,14 @@ using Elsa.Api.Extensions; using Elsa.Extensions; using Elsa.Modules.Http.Extensions; using Elsa.Persistence.InMemory.Extensions; -using Elsa.Persistence.Middleware.WorkflowExecution; using Elsa.Pipelines.WorkflowExecution.Components; +using Elsa.Runtime.Middleware; using Elsa.Runtime.ProtoActor.Extensions; using Elsa.Samples.Web2.Workflows; using Microsoft.AspNetCore.Builder; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; +using WorkflowExecutionBuilderExtensions = Elsa.Runtime.Extensions.WorkflowExecutionBuilderExtensions; var builder = WebApplication.CreateBuilder(args); @@ -26,8 +27,7 @@ builder.Services var app = builder.Build(); // Configure workflow engine execution pipeline. -app.Services.ConfigureDefaultWorkflowExecutionPipeline(pipeline => pipeline - .UsePersistence() +app.Services.ConfigureDefaultWorkflowExecutionPipeline(pipeline => WorkflowExecutionBuilderExtensions.UsePersistence(pipeline) .UseActivityScheduler() ); diff --git a/src/samples/console/Elsa.Samples.Console1/Program.cs b/src/samples/console/Elsa.Samples.Console1/Program.cs index 69946ee3d..d9a7fbc54 100644 --- a/src/samples/console/Elsa.Samples.Console1/Program.cs +++ b/src/samples/console/Elsa.Samples.Console1/Program.cs @@ -7,9 +7,10 @@ using Elsa.Extensions; using Elsa.Mediator.Extensions; using Elsa.Models; using Elsa.Persistence.InMemory.Extensions; -using Elsa.Persistence.Middleware.WorkflowExecution; using Elsa.Pipelines.ActivityExecution.Components; using Elsa.Pipelines.WorkflowExecution.Components; +using Elsa.Runtime.Extensions; +using Elsa.Runtime.Middleware; using Elsa.Samples.Console1.Workflows; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging;