diff --git a/Elsa.sln b/Elsa.sln index ed28d3d5b..b59455ce0 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -1,7 +1,7 @@  Microsoft Visual Studio Solution File, Format Version 12.00 -# Visual Studio Version 16 -VisualStudioVersion = 16.0.29509.3 +# Visual Studio Version 17 +VisualStudioVersion = 17.0.31912.275 MinimumVisualStudioVersion = 10.0.40219.1 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Core", "src\core\Elsa.Core\Elsa.Core.csproj", "{300EE2D5-54C5-46F2-AD03-BB43589EA074}" EndProject @@ -241,8 +241,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "docker", "docker", "{D7F754 docker\Dockerfile = docker\Dockerfile docker\Dockerfile-elsa-dashboard = docker\Dockerfile-elsa-dashboard docker\Dockerfile-elsa-dashboard-and-server = docker\Dockerfile-elsa-dashboard-and-server - docker\README.md = docker\README.md docker\Dockerfile-elsa-server = docker\Dockerfile-elsa-server + docker\README.md = docker\README.md EndProjectSection EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Persistence.EntityFramework.MySql", "src\persistence\Elsa.Persistence.EntityFramework\Elsa.Persistence.EntityFramework.MySql\Elsa.Persistence.EntityFramework.MySql.csproj", "{D55BC08B-CA82-478A-99F7-EDE90AC33160}" @@ -359,23 +359,27 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "ElsaDashboard.Samples.Blazo EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Samples.HttpEndpointSecurity", "src\samples\aspnet\Elsa.Samples.HttpEndpointSecurity\Elsa.Samples.HttpEndpointSecurity.csproj", "{82B115DA-E3D0-49D7-AD08-DE9387656756}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.WatchDirectoryWorker", "src\samples\worker\Elsa.Samples.WatchDirectoryWorker\Elsa.Samples.WatchDirectoryWorker.csproj", "{8EC3CF51-EBC1-4B45-881E-4B1B8190D579}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Samples.WatchDirectoryWorker", "src\samples\worker\Elsa.Samples.WatchDirectoryWorker\Elsa.Samples.WatchDirectoryWorker.csproj", "{8EC3CF51-EBC1-4B45-881E-4B1B8190D579}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.InfiniteLoopDetection", "src\samples\aspnet\Elsa.Samples.InfiniteLoopDetection\Elsa.Samples.InfiniteLoopDetection.csproj", "{A2572DBF-AF11-4A02-AF99-B343871F2D85}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Samples.InfiniteLoopDetection", "src\samples\aspnet\Elsa.Samples.InfiniteLoopDetection\Elsa.Samples.InfiniteLoopDetection.csproj", "{A2572DBF-AF11-4A02-AF99-B343871F2D85}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.Forks", "src\samples\console\Elsa.Samples.Forks\Elsa.Samples.Forks.csproj", "{AB67BA19-5BC2-4C6D-A994-61D51A6A6FD5}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Samples.Forks", "src\samples\console\Elsa.Samples.Forks\Elsa.Samples.Forks.csproj", "{AB67BA19-5BC2-4C6D-A994-61D51A6A6FD5}" EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "retention", "retention", "{235ABC3F-A075-4682-B6D7-837098BA6B00}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Retention", "src\modules\retention\Elsa.Retention\Elsa.Retention.csproj", "{FAEEA170-560C-4AF3-B4A8-92DEBD59D418}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Retention", "src\modules\retention\Elsa.Retention\Elsa.Retention.csproj", "{FAEEA170-560C-4AF3-B4A8-92DEBD59D418}" EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "workflowtesting", "workflowtesting", "{1B0F0880-8390-4A1C-9AE8-D13427626794}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.WorkflowTesting", "src\modules\Elsa.WorkflowTesting\Elsa.WorkflowTesting.csproj", "{0327324E-0B80-4A6A-A7D6-DD521657666F}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.WorkflowTesting", "src\modules\Elsa.WorkflowTesting\Elsa.WorkflowTesting.csproj", "{0327324E-0B80-4A6A-A7D6-DD521657666F}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.WorkflowTesting.Api", "src\modules\Elsa.WorkflowTesting.Api\Elsa.WorkflowTesting.Api.csproj", "{4A4DC44E-D50F-49B8-A9DE-DED26FC05166}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.WorkflowTesting.Api", "src\modules\Elsa.WorkflowTesting.Api\Elsa.WorkflowTesting.Api.csproj", "{4A4DC44E-D50F-49B8-A9DE-DED26FC05166}" EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.UniqueCorrelatedWorkflows", "src\samples\aspnet\Elsa.Samples.UniqueCorrelatedWorkflows\Elsa.Samples.UniqueCorrelatedWorkflows.csproj", "{4CD8293D-4544-4A23-BDEF-15B66D5E924F}" +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Samples.UniqueCorrelatedWorkflows", "src\samples\aspnet\Elsa.Samples.UniqueCorrelatedWorkflows\Elsa.Samples.UniqueCorrelatedWorkflows.csproj", "{4CD8293D-4544-4A23-BDEF-15B66D5E924F}" +EndProject +Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.RabbitMq", "src\activities\Elsa.Activities.RabbitMq\Elsa.Activities.RabbitMq.csproj", "{0D430863-8275-4880-BB09-C307B5EC3837}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.RabbitMqWorker", "src\samples\worker\Elsa.Samples.RabbitMqWorker\Elsa.Samples.RabbitMqWorker.csproj", "{67C58C2B-B873-47C8-B5B9-93B0587EE634}" EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution @@ -891,6 +895,14 @@ Global {4CD8293D-4544-4A23-BDEF-15B66D5E924F}.Debug|Any CPU.Build.0 = Debug|Any CPU {4CD8293D-4544-4A23-BDEF-15B66D5E924F}.Release|Any CPU.ActiveCfg = Release|Any CPU {4CD8293D-4544-4A23-BDEF-15B66D5E924F}.Release|Any CPU.Build.0 = Release|Any CPU + {0D430863-8275-4880-BB09-C307B5EC3837}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {0D430863-8275-4880-BB09-C307B5EC3837}.Debug|Any CPU.Build.0 = Debug|Any CPU + {0D430863-8275-4880-BB09-C307B5EC3837}.Release|Any CPU.ActiveCfg = Release|Any CPU + {0D430863-8275-4880-BB09-C307B5EC3837}.Release|Any CPU.Build.0 = Release|Any CPU + {67C58C2B-B873-47C8-B5B9-93B0587EE634}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {67C58C2B-B873-47C8-B5B9-93B0587EE634}.Debug|Any CPU.Build.0 = Debug|Any CPU + {67C58C2B-B873-47C8-B5B9-93B0587EE634}.Release|Any CPU.ActiveCfg = Release|Any CPU + {67C58C2B-B873-47C8-B5B9-93B0587EE634}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -1059,6 +1071,8 @@ Global {0327324E-0B80-4A6A-A7D6-DD521657666F} = {1B0F0880-8390-4A1C-9AE8-D13427626794} {4A4DC44E-D50F-49B8-A9DE-DED26FC05166} = {1B0F0880-8390-4A1C-9AE8-D13427626794} {4CD8293D-4544-4A23-BDEF-15B66D5E924F} = {22E75696-6FE9-436A-9097-EE21C603F818} + {0D430863-8275-4880-BB09-C307B5EC3837} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180} + {67C58C2B-B873-47C8-B5B9-93B0587EE634} = {E42743A0-FBDD-4150-9D53-6000496D9B87} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158} diff --git a/src/activities/Elsa.Activities.RabbitMq/Activities/IRabbitMqActivity.cs b/src/activities/Elsa.Activities.RabbitMq/Activities/IRabbitMqActivity.cs new file mode 100644 index 000000000..545a92c9c --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Activities/IRabbitMqActivity.cs @@ -0,0 +1,12 @@ +using Elsa.Services; +using System.Collections.Generic; + +namespace Elsa.Activities.RabbitMq +{ + public interface IRabbitMqActivity : IActivity + { + string ConnectionString { get; set; } + string RoutingKey { get; set; } + Dictionary Headers { get; set; } + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceived.cs b/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceived.cs new file mode 100644 index 000000000..a9f07c051 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceived.cs @@ -0,0 +1,87 @@ +using Elsa.Activities.RabbitMq.Configuration; +using Elsa.Activities.RabbitMq.Services; +using Elsa.ActivityResults; +using Elsa.Attributes; +using Elsa.Design; +using Elsa.Expressions; +using Elsa.Services; +using Elsa.Services.Models; +using Rebus.Messages; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq +{ + [Trigger( + Category = "RabbitMQ", + DisplayName = "RabbitMQ Message Received", + Description = "Triggers when RabbitMQ message matching specified routing key is received", + Outcomes = new[] { OutcomeNames.Done } + )] + public class RabbitMqMessageReceived : Activity, IRabbitMqActivity + { + private readonly IMessageReceiverClientFactory _messageReceiverClientFactory; + + public RabbitMqMessageReceived(IMessageReceiverClientFactory messageReceiverClientFactory) + { + _messageReceiverClientFactory = messageReceiverClientFactory; + } + + [ActivityInput( + Hint = "Routing Key", + Order = 1, + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string RoutingKey { get; set; } = default!; + + [ActivityInput( + Hint = "List of headers that should be present in the message", + Order = 2, + UIHint = ActivityInputUIHints.Dictionary, + DefaultSyntax = SyntaxNames.Json, + SupportedSyntaxes = new[] { SyntaxNames.Json })] + public Dictionary Headers { get; set; } = new Dictionary(); + + [ActivityInput( + Hint = "RabbitMQ connection string", + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid }, + Order = 2, + Category = PropertyCategories.Configuration)] + public string ConnectionString { get; set; } = default!; + + + [ActivityOutput(Hint = "Received message")] + public object? Output { get; set; } + + protected override async ValueTask OnExecuteAsync(ActivityExecutionContext context) => context.WorkflowExecutionContext.IsFirstPass ? ExecuteInternalAsync(context) : await SuspendInternalAsync(); + + protected override IActivityExecutionResult OnResume(ActivityExecutionContext context) => ExecuteInternalAsync(context); + + private async ValueTask SuspendInternalAsync() + { + await StartClient(); + + return Suspend(); + } + + private IActivityExecutionResult ExecuteInternalAsync(ActivityExecutionContext context) + { + var message = (TransportMessage)context.Input!; + + var messageBody = System.Text.Encoding.UTF8.GetString(message.Body); + + Output = messageBody; + + context.LogOutputProperty(this, nameof(Output), Output); + context.JournalData.Add("Headers", message.Headers); + + return Done(); + } + private async Task StartClient() + { + var config = new RabbitMqBusConfiguration(ConnectionString, RoutingKey, Headers); + + var client = await _messageReceiverClientFactory.GetReceiverAsync(config); + client.StartClient(); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceivedBuilderExtensions.cs b/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceivedBuilderExtensions.cs new file mode 100644 index 000000000..fdf03eb03 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceivedBuilderExtensions.cs @@ -0,0 +1,16 @@ +using Elsa.Builders; +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; + +namespace Elsa.Activities.RabbitMq +{ + public static class RabbitMqMessageReceivedBuilderExtensions + { + public static IActivityBuilder MessageReceived(this IBuilder builder, Action> setup, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.Then(setup, null, lineNumber, sourceFile); + + public static IActivityBuilder MessageReceived(this IBuilder builder, string connectionString, string routingKey, Dictionary headers, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.MessageReceived(setup => setup.WithConnectionString(connectionString).WithRoutingKey(routingKey).WithHeaders(headers), lineNumber, sourceFile); + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceivedExtensions.cs b/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceivedExtensions.cs new file mode 100644 index 000000000..ea8a74756 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceivedExtensions.cs @@ -0,0 +1,29 @@ +using Elsa.Builders; +using Elsa.Services.Models; +using System; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq +{ + public static class RabbitMqMessageReceivedExtensions + { + public static ISetupActivity WithConnectionString(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.ConnectionString, value!); + public static ISetupActivity WithConnectionString(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.ConnectionString, value!); + public static ISetupActivity WithConnectionString(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.ConnectionString, value!); + public static ISetupActivity WithConnectionString(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.ConnectionString, value!); + public static ISetupActivity WithConnectionString(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.ConnectionString, value!); + + public static ISetupActivity WithRoutingKey(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.RoutingKey, value!); + public static ISetupActivity WithRoutingKey(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.RoutingKey, value!); + public static ISetupActivity WithRoutingKey(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.RoutingKey, value!); + public static ISetupActivity WithRoutingKey(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.RoutingKey, value!); + public static ISetupActivity WithRoutingKey(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.RoutingKey, value!); + + public static ISetupActivity WithHeaders(this ISetupActivity messageReceived, Func>> value) => messageReceived.Set(x => x.Headers, value!); + public static ISetupActivity WithHeaders(this ISetupActivity messageReceived, Func>> value) => messageReceived.Set(x => x.Headers, value!); + public static ISetupActivity WithHeaders(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Headers, value!); + public static ISetupActivity WithHeaders(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Headers, value!); + public static ISetupActivity WithHeaders(this ISetupActivity messageReceived, Dictionary value) => messageReceived.Set(x => x.Headers, value!); + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessage.cs b/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessage.cs new file mode 100644 index 000000000..fa7c0e070 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessage.cs @@ -0,0 +1,68 @@ +using Elsa.Activities.RabbitMq.Configuration; +using Elsa.Activities.RabbitMq.Services; +using Elsa.ActivityResults; +using Elsa.Attributes; +using Elsa.Design; +using Elsa.Expressions; +using Elsa.Services; +using Elsa.Services.Models; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq +{ + [Trigger( + Category = "RabbitMQ", + DisplayName = "Send RabbitMQ message", + Description = "Send Message to RabbitMQ", + Outcomes = new[] { OutcomeNames.Done } + )] + public class SendRabbitMqMessage : Activity, IRabbitMqActivity + { + private readonly IMessageSenderClientFactory _messageSenderClientFactory; + + public SendRabbitMqMessage(IMessageSenderClientFactory messageSenderClientFactory) + { + _messageSenderClientFactory = messageSenderClientFactory; + } + + [ActivityInput( + Hint = "Topic", + Order = 1, + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string RoutingKey { get; set; } = default!; + + [ActivityInput( + Hint = "List of headers that should be present in the message", + Order = 2, + UIHint = ActivityInputUIHints.Dictionary, + DefaultSyntax = SyntaxNames.Json, + SupportedSyntaxes = new[] { SyntaxNames.Json })] + public Dictionary Headers { get; set; } = new Dictionary(); + + [ActivityInput( + Hint = "Message body", + Order = 3, + UIHint = ActivityInputUIHints.MultiLine, + SupportedSyntaxes = new[] { SyntaxNames.Json })] + public string Message { get; set; } = default!; + + [ActivityInput( + Hint = "RabbitMQ connection string", + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid }, + Order = 1, + Category = PropertyCategories.Configuration)] + public string ConnectionString { get; set; } = default!; + + protected override async ValueTask OnExecuteAsync(ActivityExecutionContext context) + { + var config = new RabbitMqBusConfiguration(ConnectionString, RoutingKey, Headers); + + var client = await _messageSenderClientFactory.GetSenderAsync(config); + + await client.PublishMessage(Message); + + return Done(); + } + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessageBuilderExtensions.cs b/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessageBuilderExtensions.cs new file mode 100644 index 000000000..b4ba340e7 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessageBuilderExtensions.cs @@ -0,0 +1,27 @@ +using Elsa.Builders; +using Elsa.Services.Models; +using System; +using System.Collections.Generic; +using System.Runtime.CompilerServices; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq +{ + public static class SendRabbitMqMessageBuilderExtensions + { + public static IActivityBuilder SendTopicMessage(this IBuilder builder, Action> setup, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.Then(setup, null, lineNumber, sourceFile); + + public static IActivityBuilder SendTopicMessage(this IBuilder builder, string connectionString, string topic, Dictionary headers, Func> message, [CallerLineNumber] int lineNumber = default, + [CallerFilePath] string? sourceFile = default) => builder.SendTopicMessage(setup => setup.WithConnectionString(connectionString).WithTopic(topic).WithHeaders(headers).WithMessage(message), lineNumber, sourceFile); + + public static IActivityBuilder SendTopicMessage(this IBuilder builder, string connectionString, string topic, Dictionary headers, Func message, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.SendTopicMessage(setup => setup.WithConnectionString(connectionString).WithTopic(topic).WithHeaders(headers).WithMessage(message), lineNumber, sourceFile); + + public static IActivityBuilder SendTopicMessage(this IBuilder builder, string connectionString, string topic, Dictionary headers, Func message, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.SendTopicMessage(setup => setup.WithConnectionString(connectionString).WithTopic(topic).WithHeaders(headers).WithMessage(message), lineNumber, sourceFile); + + public static IActivityBuilder SendTopicMessage(this IBuilder builder, string connectionString, string topic, Dictionary headers, string message, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.SendTopicMessage(setup => setup.WithConnectionString(connectionString).WithTopic(topic).WithHeaders(headers).WithMessage(message), lineNumber, sourceFile); + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessageExtensions.cs b/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessageExtensions.cs new file mode 100644 index 000000000..e83694d6f --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessageExtensions.cs @@ -0,0 +1,35 @@ +using Elsa.Builders; +using Elsa.Services.Models; +using System; +using System.Collections.Generic; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq +{ + public static class SendRabbitMqMessageExtensions + { + public static ISetupActivity WithConnectionString(this ISetupActivity sendMessage, Func> value) => sendMessage.Set(x => x.ConnectionString, value!); + public static ISetupActivity WithConnectionString(this ISetupActivity sendMessage, Func> value) => sendMessage.Set(x => x.ConnectionString, value!); + public static ISetupActivity WithConnectionString(this ISetupActivity sendMessage, Func value) => sendMessage.Set(x => x.ConnectionString, value!); + public static ISetupActivity WithConnectionString(this ISetupActivity sendMessage, Func value) => sendMessage.Set(x => x.ConnectionString, value!); + public static ISetupActivity WithConnectionString(this ISetupActivity sendMessage, string value) => sendMessage.Set(x => x.ConnectionString, value!); + + public static ISetupActivity WithTopic(this ISetupActivity sendMessage, Func> value) => sendMessage.Set(x => x.RoutingKey, value!); + public static ISetupActivity WithTopic(this ISetupActivity sendMessage, Func> value) => sendMessage.Set(x => x.RoutingKey, value!); + public static ISetupActivity WithTopic(this ISetupActivity sendMessage, Func value) => sendMessage.Set(x => x.RoutingKey, value!); + public static ISetupActivity WithTopic(this ISetupActivity sendMessage, Func value) => sendMessage.Set(x => x.RoutingKey, value!); + public static ISetupActivity WithTopic(this ISetupActivity sendMessage, string value) => sendMessage.Set(x => x.RoutingKey, value!); + + public static ISetupActivity WithHeaders(this ISetupActivity sendMessage, Func>> value) => sendMessage.Set(x => x.Headers, value!); + public static ISetupActivity WithHeaders(this ISetupActivity sendMessage, Func>> value) => sendMessage.Set(x => x.Headers, value!); + public static ISetupActivity WithHeaders(this ISetupActivity sendMessage, Func> value) => sendMessage.Set(x => x.Headers, value!); + public static ISetupActivity WithHeaders(this ISetupActivity sendMessage, Func> value) => sendMessage.Set(x => x.Headers, value!); + public static ISetupActivity WithHeaders(this ISetupActivity sendMessage, Dictionary value) => sendMessage.Set(x => x.Headers, value!); + + public static ISetupActivity WithMessage(this ISetupActivity sendMessage, Func> value) => sendMessage.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity sendMessage, Func> value) => sendMessage.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity sendMessage, Func value) => sendMessage.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity sendMessage, Func value) => sendMessage.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity sendMessage, string value) => sendMessage.Set(x => x.Message, value!); + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Bookmarks/MessageReceivedBookmark.cs b/src/activities/Elsa.Activities.RabbitMq/Bookmarks/MessageReceivedBookmark.cs new file mode 100644 index 000000000..0d45f5329 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Bookmarks/MessageReceivedBookmark.cs @@ -0,0 +1,40 @@ +using Elsa.Services; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Bookmarks +{ + public class MessageReceivedBookmark : IBookmark + { + public MessageReceivedBookmark() + { + } + + public MessageReceivedBookmark(string routingKey, string connectionString, Dictionary headers) + { + RoutingKey = routingKey; + ConnectionString = connectionString; + Headers = headers ?? new Dictionary(); + + } + + public string RoutingKey { get; set; } = default!; + public string ConnectionString { get; set; } = default!; + public Dictionary Headers { get; set; } = default!; + } + + public class QueueMessageReceivedBookmarkProvider : BookmarkProvider + { + public override async ValueTask> GetBookmarksAsync(BookmarkProviderContext context, CancellationToken cancellationToken) => + new[] + { + Result(new MessageReceivedBookmark + { + RoutingKey = (await context.ReadActivityPropertyAsync(x => x.RoutingKey, cancellationToken))!, + ConnectionString = (await context.ReadActivityPropertyAsync(x => x.ConnectionString, cancellationToken))!, + Headers = (await context.ReadActivityPropertyAsync(x => x.Headers, cancellationToken))! + }) + }; + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Configuration/RabbitMqBusConfiguration.cs b/src/activities/Elsa.Activities.RabbitMq/Configuration/RabbitMqBusConfiguration.cs new file mode 100644 index 000000000..95f56db9a --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Configuration/RabbitMqBusConfiguration.cs @@ -0,0 +1,26 @@ +using System.Collections.Generic; +using System.Linq; + +namespace Elsa.Activities.RabbitMq.Configuration +{ + public class RabbitMqBusConfiguration + { + public string ConnectionString { get; } + public string RoutingKey { get; } + public Dictionary Headers { get; } + + public RabbitMqBusConfiguration(string connectionString, string routingKey, Dictionary headers) + { + ConnectionString = connectionString; + RoutingKey = routingKey; + Headers = headers ?? new Dictionary(); + } + + public override int GetHashCode() + { + var headersString = string.Concat(Headers.Select((x, y) => string.Concat(x, y))); + + return System.HashCode.Combine(ConnectionString, RoutingKey, headersString); + } + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Consumers/RestartRabbitMqBusConsumer.cs b/src/activities/Elsa.Activities.RabbitMq/Consumers/RestartRabbitMqBusConsumer.cs new file mode 100644 index 000000000..839c2da97 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Consumers/RestartRabbitMqBusConsumer.cs @@ -0,0 +1,15 @@ +using Elsa.Activities.RabbitMq.Services; +using Elsa.Events; +using Rebus.Handlers; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Consumers +{ + public class RestartRabbitMqBusConsumer : IHandleMessages, IHandleMessages + { + private readonly IRabbitMqQueueStarter _rabbitMqQueueStarter; + public RestartRabbitMqBusConsumer(IRabbitMqQueueStarter rabbitMqQueueStarter) => _rabbitMqQueueStarter = rabbitMqQueueStarter; + public Task Handle(WorkflowDefinitionPublished message) => _rabbitMqQueueStarter.CreateWorkersAsync(); + public Task Handle(WorkflowDefinitionRetracted message) => _rabbitMqQueueStarter.CreateWorkersAsync(); + } +} diff --git a/src/activities/Elsa.Activities.RabbitMq/Elsa.Activities.RabbitMq.csproj b/src/activities/Elsa.Activities.RabbitMq/Elsa.Activities.RabbitMq.csproj new file mode 100644 index 000000000..fa021b9eb --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Elsa.Activities.RabbitMq.csproj @@ -0,0 +1,18 @@ + + + + net5.0 + Elsa.Activities.RabbitMq + latest + + + + + + + + + + + + diff --git a/src/activities/Elsa.Activities.RabbitMq/Extensions/ServiceCollectionExtensions.cs b/src/activities/Elsa.Activities.RabbitMq/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 000000000..3b99197c9 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Extensions/ServiceCollectionExtensions.cs @@ -0,0 +1,39 @@ +using Elsa.Activities.RabbitMq.Bookmarks; +using Elsa.Activities.RabbitMq.Consumers; +using Elsa.Activities.RabbitMq.Services; +using Elsa.Activities.RabbitMq.StartupTasks; +using Elsa.Activities.RabbitMq.Testing; +using Elsa.Events; +using Elsa.Options; +using Elsa.Runtime; +using Elsa.Services; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Activities.RabbitMq +{ + public static class ServiceCollectionExtensions + { + public static ElsaOptionsBuilder AddRabbitMqActivities(this ElsaOptionsBuilder options) + { + options.Services + .AddSingleton() + .AddSingleton(sp => sp.GetRequiredService()) + .AddSingleton(sp => sp.GetRequiredService()) + .AddSingleton() + .AddSingleton>() + .AddSingleton() + .AddNotificationHandlersFrom() + .AddStartupTask() + .AddBookmarkProvider(); + + options.AddPubSubConsumer("WorkflowDefinitionEvents"); + options.AddPubSubConsumer("WorkflowDefinitionEvents"); + + options + .AddActivity() + .AddActivity(); + + return options; + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Services/BusClientFactory.cs b/src/activities/Elsa.Activities.RabbitMq/Services/BusClientFactory.cs new file mode 100644 index 000000000..fe4bfb371 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Services/BusClientFactory.cs @@ -0,0 +1,91 @@ +using Elsa.Activities.RabbitMq.Configuration; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Services +{ + public class BusClientFactory : IMessageReceiverClientFactory, IMessageSenderClientFactory + { + private readonly IDictionary _receivers = new Dictionary(); + private readonly IDictionary _senders = new Dictionary(); + + private readonly SemaphoreSlim _semaphore = new(1); + + public async Task GetReceiverAsync(RabbitMqBusConfiguration configuration, CancellationToken cancellationToken) + { + await _semaphore.WaitAsync(cancellationToken); + + try + { + if (_receivers.TryGetValue(configuration.GetHashCode(), out var messageReceiver)) + return messageReceiver; + + var newMessageReceiver = new Client(configuration); + _receivers.Add(configuration.GetHashCode(), newMessageReceiver); + + return newMessageReceiver; + } + finally + { + _semaphore.Release(); + } + } + + public async Task DisposeReceiverAsync(IClient receiverClient, CancellationToken cancellationToken = default) + { + await _semaphore.WaitAsync(cancellationToken); + + var key = GetKeyFor(receiverClient); + + try + { + _receivers.Remove(key); + receiverClient.Dispose(); + } + finally + { + _semaphore.Release(); + } + } + + public async Task GetSenderAsync(RabbitMqBusConfiguration configuration, CancellationToken cancellationToken) + { + await _semaphore.WaitAsync(cancellationToken); + + try + { + if (_senders.TryGetValue(configuration.GetHashCode(), out var messageSender)) + return messageSender; + + var newMessageSender = new Client(configuration); + _senders.Add(configuration.GetHashCode(), newMessageSender); + + return newMessageSender; + } + finally + { + _semaphore.Release(); + } + } + + public async Task DisposeSenderAsync(IClient senderClient, CancellationToken cancellationToken = default) + { + await _semaphore.WaitAsync(cancellationToken); + + var key = GetKeyFor(senderClient); + + try + { + _senders.Remove(key); + senderClient.Dispose(); + } + finally + { + _semaphore.Release(); + } + } + + private static int GetKeyFor(IClient client) => client.Configuration.GetHashCode(); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Services/Client.cs b/src/activities/Elsa.Activities.RabbitMq/Services/Client.cs new file mode 100644 index 000000000..3a9a52f93 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Services/Client.cs @@ -0,0 +1,86 @@ +using Elsa.Activities.RabbitMq.Configuration; +using Rebus.Activation; +using Rebus.Bus; +using Rebus.Config; +using Rebus.Messages; +using Rebus.Routing.TransportMessages; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Services +{ + public class Client : IClient + { + private BuiltinHandlerActivator _activator; + private IBus _bus; + + public RabbitMqBusConfiguration Configuration { get; } + + public Client(RabbitMqBusConfiguration configuration) + { + Configuration = configuration; + _activator = new BuiltinHandlerActivator(); + } + + public void SubscribeWithHandler(Func handler) + { + if (_bus != null) return; + + _bus = Configure + .With(_activator) + .Options(o => o.SetNumberOfWorkers(0)) + .Routing(r => r.AddTransportMessageForwarder(async transportMessage => + { + if (!AllHeadersMatch(transportMessage.Headers)) return ForwardAction.None; + + await handler(transportMessage, CancellationToken.None); + + return ForwardAction.Ignore(); + })) + .Transport(t => + { + t.UseRabbitMq(Configuration.ConnectionString, $"Elsa{Guid.NewGuid().ToString("n").ToUpper()}").InputQueueOptions(o => o.SetAutoDelete(autoDelete: true)); + }) + .Start(); + + _bus.Advanced.Topics.Subscribe(Configuration.RoutingKey); + } + + public async Task PublishMessage(string message) + { + if (_bus == null) ConfigureAsOneWayClient(); + + await _bus.Advanced.Topics.Publish(Configuration.RoutingKey, message, Configuration.Headers); + } + + public void Dispose() + { + _activator.Dispose(); + } + + public void StartClient() + { + if (_bus.Advanced.Workers.Count == 0) + _bus.Advanced.Workers.SetNumberOfWorkers(1); + } + + public void StopClient() + { + if (_bus.Advanced.Workers.Count == 1) + _bus.Advanced.Workers.SetNumberOfWorkers(0); + } + + private void ConfigureAsOneWayClient() + { + _bus = Configure + .With(_activator) + .Transport(t => t.UseRabbitMqAsOneWayClient(Configuration.ConnectionString).InputQueueOptions(o => o.SetAutoDelete(autoDelete: true))) + .Start(); + } + + private bool AllHeadersMatch(Dictionary messageHeaders) => Configuration.Headers.All(x => messageHeaders.ContainsKey(x.Key) && messageHeaders[x.Key] == x.Value); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Services/IClient.cs b/src/activities/Elsa.Activities.RabbitMq/Services/IClient.cs new file mode 100644 index 000000000..028d7deb4 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Services/IClient.cs @@ -0,0 +1,18 @@ +using Elsa.Activities.RabbitMq.Configuration; +using Rebus.Messages; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Services +{ + public interface IClient + { + RabbitMqBusConfiguration Configuration { get; } + void SubscribeWithHandler(Func handler); + Task PublishMessage(string message); + void StartClient(); + void StopClient(); + void Dispose(); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Services/IMessageReceiverClientFactory.cs b/src/activities/Elsa.Activities.RabbitMq/Services/IMessageReceiverClientFactory.cs new file mode 100644 index 000000000..d73e14c40 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Services/IMessageReceiverClientFactory.cs @@ -0,0 +1,12 @@ +using Elsa.Activities.RabbitMq.Configuration; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Services +{ + public interface IMessageReceiverClientFactory + { + Task GetReceiverAsync(RabbitMqBusConfiguration config, CancellationToken cancellationToken = default); + Task DisposeReceiverAsync(IClient receiverClient, CancellationToken cancellationToken = default); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Services/IMessageSenderClientFactory.cs b/src/activities/Elsa.Activities.RabbitMq/Services/IMessageSenderClientFactory.cs new file mode 100644 index 000000000..e0ef34efe --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Services/IMessageSenderClientFactory.cs @@ -0,0 +1,12 @@ +using Elsa.Activities.RabbitMq.Configuration; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Services +{ + public interface IMessageSenderClientFactory + { + Task GetSenderAsync(RabbitMqBusConfiguration config, CancellationToken cancellationToken = default); + Task DisposeSenderAsync(IClient senderClient, CancellationToken cancellationToken = default); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Services/IRabbitMqQueueStarter.cs b/src/activities/Elsa.Activities.RabbitMq/Services/IRabbitMqQueueStarter.cs new file mode 100644 index 000000000..1d687ff1b --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Services/IRabbitMqQueueStarter.cs @@ -0,0 +1,16 @@ +using Elsa.Activities.RabbitMq.Configuration; +using Elsa.Services.Models; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Services +{ + public interface IRabbitMqQueueStarter + { + Task CreateWorkersAsync(CancellationToken cancellationToken = default); + Task CreateWorkerAsync(RabbitMqBusConfiguration config, CancellationToken cancellationToken = default); + IAsyncEnumerable GetConfigurationsAsync(Func? predicate, CancellationToken cancellationToken) where T : IRabbitMqActivity; + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Services/RabbitMqQueueStarter.cs b/src/activities/Elsa.Activities.RabbitMq/Services/RabbitMqQueueStarter.cs new file mode 100644 index 000000000..36ef828a4 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Services/RabbitMqQueueStarter.cs @@ -0,0 +1,115 @@ +using Elsa.Activities.RabbitMq.Configuration; +using Elsa.Services; +using Elsa.Services.Models; +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Services +{ + public class RabbitMqQueueStarter : IRabbitMqQueueStarter + { + private readonly SemaphoreSlim _semaphore = new(1); + private readonly IServiceScopeFactory _scopeFactory; + private readonly IServiceProvider _serviceProvider; + private readonly ICollection _workers; + private readonly IMessageReceiverClientFactory _messageReceiverClientFactory; + private readonly ILogger _logger; + + public RabbitMqQueueStarter( + IConfiguration configuration, + IMessageReceiverClientFactory messageReceiverClientFactory, + IServiceScopeFactory scopeFactory, + IServiceProvider serviceProvider, + ILogger logger) + { + _messageReceiverClientFactory = messageReceiverClientFactory; + _scopeFactory = scopeFactory; + _serviceProvider = serviceProvider; + _logger = logger; + _workers = new List(); + } + + public async Task CreateWorkersAsync(CancellationToken cancellationToken = default) + { + await _semaphore.WaitAsync(cancellationToken); + + try + { + await DisposeExistingWorkersAsync(); + + var receiverConfigs = (await GetConfigurationsAsync(null, cancellationToken).ToListAsync(cancellationToken)).GroupBy(c => c.GetHashCode()).Select(x => x.First()); + + foreach (var config in receiverConfigs) + { + try + { + _workers.Add(await CreateWorkerAsync(config, cancellationToken)); + } + catch (Exception e) + { + _logger.LogWarning(e, "Failed to create a receiver for routing key {RoutingKey}", config.RoutingKey); + } + } + } + finally + { + _semaphore.Release(); + } + } + + public async Task CreateWorkerAsync(RabbitMqBusConfiguration config, CancellationToken cancellationToken = default) + { + var receiver = await _messageReceiverClientFactory.GetReceiverAsync(config, cancellationToken); + return ActivatorUtilities.CreateInstance(_serviceProvider, (Func)DisposeWorkerAsync, receiver); + } + + private async Task DisposeWorkerAsync(IClient messageReceiver) => await _messageReceiverClientFactory.DisposeReceiverAsync(messageReceiver); + + public async IAsyncEnumerable GetConfigurationsAsync(Func? predicate, [EnumeratorCancellation] CancellationToken cancellationToken) where T : IRabbitMqActivity + { + using var scope = _scopeFactory.CreateScope(); + var workflowRegistry = scope.ServiceProvider.GetRequiredService(); + var workflowBlueprintReflector = scope.ServiceProvider.GetRequiredService(); + var workflows = await workflowRegistry.ListActiveAsync(cancellationToken); + + var query = + from workflow in workflows + from activity in workflow.Activities + where activity.Type == typeof(T).Name + select workflow; + + var filteredQuery = predicate == null ? query : query.Where(predicate); + + foreach (var workflow in filteredQuery) + { + var workflowBlueprintWrapper = await workflowBlueprintReflector.ReflectAsync(scope.ServiceProvider, workflow, cancellationToken); + + foreach (var activity in workflowBlueprintWrapper.Filter()) + { + var connectionString = await activity.EvaluatePropertyValueAsync(x => x.ConnectionString, cancellationToken); + var routingKey = await activity.EvaluatePropertyValueAsync(x => x.RoutingKey, cancellationToken); + var headers = await activity.EvaluatePropertyValueAsync(x => x.Headers, cancellationToken); + + var config = new RabbitMqBusConfiguration(connectionString, routingKey, headers); + + yield return config!; + } + } + } + private async Task DisposeExistingWorkersAsync() + { + foreach (var worker in _workers.ToList()) + { + await worker.DisposeAsync(); + _workers.Remove(worker); + } + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Services/Worker.cs b/src/activities/Elsa.Activities.RabbitMq/Services/Worker.cs new file mode 100644 index 000000000..1452d0501 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Services/Worker.cs @@ -0,0 +1,60 @@ +using Elsa.Activities.RabbitMq.Bookmarks; +using Elsa.Models; +using Elsa.Services; +using Elsa.Services.Models; +using Microsoft.Extensions.Logging; +using Rebus.Messages; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Services +{ + public class Worker : IAsyncDisposable + { + private readonly IClient _client; + private readonly ILogger _logger; + private readonly Func _disposeReceiverAction; + private readonly Scoped _workflowLaunchpad; + private string ActivityType => nameof(RabbitMqMessageReceived); + private TimeSpan _delay = TimeSpan.FromMilliseconds(200); + + public Worker( + Scoped workflowLaunchpad, + IClient client, + Func disposeReceiverAction, + ILogger logger) + { + _client = client; + _disposeReceiverAction = disposeReceiverAction; + _logger = logger; + _workflowLaunchpad = workflowLaunchpad; + + _client.SubscribeWithHandler(OnMessageReceived); + } + + public async ValueTask DisposeAsync() => await _disposeReceiverAction(_client); + + private async Task OnMessageReceived(TransportMessage message, CancellationToken cancellationToken) + { + _logger.LogDebug("Message received for routing key {RoutingKey}", _client.Configuration.RoutingKey); + + await TriggerWorkflowsAsync(message, cancellationToken); + + _client.StopClient(); + } + + private async Task TriggerWorkflowsAsync(TransportMessage message, CancellationToken cancellationToken) + { + //avoid handler being triggered earlier than workflow is suspended + await Task.Delay(_delay, cancellationToken); + + var config = _client.Configuration; + + var bookmark = new MessageReceivedBookmark(config.RoutingKey, config.ConnectionString, config.Headers); + var launchContext = new WorkflowsQuery(ActivityType, bookmark); + + await _workflowLaunchpad.UseServiceAsync(service => service.CollectAndDispatchWorkflowsAsync(launchContext, new WorkflowInput(message), cancellationToken)); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Startup.cs b/src/activities/Elsa.Activities.RabbitMq/Startup.cs new file mode 100644 index 000000000..1bf1a7861 --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Startup.cs @@ -0,0 +1,16 @@ +using Elsa.Attributes; +using Elsa.Options; +using Elsa.Services.Startup; +using Microsoft.Extensions.Configuration; + +namespace Elsa.Activities.RabbitMq +{ + [Feature("RabbitMq")] + public class Startup : StartupBase + { + public override void ConfigureElsa(ElsaOptionsBuilder elsa, IConfiguration configuration) + { + elsa.AddRabbitMqActivities(); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/StartupTasks/StartRabbitMqQueues.cs b/src/activities/Elsa.Activities.RabbitMq/StartupTasks/StartRabbitMqQueues.cs new file mode 100644 index 000000000..b185cf02a --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/StartupTasks/StartRabbitMqQueues.cs @@ -0,0 +1,19 @@ +using System.Threading; +using System.Threading.Tasks; +using Elsa.Activities.RabbitMq.Services; +using Elsa.Services; + +namespace Elsa.Activities.RabbitMq.StartupTasks +{ + public class StartRabbitMqQueues : IStartupTask + { + private readonly IRabbitMqQueueStarter _rabbitMqQueueStarter; + public StartRabbitMqQueues(IRabbitMqQueueStarter rabbitMqQueueStarter) + { + _rabbitMqQueueStarter = rabbitMqQueueStarter; + } + public int Order => 2000; + + public Task ExecuteAsync(CancellationToken stoppingToken) => _rabbitMqQueueStarter.CreateWorkersAsync(stoppingToken); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Testing/Handlers/ConfigureRabbitMqActivitiesForTestHandler.cs b/src/activities/Elsa.Activities.RabbitMq/Testing/Handlers/ConfigureRabbitMqActivitiesForTestHandler.cs new file mode 100644 index 000000000..498dfff0b --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Testing/Handlers/ConfigureRabbitMqActivitiesForTestHandler.cs @@ -0,0 +1,63 @@ +using Elsa.Events; +using Elsa.Testing.Events; +using MediatR; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Testing +{ + public class ConfigureRabbitMqActivitiesForTestHandler : INotificationHandler, INotificationHandler, INotificationHandler, INotificationHandler + { + private readonly IRabbitMqTestQueueManager _rabbitMqTestQueueManager; + + public ConfigureRabbitMqActivitiesForTestHandler(IRabbitMqTestQueueManager rabbitMqTestQueueManager) + { + _rabbitMqTestQueueManager = rabbitMqTestQueueManager; + } + + public async Task Handle(WorkflowExecuting notification, CancellationToken cancellationToken) + { + + var isTest = Convert.ToBoolean(notification.WorkflowExecutionContext.WorkflowInstance.GetMetadata("isTest")); + + if (!isTest) return; + + var workflowId = notification.WorkflowExecutionContext.WorkflowBlueprint.Id; + var workflowInstanceId = notification.WorkflowExecutionContext.WorkflowInstance.Id; + + + await _rabbitMqTestQueueManager.CreateTestWorkersAsync(workflowId, workflowInstanceId, cancellationToken); + } + + public Task Handle(WorkflowFaulted notification, CancellationToken cancellationToken) + { + return HandleTesttWorkflowExecutionFinished(notification, cancellationToken); + } + + public Task Handle(WorkflowCompleted notification, CancellationToken cancellationToken) + { + return HandleTesttWorkflowExecutionFinished(notification, cancellationToken); + } + + public Task Handle(WorkflowTestExecutionStopped notification, CancellationToken cancellationToken) + { + _rabbitMqTestQueueManager.DisposeTestWorkersAsync(notification.WorkflowInstanceId); + + return Task.CompletedTask; + } + + private Task HandleTesttWorkflowExecutionFinished(WorkflowNotification notification, CancellationToken cancellationToken) + { + var isTest = Convert.ToBoolean(notification.WorkflowExecutionContext.WorkflowInstance.GetMetadata("isTest")); + + if (!isTest) return Task.CompletedTask; + + var workflowInstanceId = notification.WorkflowExecutionContext.WorkflowInstance.Id; + + _rabbitMqTestQueueManager.DisposeTestWorkersAsync(workflowInstanceId); + + return Task.CompletedTask; + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Testing/IRabbitMqTestQueueManager.cs b/src/activities/Elsa.Activities.RabbitMq/Testing/IRabbitMqTestQueueManager.cs new file mode 100644 index 000000000..48c9891ec --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Testing/IRabbitMqTestQueueManager.cs @@ -0,0 +1,11 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Testing +{ + public interface IRabbitMqTestQueueManager + { + Task CreateTestWorkersAsync(string workflowId, string workflowInstanceId, CancellationToken cancellationToken = default); + Task DisposeTestWorkersAsync(string workflowInstanceId); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.RabbitMq/Testing/RabbitMqTestQueueManager.cs b/src/activities/Elsa.Activities.RabbitMq/Testing/RabbitMqTestQueueManager.cs new file mode 100644 index 000000000..e226ec3df --- /dev/null +++ b/src/activities/Elsa.Activities.RabbitMq/Testing/RabbitMqTestQueueManager.cs @@ -0,0 +1,72 @@ +using Elsa.Activities.RabbitMq.Services; +using Microsoft.Extensions.Logging; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.RabbitMq.Testing +{ + public class RabbitMqTestQueueManager : IRabbitMqTestQueueManager + { + private readonly SemaphoreSlim _semaphore = new(1); + private readonly IDictionary> _workers; + private readonly IRabbitMqQueueStarter _rabbitMqQueueStarter; + private readonly ILogger _logger; + + public RabbitMqTestQueueManager( + IRabbitMqQueueStarter rabbitMqQueueStarter, + ILogger logger) + { + _rabbitMqQueueStarter = rabbitMqQueueStarter; + _logger = logger; + _workers = new Dictionary>(); + } + + public async Task CreateTestWorkersAsync(string workflowId, string workflowInstanceId, CancellationToken cancellationToken = default) + { + await _semaphore.WaitAsync(cancellationToken); + + try + { + + if (_workers.ContainsKey(workflowInstanceId)) + { + if (_workers[workflowInstanceId].Count > 0) + return; + } + else + _workers[workflowInstanceId] = new List(); + + var workerConfigs = (await _rabbitMqQueueStarter.GetConfigurationsAsync(x => x.Id == workflowId, cancellationToken).ToListAsync(cancellationToken)).Distinct(); + + foreach (var config in workerConfigs) + { + try + { + _workers[workflowInstanceId].Add(await _rabbitMqQueueStarter.CreateWorkerAsync(config, cancellationToken)); + } + catch (Exception e) + { + _logger.LogWarning(e, "Failed to create a test receiver for routing key {RoutingKey}", config.RoutingKey); + } + } + } + finally + { + _semaphore.Release(); + } + } + + public async Task DisposeTestWorkersAsync(string workflowInstance) + { + foreach (var worker in _workers[workflowInstance]) + { + await worker.DisposeAsync(); + } + + _workers[workflowInstance].Clear(); + } + } +} \ No newline at end of file diff --git a/src/core/Elsa.Abstractions/Design/ActivityInputUIHints.cs b/src/core/Elsa.Abstractions/Design/ActivityInputUIHints.cs index e554cba9f..4ba977aeb 100644 --- a/src/core/Elsa.Abstractions/Design/ActivityInputUIHints.cs +++ b/src/core/Elsa.Abstractions/Design/ActivityInputUIHints.cs @@ -10,6 +10,7 @@ namespace Elsa.Design public const string Dropdown = "dropdown"; public const string MultiText = "multi-text"; public const string CodeEditor = "code-editor"; + public const string Dictionary = "dictionary"; /// /// An editor that allows the user to write a blob of JSON. diff --git a/src/core/Elsa.Abstractions/Design/PropertyCategories.cs b/src/core/Elsa.Abstractions/Design/PropertyCategories.cs index 3c2e2d4fe..e7c5abccf 100644 --- a/src/core/Elsa.Abstractions/Design/PropertyCategories.cs +++ b/src/core/Elsa.Abstractions/Design/PropertyCategories.cs @@ -4,5 +4,6 @@ { public const string Advanced = "Advanced"; public const string Hosting = "Hosting"; + public const string Configuration = "Configuration"; } } \ No newline at end of file diff --git a/src/designer/elsa-workflows-studio/src/components.d.ts b/src/designer/elsa-workflows-studio/src/components.d.ts index 4811d4a43..4b4285ec2 100644 --- a/src/designer/elsa-workflows-studio/src/components.d.ts +++ b/src/designer/elsa-workflows-studio/src/components.d.ts @@ -64,6 +64,12 @@ export namespace Components { "selectedActivityIds": Array; "showActivityEditor": (activity: ActivityModel, animate: boolean) => Promise; } + interface ElsaDictionaryProperty { + "activityModel": ActivityModel; + "propertyDescriptor": ActivityPropertyDescriptor; + "propertyModel": ActivityDefinitionProperty; + "serverUrl": string; + } interface ElsaDropdownButton { "icon"?: any; "items": Array; @@ -408,6 +414,12 @@ declare global { prototype: HTMLElsaDesignerTreeElement; new (): HTMLElsaDesignerTreeElement; }; + interface HTMLElsaDictionaryPropertyElement extends Components.ElsaDictionaryProperty, HTMLStencilElement { + } + var HTMLElsaDictionaryPropertyElement: { + prototype: HTMLElsaDictionaryPropertyElement; + new (): HTMLElsaDictionaryPropertyElement; + }; interface HTMLElsaDropdownButtonElement extends Components.ElsaDropdownButton, HTMLStencilElement { } var HTMLElsaDropdownButtonElement: { @@ -718,6 +730,7 @@ declare global { "elsa-control": HTMLElsaControlElement; "elsa-copy-button": HTMLElsaCopyButtonElement; "elsa-designer-tree": HTMLElsaDesignerTreeElement; + "elsa-dictionary-property": HTMLElsaDictionaryPropertyElement; "elsa-dropdown-button": HTMLElsaDropdownButtonElement; "elsa-dropdown-property": HTMLElsaDropdownPropertyElement; "elsa-expression-editor": HTMLElsaExpressionEditorElement; @@ -818,6 +831,12 @@ declare namespace LocalJSX { "onWorkflow-changed"?: (event: CustomEvent) => void; "selectedActivityIds"?: Array; } + interface ElsaDictionaryProperty { + "activityModel"?: ActivityModel; + "propertyDescriptor"?: ActivityPropertyDescriptor; + "propertyModel"?: ActivityDefinitionProperty; + "serverUrl"?: string; + } interface ElsaDropdownButton { "icon"?: any; "items"?: Array; @@ -1114,6 +1133,7 @@ declare namespace LocalJSX { "elsa-control": ElsaControl; "elsa-copy-button": ElsaCopyButton; "elsa-designer-tree": ElsaDesignerTree; + "elsa-dictionary-property": ElsaDictionaryProperty; "elsa-dropdown-button": ElsaDropdownButton; "elsa-dropdown-property": ElsaDropdownProperty; "elsa-expression-editor": ElsaExpressionEditor; @@ -1179,6 +1199,7 @@ declare module "@stencil/core" { "elsa-control": LocalJSX.ElsaControl & JSXBase.HTMLAttributes; "elsa-copy-button": LocalJSX.ElsaCopyButton & JSXBase.HTMLAttributes; "elsa-designer-tree": LocalJSX.ElsaDesignerTree & JSXBase.HTMLAttributes; + "elsa-dictionary-property": LocalJSX.ElsaDictionaryProperty & JSXBase.HTMLAttributes; "elsa-dropdown-button": LocalJSX.ElsaDropdownButton & JSXBase.HTMLAttributes; "elsa-dropdown-property": LocalJSX.ElsaDropdownProperty & JSXBase.HTMLAttributes; "elsa-expression-editor": LocalJSX.ElsaExpressionEditor & JSXBase.HTMLAttributes; diff --git a/src/designer/elsa-workflows-studio/src/components/editors/elsa-property-editor/readme.md b/src/designer/elsa-workflows-studio/src/components/editors/elsa-property-editor/readme.md index 9d4d73c6a..911374d83 100644 --- a/src/designer/elsa-workflows-studio/src/components/editors/elsa-property-editor/readme.md +++ b/src/designer/elsa-workflows-studio/src/components/editors/elsa-property-editor/readme.md @@ -31,6 +31,7 @@ - [elsa-check-list-property](../properties/elsa-check-list-property) - [elsa-checkbox-property](../properties/elsa-checkbox-property) + - [elsa-dictionary-property](../properties/elsa-dictionary-property) - [elsa-dropdown-property](../properties/elsa-dropdown-property) - [elsa-json-property](../properties/elsa-json-property) - [elsa-multi-line-property](../properties/elsa-multi-line-property) @@ -51,6 +52,7 @@ graph TD; elsa-expression-editor --> context-consumer elsa-check-list-property --> elsa-property-editor elsa-checkbox-property --> elsa-property-editor + elsa-dictionary-property --> elsa-property-editor elsa-dropdown-property --> elsa-property-editor elsa-json-property --> elsa-property-editor elsa-multi-line-property --> elsa-property-editor diff --git a/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-dictionary-property/elsa-dictionary-property.tsx b/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-dictionary-property/elsa-dictionary-property.tsx new file mode 100644 index 000000000..ab0b1731c --- /dev/null +++ b/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-dictionary-property/elsa-dictionary-property.tsx @@ -0,0 +1,124 @@ +import {Component, h, Host, Prop, State} from '@stencil/core'; +import {ActivityDefinitionProperty, ActivityModel, ActivityPropertyDescriptor, SyntaxNames} from "../../../../models"; +import Tunnel from "../../../../data/workflow-editor"; +import {IconColor, IconName, iconProvider} from "../../../../services/icon-provider" + +@Component({ + tag: 'elsa-dictionary-property', + shadow: false, +}) +export class ElsaDictionaryProperty { + + @Prop() propertyDescriptor: ActivityPropertyDescriptor; + @Prop() propertyModel: ActivityDefinitionProperty; + @Prop() activityModel: ActivityModel; + @Prop({mutable: true}) serverUrl: string; + @State() currentValue: [string, string][]; + + async componentWillLoad() { + this.currentValue = this.jsonToDictionary(this.propertyModel.expressions[SyntaxNames.Json] || null); + if (this.currentValue.length === 0) this.currentValue = [['', '']]; + } + + jsonToDictionary = (json: string): [string, string][] => { + if (!json) return [['', '']]; + + const parsedValue = JSON.parse(json); + + return Object.keys(parsedValue).map(key => [key, parsedValue[key]]); + } + + dictionaryToJson = (dictionary: [string, string][]) => { + const filteredDictionary = this.removeInvalidKeys(dictionary); + + if (filteredDictionary.length === 0) return null; + + return JSON.stringify(Object.fromEntries(filteredDictionary)); + } + + removeInvalidKeys = (dictionary: [string, string][]) => { + const filteredDictionary = []; + + dictionary.forEach(x => { + const key = x[0].trim(); + if (key !== '' && !filteredDictionary.some(y => y[0].trim() === key)) + filteredDictionary.push(x); + }); + + return filteredDictionary; + } + + onRowAdded = () => { + //changing contents of array won't trigger state change, + //need to update the reference by creating new array + this.currentValue = [...this.currentValue, ['', '']]; + } + + onRowDeleted = (index: number) => { + const newValue = this.currentValue.filter((x, i) => i !== index); + + if (newValue.length === 0) newValue.push(['', '']); + + this.currentValue = newValue; + this.propertyModel.expressions[SyntaxNames.Json] = this.dictionaryToJson(newValue); + } + + onDefaultSyntaxValueChanged(e: CustomEvent) { + this.currentValue = this.jsonToDictionary(e.detail); + } + + onKeyChanged(e: Event, index: number) { + const input = e.currentTarget as HTMLInputElement; + this.currentValue[index][0] = input.value + this.propertyModel.expressions[SyntaxNames.Json] = this.dictionaryToJson(this.currentValue); + } + + onValueChanged(e: Event, index: number) { + const input = e.currentTarget as HTMLInputElement; + this.currentValue[index][1] = input.value + this.propertyModel.expressions[SyntaxNames.Json] = this.dictionaryToJson(this.currentValue); + } + + render() { + const propertyDescriptor = this.propertyDescriptor; + const propertyModel = this.propertyModel; + const fieldId = propertyDescriptor.name; + const items = this.currentValue; + + return ( + this.onDefaultSyntaxValueChanged(e)} + single-line={true}> + {items.map((item, index) => { + const keyInputId = `${fieldId}_${index}_key}`; + const valueInputId = `${fieldId}_${index}_value}`; + const [key, value] = item; + const isLast = index === (items.length - 1); + + return ( +
+ this.onKeyChanged(e, index)} + placeholder="Name" + class="disabled:elsa-opacity-50 disabled:elsa-cursor-not-allowed focus:elsa-ring-blue-500 focus:elsa-border-blue-500 elsa-border-gray-300 sm:elsa-text-sm elsa-rounded-md elsa-w-5/12"/> + this.onValueChanged(e, index)} + placeholder="Value" + class="disabled:elsa-opacity-50 disabled:elsa-cursor-not-allowed focus:elsa-ring-blue-500 focus:elsa-border-blue-500 elsa-border-gray-300 sm:elsa-text-sm elsa-rounded-md elsa-w-5/12"/> +
+ + {isLast && } +
+
+ ); + })} + + ); + } +} + +Tunnel.injectProps(ElsaDictionaryProperty, ['serverUrl']); diff --git a/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-dictionary-property/readme.md b/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-dictionary-property/readme.md new file mode 100644 index 000000000..113ce0b57 --- /dev/null +++ b/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-dictionary-property/readme.md @@ -0,0 +1,38 @@ +# elsa-dictionary-property + + + + + + +## Properties + +| Property | Attribute | Description | Type | Default | +| -------------------- | ------------ | ----------- | ---------------------------- | ----------- | +| `propertyDescriptor` | -- | | `ActivityPropertyDescriptor` | `undefined` | +| `propertyModel` | -- | | `ActivityDefinitionProperty` | `undefined` | +| `serverUrl` | `server-url` | | `string` | `undefined` | + + +## Dependencies + +### Depends on + +- [elsa-property-editor](../../elsa-property-editor) +- context-consumer + +### Graph +```mermaid +graph TD; + elsa-dictionary-property --> elsa-property-editor + elsa-dictionary-property --> context-consumer + elsa-property-editor --> elsa-multi-expression-editor + elsa-multi-expression-editor --> elsa-expression-editor + elsa-expression-editor --> elsa-monaco + elsa-expression-editor --> context-consumer + style elsa-dictionary-property fill:#f9f,stroke:#333,stroke-width:4px +``` + +---------------------------------------------- + +*Built with [StencilJS](https://stenciljs.com/)* diff --git a/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-switch-cases-property/elsa-switch-cases-property.tsx b/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-switch-cases-property/elsa-switch-cases-property.tsx index 2a8ae4402..15fc7d013 100644 --- a/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-switch-cases-property/elsa-switch-cases-property.tsx +++ b/src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-switch-cases-property/elsa-switch-cases-property.tsx @@ -8,6 +8,7 @@ import { } from "../../../../models"; import {mapSyntaxToLanguage, parseJson} from "../../../../utils/utils"; import {SwitchCase} from "./models"; +import {IconColor, IconName, iconProvider} from "../../../../services/icon-provider"; @Component({ tag: 'elsa-switch-cases-property', @@ -130,13 +131,7 @@ export class ElsaSwitchCasesProperty { @@ -182,12 +177,7 @@ export class ElsaSwitchCasesProperty { diff --git a/src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-definition-editor-screen/elsa-workflow-definition-editor-screen.tsx b/src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-definition-editor-screen/elsa-workflow-definition-editor-screen.tsx index 4242e4f36..c7c3a8e43 100644 --- a/src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-definition-editor-screen/elsa-workflow-definition-editor-screen.tsx +++ b/src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-definition-editor-screen/elsa-workflow-definition-editor-screen.tsx @@ -505,6 +505,7 @@ export class ElsaWorkflowDefinitionEditorScreen { onTestActivityMessageReceived = async args => { const message = args as WorkflowTestActivityMessage; + if (!!message) { this.workflowInstanceId = message.workflowInstanceId; this.workflowTestActivityMessages = this.workflowTestActivityMessages.filter(x => x.activityId !== message.activityId); diff --git a/src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-test-panel/elsa-workflow-test-panel.tsx b/src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-test-panel/elsa-workflow-test-panel.tsx index bb8fe5388..c4988bac8 100644 --- a/src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-test-panel/elsa-workflow-test-panel.tsx +++ b/src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-test-panel/elsa-workflow-test-panel.tsx @@ -62,6 +62,10 @@ export class ElsaWorkflowTestPanel { this.workflowStarted = false; } + if (message.workflowStatus === 'Suspended'){ + this.workflowStarted = true; + } + if (!this.message){ this.message = message; } @@ -120,6 +124,7 @@ export class ElsaWorkflowTestPanel { if (!!message) { const client = await createElsaClient(this.serverUrl); await client.workflowInstancesApi.delete(message.workflowInstanceId); + await client.workflowTestApi.stop({ workflowInstanceId: message.workflowInstanceId }); } this.message = null; diff --git a/src/designer/elsa-workflows-studio/src/drivers/dictionary-driver.tsx b/src/designer/elsa-workflows-studio/src/drivers/dictionary-driver.tsx new file mode 100644 index 000000000..f5644bffa --- /dev/null +++ b/src/designer/elsa-workflows-studio/src/drivers/dictionary-driver.tsx @@ -0,0 +1,15 @@ +import {PropertyDisplayDriver} from "../services"; +import {ActivityModel, ActivityPropertyDescriptor} from "../models"; +import {h} from "@stencil/core"; +import {getOrCreateProperty} from "../utils/utils"; + +export class DictionaryDriver implements PropertyDisplayDriver { + + display(activity: ActivityModel, property: ActivityPropertyDescriptor) { + const prop = getOrCreateProperty(activity, property.name); + return ; + } + + update(activity: ActivityModel, property: ActivityPropertyDescriptor, form: FormData) { + } +} diff --git a/src/designer/elsa-workflows-studio/src/drivers/index.ts b/src/designer/elsa-workflows-studio/src/drivers/index.ts index 78de3ca12..df4654bfb 100644 --- a/src/designer/elsa-workflows-studio/src/drivers/index.ts +++ b/src/designer/elsa-workflows-studio/src/drivers/index.ts @@ -8,4 +8,5 @@ export * from './multiline-driver'; export * from './null-property-driver'; export * from './radio-list-driver'; export * from './single-line-driver'; -export * from './switch-case-builder-driver'; \ No newline at end of file +export * from './switch-case-builder-driver'; +export * from './dictionary-driver'; diff --git a/src/designer/elsa-workflows-studio/src/globals/tailwind.css b/src/designer/elsa-workflows-studio/src/globals/tailwind.css index c63fca09a..4f6761b54 100644 --- a/src/designer/elsa-workflows-studio/src/globals/tailwind.css +++ b/src/designer/elsa-workflows-studio/src/globals/tailwind.css @@ -2406,6 +2406,10 @@ select { width: 3rem; } +.elsa-w-24 { + width: 6rem; +} + .elsa-w-48 { width: 12rem; } @@ -2430,6 +2434,10 @@ select { width: 25%; } +.elsa-w-5\/12 { + width: 41.666667%; +} + .elsa-w-8\/12 { width: 66.666667%; } diff --git a/src/designer/elsa-workflows-studio/src/index.html b/src/designer/elsa-workflows-studio/src/index.html index 772ade8c6..ab0a97ebc 100644 --- a/src/designer/elsa-workflows-studio/src/index.html +++ b/src/designer/elsa-workflows-studio/src/index.html @@ -70,12 +70,10 @@ // Register a sample middleware. e.service.register({ onRequest(request) { - console.log('onRequest'); return request; }, onResponse(response) { - console.log('onResponse'); return response; } }); diff --git a/src/designer/elsa-workflows-studio/src/plugins/default-drivers-plugin.ts b/src/designer/elsa-workflows-studio/src/plugins/default-drivers-plugin.ts index 8a5f88df3..32c3729a1 100644 --- a/src/designer/elsa-workflows-studio/src/plugins/default-drivers-plugin.ts +++ b/src/designer/elsa-workflows-studio/src/plugins/default-drivers-plugin.ts @@ -1,5 +1,5 @@ import {propertyDisplayManager, ElsaPlugin, PropertyDisplayDriver} from "../services"; -import {JsonDriver, RadioListDriver, CheckboxDriver, CheckListDriver, CodeEditorDriver, DropdownDriver, MultilineDriver, MultiTextDriver, SingleLineDriver, SwitchCaseBuilderDriver} from "../drivers"; +import {JsonDriver, RadioListDriver, CheckboxDriver, CheckListDriver, CodeEditorDriver, DictionaryDriver, DropdownDriver, MultilineDriver, MultiTextDriver, SingleLineDriver, SwitchCaseBuilderDriver} from "../drivers"; import {ElsaStudio} from "../models"; export class DefaultDriversPlugin implements ElsaPlugin { @@ -14,6 +14,7 @@ export class DefaultDriversPlugin implements ElsaPlugin { this.addDriver('multi-text', () => new MultiTextDriver()); this.addDriver('code-editor', () => new CodeEditorDriver()); this.addDriver('switch-case-builder', () => new SwitchCaseBuilderDriver()); + this.addDriver('dictionary', () => new DictionaryDriver()); } addDriver(controlType: string, c: (elsaStudio: ElsaStudio) => T) { diff --git a/src/designer/elsa-workflows-studio/src/services/activity-icon-provider.tsx b/src/designer/elsa-workflows-studio/src/services/activity-icon-provider.tsx index a0f62e540..d74507f1c 100644 --- a/src/designer/elsa-workflows-studio/src/services/activity-icon-provider.tsx +++ b/src/designer/elsa-workflows-studio/src/services/activity-icon-provider.tsx @@ -60,7 +60,9 @@ export class ActivityIconProvider { 'WriteHttpResponse': () => , 'Redirect': () => , 'SendEmail': () => , - 'Webhook': () => + 'Webhook': () => , + 'RabbitMqMessageReceived': () => , + 'SendRabbitMqMessage': () => }; register(activityType: string, icon: string) { diff --git a/src/designer/elsa-workflows-studio/src/services/elsa-client.ts b/src/designer/elsa-workflows-studio/src/services/elsa-client.ts index 16fe3cae2..1e06239e5 100644 --- a/src/designer/elsa-workflows-studio/src/services/elsa-client.ts +++ b/src/designer/elsa-workflows-studio/src/services/elsa-client.ts @@ -124,6 +124,9 @@ export const createElsaClient = async function (serverUrl: string): Promise { await httpClient.post(`v1/workflow-test/restartFromActivity`, request); + }, + stop: async (request) => { + await httpClient.post(`v1/workflow-test/stop`, request); } }, workflowRegistryApi: { @@ -289,6 +292,7 @@ export interface WorkflowTestApi { execute(request: WorkflowTestExecuteRequest): Promise; restartFromActivity(request: WorkflowTestRestartFromActivityRequest): Promise; + stop(request: WorkflowTestStopRequest): Promise; } export interface WorkflowRegistryApi { @@ -388,6 +392,10 @@ export interface WorkflowTestRestartFromActivityRequest { signalRConnectionId: string } +export interface WorkflowTestStopRequest { + workflowInstanceId: string +} + export interface ExportWorkflowResponse { fileName: string; data: Blob; diff --git a/src/designer/elsa-workflows-studio/src/services/icon-provider.tsx b/src/designer/elsa-workflows-studio/src/services/icon-provider.tsx new file mode 100644 index 000000000..030e3e0cb --- /dev/null +++ b/src/designer/elsa-workflows-studio/src/services/icon-provider.tsx @@ -0,0 +1,57 @@ +import {h} from '@stencil/core'; +import {Map} from '../utils/utils'; + +export enum IconName { + Plus = 'plus', + TrashBinOutline = 'trash-bin-outline' +} + +export enum IconColor { + Blue = 'blue', + Gray = 'gray', + Green = 'green', + Red = 'red', + Default = 'currentColor' +} + +export interface IconProviderOptions { + color?: IconColor, + hoverColor?: IconColor +} + +export class IconProvider { + private map: Map<(options?: IconProviderOptions) => any> = { + 'plus': (options?: IconProviderOptions) => + + + + + , + 'trash-bin-outline': (options?: IconProviderOptions) => + + + + + + + }; + + getIcon(name: IconName, options?: IconProviderOptions): any { + const provider = this.map[name]; + + if (!provider) + return undefined; + + return provider(options); + } +} + +export const iconProvider = new IconProvider(); diff --git a/src/modules/Elsa.WorkflowTesting.Api/Endpoints/Stop.cs b/src/modules/Elsa.WorkflowTesting.Api/Endpoints/Stop.cs new file mode 100644 index 000000000..5b6cafa44 --- /dev/null +++ b/src/modules/Elsa.WorkflowTesting.Api/Endpoints/Stop.cs @@ -0,0 +1,44 @@ +using Elsa.Server.Api.ActionFilters; +using Elsa.Testing.Api.Models; +using Elsa.Testing.Events; +using MediatR; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Mvc; +using Swashbuckle.AspNetCore.Annotations; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Testing.Api.Endpoints +{ + [ApiController] + [ApiVersion("1")] + [Route("v{apiVersion:apiVersion}/workflow-test/stop")] + [Produces("application/json")] + public class Stop : Controller + { + private readonly IMediator _mediator; + + public Stop(IMediator mediator) + { + _mediator = mediator; + } + + + [HttpPost] + [ElsaJsonFormatter] + [ProducesResponseType(StatusCodes.Status200OK, Type = typeof(WorkflowTestStopRequest))] + [ProducesResponseType(StatusCodes.Status404NotFound)] + [SwaggerOperation( + Summary = "Stops the specified workflow definition execution in test mode.", + Description = "Stops the specified workflow definition execution in test mode.", + OperationId = "WorkflowTest.Stop", + Tags = new[] { "WorkflowTest" }) + ] + public async Task Handle([FromBody] WorkflowTestStopRequest request, CancellationToken cancellationToken = default) + { + await _mediator.Publish(new WorkflowTestExecutionStopped(request.WorkflowInstanceId), cancellationToken); + + return Ok(); + } + } +} diff --git a/src/modules/Elsa.WorkflowTesting.Api/Handlers/ActivityExecutionResultExecutedHandler.cs b/src/modules/Elsa.WorkflowTesting.Api/Handlers/ActivityExecutionResultExecutedHandler.cs index a3b629a67..7f5844f91 100644 --- a/src/modules/Elsa.WorkflowTesting.Api/Handlers/ActivityExecutionResultExecutedHandler.cs +++ b/src/modules/Elsa.WorkflowTesting.Api/Handlers/ActivityExecutionResultExecutedHandler.cs @@ -1,7 +1,3 @@ -using System; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; using Elsa.ActivityResults; using Elsa.Events; using Elsa.Models; @@ -11,6 +7,10 @@ using Elsa.WorkflowTesting.Services; using MediatR; using Newtonsoft.Json; using Newtonsoft.Json.Linq; +using System; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; namespace Elsa.WorkflowTesting.Api.Handlers { diff --git a/src/modules/Elsa.WorkflowTesting.Api/Handlers/ConfigureWorkflowContextForTestHandler.cs b/src/modules/Elsa.WorkflowTesting.Api/Handlers/ConfigureWorkflowContextForTestHandler.cs index 6d4d5fff1..2385a7d7d 100644 --- a/src/modules/Elsa.WorkflowTesting.Api/Handlers/ConfigureWorkflowContextForTestHandler.cs +++ b/src/modules/Elsa.WorkflowTesting.Api/Handlers/ConfigureWorkflowContextForTestHandler.cs @@ -1,10 +1,10 @@ +using Elsa.Events; +using MediatR; using System; using System.Threading; using System.Threading.Tasks; -using Elsa.Events; -using MediatR; -namespace Elsa.Server.Api.Handlers +namespace Elsa.WorkflowTesting.Handlers { public class ConfigureWorkflowContextForTestHandler : INotificationHandler { diff --git a/src/modules/Elsa.WorkflowTesting.Api/Models/WorkflowTestStopRequest.cs b/src/modules/Elsa.WorkflowTesting.Api/Models/WorkflowTestStopRequest.cs new file mode 100644 index 000000000..012206084 --- /dev/null +++ b/src/modules/Elsa.WorkflowTesting.Api/Models/WorkflowTestStopRequest.cs @@ -0,0 +1,7 @@ +namespace Elsa.Testing.Api.Models +{ + public class WorkflowTestStopRequest + { + public string WorkflowInstanceId { get; init; } + } +} diff --git a/src/modules/Elsa.WorkflowTesting/Elsa.WorkflowTesting.csproj b/src/modules/Elsa.WorkflowTesting/Elsa.WorkflowTesting.csproj index 06d938bf0..c92ed7008 100644 --- a/src/modules/Elsa.WorkflowTesting/Elsa.WorkflowTesting.csproj +++ b/src/modules/Elsa.WorkflowTesting/Elsa.WorkflowTesting.csproj @@ -5,6 +5,10 @@ enable default + + + + diff --git a/src/modules/Elsa.WorkflowTesting/Events/WorkflowTestExecutionStopped.cs b/src/modules/Elsa.WorkflowTesting/Events/WorkflowTestExecutionStopped.cs new file mode 100644 index 000000000..88d1b7246 --- /dev/null +++ b/src/modules/Elsa.WorkflowTesting/Events/WorkflowTestExecutionStopped.cs @@ -0,0 +1,14 @@ +using MediatR; + +namespace Elsa.Testing.Events +{ + public class WorkflowTestExecutionStopped : INotification + { + public string WorkflowInstanceId { get; } + + public WorkflowTestExecutionStopped(string workflowInstanceId) + { + WorkflowInstanceId = workflowInstanceId; + } + } +} diff --git a/src/modules/Elsa.WorkflowTesting/Services/IWorkflowTestService.cs b/src/modules/Elsa.WorkflowTesting/Services/IWorkflowTestService.cs index 7d1f50766..4266b5ac9 100644 --- a/src/modules/Elsa.WorkflowTesting/Services/IWorkflowTestService.cs +++ b/src/modules/Elsa.WorkflowTesting/Services/IWorkflowTestService.cs @@ -1,5 +1,5 @@ -using System.Threading.Tasks; using Elsa.WorkflowTesting.Messages; +using System.Threading.Tasks; namespace Elsa.WorkflowTesting.Services { diff --git a/src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj b/src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj index a177a43a5..976eb2367 100644 --- a/src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj +++ b/src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj @@ -18,6 +18,7 @@ + diff --git a/src/samples/server/Elsa.Samples.Server.Host/Startup.cs b/src/samples/server/Elsa.Samples.Server.Host/Startup.cs index e92fbdd9a..7380bb84d 100644 --- a/src/samples/server/Elsa.Samples.Server.Host/Startup.cs +++ b/src/samples/server/Elsa.Samples.Server.Host/Startup.cs @@ -43,6 +43,7 @@ namespace Elsa.Samples.Server.Host typeof(Elsa.Activities.Email.Startup), typeof(Elsa.Activities.Telnyx.Startup), typeof(Elsa.Activities.File.Startup), + typeof(Elsa.Activities.RabbitMq.Startup), typeof(Persistence.EntityFramework.Sqlite.Startup), typeof(Persistence.EntityFramework.SqlServer.Startup), typeof(Persistence.EntityFramework.MySql.Startup), diff --git a/src/samples/server/Elsa.Samples.Server.Host/appsettings.json b/src/samples/server/Elsa.Samples.Server.Host/appsettings.json index 9309cdd18..6dbd59621 100644 --- a/src/samples/server/Elsa.Samples.Server.Host/appsettings.json +++ b/src/samples/server/Elsa.Samples.Server.Host/appsettings.json @@ -50,6 +50,10 @@ "Enabled": true, "Framework": "EntityFrameworkCore", "ConnectionStringIdentifier": "Sqlite" + }, + "RabbitMq": { + "Enabled": true, + "ConnectionStringIdentifier": "RabbitMq" } }, "WorkflowChannels": { diff --git a/src/samples/worker/Elsa.Samples.RabbitMqWorker/Elsa.Samples.RabbitMqWorker.csproj b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Elsa.Samples.RabbitMqWorker.csproj new file mode 100644 index 000000000..ef8ec1cd3 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Elsa.Samples.RabbitMqWorker.csproj @@ -0,0 +1,45 @@ + + + + Exe + net6.0 + enable + enable + + + + + + + + + + PreserveNewest + true + PreserveNewest + + + PreserveNewest + true + PreserveNewest + + + + + + + + + + + + + + + + true + Never + + + + diff --git a/src/samples/worker/Elsa.Samples.RabbitMqWorker/Program.cs b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Program.cs new file mode 100644 index 000000000..1989a9b3c --- /dev/null +++ b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Program.cs @@ -0,0 +1,30 @@ +using Elsa.Activities.RabbitMq; +using Elsa.Samples.RabbitMqWorker.Workflows; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; + +namespace Elsa.Samples.RabbitMqWorker +{ + public class Program + { + public static void Main(string[] args) + { + CreateHostBuilder(args).Build().Run(); + } + + public static IHostBuilder CreateHostBuilder(string[] args) => + Host.CreateDefaultBuilder(args) + .ConfigureServices((hostContext, services) => + { + services + .AddElsa(options => options + .AddConsoleActivities() + .AddQuartzTemporalActivities() + .AddRabbitMqActivities() + .AddWorkflow() + .AddWorkflow() + .StartWorkflow() + ); + }); + } +} \ No newline at end of file diff --git a/src/samples/worker/Elsa.Samples.RabbitMqWorker/Properties/launchSettings.json b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Properties/launchSettings.json new file mode 100644 index 000000000..ffa540ec8 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Properties/launchSettings.json @@ -0,0 +1,11 @@ +{ + "profiles": { + "Elsa.Samples.RabbitMqWorker": { + "commandName": "Project", + "dotnetRunMessages": "true", + "environmentVariables": { + "DOTNET_ENVIRONMENT": "Development" + } + } + } +} diff --git a/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/ConsumerWorkflow.cs b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/ConsumerWorkflow.cs new file mode 100644 index 000000000..7dc24b29b --- /dev/null +++ b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/ConsumerWorkflow.cs @@ -0,0 +1,30 @@ +using Elsa.Activities.Console; +using Elsa.Activities.RabbitMq; +using Elsa.Activities.Temporal; +using Elsa.Builders; +using Microsoft.Extensions.Configuration; +using NodaTime; + +namespace Elsa.Samples.RabbitMqWorker.Workflows +{ + public class ConsumerWorkflow : IWorkflow + { + private readonly string _connectionString; + public ConsumerWorkflow(IConfiguration configuration) + { + _connectionString = configuration.GetConnectionString("RabbitMq"); + } + + public void Build(IWorkflowBuilder builder) + { + builder + .Timer(Duration.FromSeconds(5)) + .MessageReceived(_connectionString, "Podcasts.Weather", null) + .WriteLine(context => + { + var message = context.GetInput(); + return $"Received a weather update saying {message}"; + }); + } + } +} diff --git a/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/ProducerWorkflow.cs b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/ProducerWorkflow.cs new file mode 100644 index 000000000..08ca1eb2f --- /dev/null +++ b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/ProducerWorkflow.cs @@ -0,0 +1,26 @@ +using Elsa.Activities.Console; +using Elsa.Activities.RabbitMq; +using Elsa.Activities.Temporal; +using Elsa.Builders; +using Microsoft.Extensions.Configuration; +using NodaTime; + +namespace Elsa.Samples.RabbitMqWorker.Workflows +{ + public class ProducerWorkflow : IWorkflow + { + private readonly string _connectionString; + public ProducerWorkflow(IConfiguration configuration) + { + _connectionString = configuration.GetConnectionString("RabbitMq"); + } + + public void Build(IWorkflowBuilder builder) + { + builder + .Timer(Duration.FromSeconds(5)) + .WriteLine("Sending a weather update with the \"Podcasts.Weather\" topic.") + .SendTopicMessage(_connectionString, "Podcasts.Weather", null, "Cloudy with a chance of meatballs"); + } + } +} \ No newline at end of file diff --git a/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/SendAndReceiveWorkflow.cs b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/SendAndReceiveWorkflow.cs new file mode 100644 index 000000000..c02a721a7 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/SendAndReceiveWorkflow.cs @@ -0,0 +1,29 @@ +using Elsa.Activities.Console; +using Elsa.Activities.RabbitMq; +using Elsa.Builders; +using Microsoft.Extensions.Configuration; + +namespace Elsa.Samples.RabbitMqWorker.Workflows +{ + public class SendAndReceiveWorkflow : IWorkflow + { + private readonly string _connectionString; + public SendAndReceiveWorkflow(IConfiguration configuration) + { + _connectionString = configuration.GetConnectionString("RabbitMq"); + } + + public void Build(IWorkflowBuilder builder) => builder + .WriteLine(ctx => + { + var correlationId = Guid.NewGuid().ToString("n"); + ctx.WorkflowInstance.CorrelationId = correlationId; + + return $"Start! - correlationId: {correlationId}"; + }) + .SendTopicMessage(_connectionString, "Greetings", null, "Greetings from RabbitMQ") + .MessageReceived(_connectionString, "Greetings", null) + .WriteLine(ctx => "End: " + (string)ctx.Input); + + } +} \ No newline at end of file diff --git a/src/samples/worker/Elsa.Samples.RabbitMqWorker/appsettings.Development.json b/src/samples/worker/Elsa.Samples.RabbitMqWorker/appsettings.Development.json new file mode 100644 index 000000000..8983e0fc1 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.RabbitMqWorker/appsettings.Development.json @@ -0,0 +1,9 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Information", + "Microsoft": "Warning", + "Microsoft.Hosting.Lifetime": "Information" + } + } +} diff --git a/src/samples/worker/Elsa.Samples.RabbitMqWorker/appsettings.json b/src/samples/worker/Elsa.Samples.RabbitMqWorker/appsettings.json new file mode 100644 index 000000000..344524b6e --- /dev/null +++ b/src/samples/worker/Elsa.Samples.RabbitMqWorker/appsettings.json @@ -0,0 +1,12 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Information", + "Microsoft": "Warning", + "Microsoft.Hosting.Lifetime": "Information" + } + }, + "ConnectionStrings": { + "RabbitMq": "" + } +} diff --git a/src/server/Elsa.Server.Api/Extensions/ServiceCollectionExtensions.cs b/src/server/Elsa.Server.Api/Extensions/ServiceCollectionExtensions.cs index 8491e21f1..0aaa08bd2 100644 --- a/src/server/Elsa.Server.Api/Extensions/ServiceCollectionExtensions.cs +++ b/src/server/Elsa.Server.Api/Extensions/ServiceCollectionExtensions.cs @@ -1,4 +1,3 @@ -using System; using Elsa; using Elsa.Models; using Elsa.Server.Api; @@ -10,6 +9,7 @@ using Microsoft.OpenApi.Any; using Microsoft.OpenApi.Models; using Swashbuckle.AspNetCore.Filters; using Swashbuckle.AspNetCore.SwaggerGen; +using System; // ReSharper disable once CheckNamespace namespace Microsoft.Extensions.DependencyInjection @@ -53,6 +53,7 @@ namespace Microsoft.Extensions.DependencyInjection .AddSingleton() .AddAutoMapperProfile() .AddSignalR(); + return services; }