SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Closes #1660: Added RabbitMQ activities (#2547)

Browse source 
* Added RabbitMQ activities

* Added RabbitMQ activities

* RabbitMQ: cleaned up & refactored, added sample project

* reverted change that re-written other update during conflict resolve

* removed queue listener toggle for RabbitMQ receiver, refactored workers
This commit is contained in:
Konstantins Vedenins 2021-12-16 15:33:03 +02:00 committed by GitHub
parent d249d2b3be
commit f8c174df14
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
62 changed files with 1595 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
Elsa.sln
View file

@ -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}

12
src/activities/Elsa.Activities.RabbitMq/Activities/IRabbitMqActivity.cs Normal file
View file

@ -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<string, string> Headers { get; set; }
}
}

87
src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceived.cs Normal file
View file

@ -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<string, string> Headers { get; set; } = new Dictionary<string, string>();
[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<IActivityExecutionResult> OnExecuteAsync(ActivityExecutionContext context) => context.WorkflowExecutionContext.IsFirstPass ? ExecuteInternalAsync(context) : await SuspendInternalAsync();
protected override IActivityExecutionResult OnResume(ActivityExecutionContext context) => ExecuteInternalAsync(context);
private async ValueTask<IActivityExecutionResult> 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();
}
}
}

16
src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceivedBuilderExtensions.cs Normal file
View file

@ -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<ISetupActivity<RabbitMqMessageReceived>> 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<string, string> headers, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) =>
builder.MessageReceived(setup => setup.WithConnectionString(connectionString).WithRoutingKey(routingKey).WithHeaders(headers), lineNumber, sourceFile);
}
}

29
src/activities/Elsa.Activities.RabbitMq/Activities/RabbitMqMessageReceived/RabbitMqMessageReceivedExtensions.cs Normal file
View file

@ -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<RabbitMqMessageReceived> WithConnectionString(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.ConnectionString, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithConnectionString(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.ConnectionString, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithConnectionString(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<string> value) => messageReceived.Set(x => x.ConnectionString, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithConnectionString(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.ConnectionString, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithConnectionString(this ISetupActivity<RabbitMqMessageReceived> messageReceived, string value) => messageReceived.Set(x => x.ConnectionString, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithRoutingKey(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.RoutingKey, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithRoutingKey(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.RoutingKey, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithRoutingKey(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<string> value) => messageReceived.Set(x => x.RoutingKey, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithRoutingKey(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.RoutingKey, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithRoutingKey(this ISetupActivity<RabbitMqMessageReceived> messageReceived, string value) => messageReceived.Set(x => x.RoutingKey, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithHeaders(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<Dictionary<string, string>>> value) => messageReceived.Set(x => x.Headers, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithHeaders(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ValueTask<Dictionary<string, string>>> value) => messageReceived.Set(x => x.Headers, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithHeaders(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<Dictionary<string, string>> value) => messageReceived.Set(x => x.Headers, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithHeaders(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Func<ActivityExecutionContext, Dictionary<string, string>> value) => messageReceived.Set(x => x.Headers, value!);
public static ISetupActivity<RabbitMqMessageReceived> WithHeaders(this ISetupActivity<RabbitMqMessageReceived> messageReceived, Dictionary<string, string> value) => messageReceived.Set(x => x.Headers, value!);
}
}

68
src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessage.cs Normal file
View file

@ -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<string, string> Headers { get; set; } = new Dictionary<string, string>();
[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<IActivityExecutionResult> OnExecuteAsync(ActivityExecutionContext context)
{
var config = new RabbitMqBusConfiguration(ConnectionString, RoutingKey, Headers);
var client = await _messageSenderClientFactory.GetSenderAsync(config);
await client.PublishMessage(Message);
return Done();
}
}
}

27
src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessageBuilderExtensions.cs Normal file
View file

@ -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<ISetupActivity<SendRabbitMqMessage>> 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<string, string> headers, Func<ActivityExecutionContext, ValueTask<string>> 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<string, string> headers, Func<ActivityExecutionContext, string> 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<string, string> headers, Func<string> 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<string, string> 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);
}
}

35
src/activities/Elsa.Activities.RabbitMq/Activities/SendRabbitMqMessage/SendRabbitMqMessageExtensions.cs Normal file
View file

@ -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<SendRabbitMqMessage> WithConnectionString(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ActivityExecutionContext, ValueTask<string>> value) => sendMessage.Set(x => x.ConnectionString, value!);
public static ISetupActivity<SendRabbitMqMessage> WithConnectionString(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ValueTask<string>> value) => sendMessage.Set(x => x.ConnectionString, value!);
public static ISetupActivity<SendRabbitMqMessage> WithConnectionString(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<string> value) => sendMessage.Set(x => x.ConnectionString, value!);
public static ISetupActivity<SendRabbitMqMessage> WithConnectionString(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ActivityExecutionContext, string> value) => sendMessage.Set(x => x.ConnectionString, value!);
public static ISetupActivity<SendRabbitMqMessage> WithConnectionString(this ISetupActivity<SendRabbitMqMessage> sendMessage, string value) => sendMessage.Set(x => x.ConnectionString, value!);
public static ISetupActivity<SendRabbitMqMessage> WithTopic(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ActivityExecutionContext, ValueTask<string>> value) => sendMessage.Set(x => x.RoutingKey, value!);
public static ISetupActivity<SendRabbitMqMessage> WithTopic(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ValueTask<string>> value) => sendMessage.Set(x => x.RoutingKey, value!);
public static ISetupActivity<SendRabbitMqMessage> WithTopic(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<string> value) => sendMessage.Set(x => x.RoutingKey, value!);
public static ISetupActivity<SendRabbitMqMessage> WithTopic(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ActivityExecutionContext, string> value) => sendMessage.Set(x => x.RoutingKey, value!);
public static ISetupActivity<SendRabbitMqMessage> WithTopic(this ISetupActivity<SendRabbitMqMessage> sendMessage, string value) => sendMessage.Set(x => x.RoutingKey, value!);
public static ISetupActivity<SendRabbitMqMessage> WithHeaders(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ActivityExecutionContext, ValueTask<Dictionary<string, string>>> value) => sendMessage.Set(x => x.Headers, value!);
public static ISetupActivity<SendRabbitMqMessage> WithHeaders(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ValueTask<Dictionary<string, string>>> value) => sendMessage.Set(x => x.Headers, value!);
public static ISetupActivity<SendRabbitMqMessage> WithHeaders(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<Dictionary<string, string>> value) => sendMessage.Set(x => x.Headers, value!);
public static ISetupActivity<SendRabbitMqMessage> WithHeaders(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ActivityExecutionContext, Dictionary<string, string>> value) => sendMessage.Set(x => x.Headers, value!);
public static ISetupActivity<SendRabbitMqMessage> WithHeaders(this ISetupActivity<SendRabbitMqMessage> sendMessage, Dictionary<string, string> value) => sendMessage.Set(x => x.Headers, value!);
public static ISetupActivity<SendRabbitMqMessage> WithMessage(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ActivityExecutionContext, ValueTask<string>> value) => sendMessage.Set(x => x.Message, value!);
public static ISetupActivity<SendRabbitMqMessage> WithMessage(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ValueTask<string>> value) => sendMessage.Set(x => x.Message, value!);
public static ISetupActivity<SendRabbitMqMessage> WithMessage(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<string> value) => sendMessage.Set(x => x.Message, value!);
public static ISetupActivity<SendRabbitMqMessage> WithMessage(this ISetupActivity<SendRabbitMqMessage> sendMessage, Func<ActivityExecutionContext, string> value) => sendMessage.Set(x => x.Message, value!);
public static ISetupActivity<SendRabbitMqMessage> WithMessage(this ISetupActivity<SendRabbitMqMessage> sendMessage, string value) => sendMessage.Set(x => x.Message, value!);
}
}

40
src/activities/Elsa.Activities.RabbitMq/Bookmarks/MessageReceivedBookmark.cs Normal file
View file

@ -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<string, string> headers)
{
RoutingKey = routingKey;
ConnectionString = connectionString;
Headers = headers ?? new Dictionary<string, string>();
}
public string RoutingKey { get; set; } = default!;
public string ConnectionString { get; set; } = default!;
public Dictionary<string, string> Headers { get; set; } = default!;
}
public class QueueMessageReceivedBookmarkProvider : BookmarkProvider<MessageReceivedBookmark, RabbitMqMessageReceived>
{
public override async ValueTask<IEnumerable<BookmarkResult>> GetBookmarksAsync(BookmarkProviderContext<RabbitMqMessageReceived> 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))!
})
};
}
}

26
src/activities/Elsa.Activities.RabbitMq/Configuration/RabbitMqBusConfiguration.cs Normal file
View file

@ -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<string, string> Headers { get; }
public RabbitMqBusConfiguration(string connectionString, string routingKey, Dictionary<string, string> headers)
{
ConnectionString = connectionString;
RoutingKey = routingKey;
Headers = headers ?? new Dictionary<string, string>();
}
public override int GetHashCode()
{
var headersString = string.Concat(Headers.Select((x, y) => string.Concat(x, y)));
return System.HashCode.Combine(ConnectionString, RoutingKey, headersString);
}
}
}

15
src/activities/Elsa.Activities.RabbitMq/Consumers/RestartRabbitMqBusConsumer.cs Normal file
View file

@ -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<WorkflowDefinitionPublished>, IHandleMessages<WorkflowDefinitionRetracted>
{
private readonly IRabbitMqQueueStarter _rabbitMqQueueStarter;
public RestartRabbitMqBusConsumer(IRabbitMqQueueStarter rabbitMqQueueStarter) => _rabbitMqQueueStarter = rabbitMqQueueStarter;
public Task Handle(WorkflowDefinitionPublished message) => _rabbitMqQueueStarter.CreateWorkersAsync();
public Task Handle(WorkflowDefinitionRetracted message) => _rabbitMqQueueStarter.CreateWorkersAsync();
}
}

18
src/activities/Elsa.Activities.RabbitMq/Elsa.Activities.RabbitMq.csproj Normal file
View file

@ -0,0 +1,18 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFramework>net5.0</TargetFramework>
<RootNamespace>Elsa.Activities.RabbitMq</RootNamespace>
<LangVersion>latest</LangVersion>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />
<ProjectReference Include="..\..\modules\Elsa.WorkflowTesting\Elsa.WorkflowTesting.csproj" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="Rebus.RabbitMq" Version="7.3.0" />
</ItemGroup>
</Project>

39
src/activities/Elsa.Activities.RabbitMq/Extensions/ServiceCollectionExtensions.cs Normal file
View file

@ -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<BusClientFactory>()
.AddSingleton<IMessageReceiverClientFactory>(sp => sp.GetRequiredService<BusClientFactory>())
.AddSingleton<IMessageSenderClientFactory>(sp => sp.GetRequiredService<BusClientFactory>())
.AddSingleton<IRabbitMqQueueStarter, RabbitMqQueueStarter>()
.AddSingleton<Scoped<IWorkflowLaunchpad>>()
.AddSingleton<IRabbitMqTestQueueManager, RabbitMqTestQueueManager>()
.AddNotificationHandlersFrom<ConfigureRabbitMqActivitiesForTestHandler>()
.AddStartupTask<StartRabbitMqQueues>()
.AddBookmarkProvider<QueueMessageReceivedBookmarkProvider>();
options.AddPubSubConsumer<RestartRabbitMqBusConsumer, WorkflowDefinitionPublished>("WorkflowDefinitionEvents");
options.AddPubSubConsumer<RestartRabbitMqBusConsumer, WorkflowDefinitionRetracted>("WorkflowDefinitionEvents");
options
.AddActivity<RabbitMqMessageReceived>()
.AddActivity<SendRabbitMqMessage>();
return options;
}
}
}

91
src/activities/Elsa.Activities.RabbitMq/Services/BusClientFactory.cs Normal file
View file

@ -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<int, IClient> _receivers = new Dictionary<int, IClient>();
private readonly IDictionary<int, IClient> _senders = new Dictionary<int, IClient>();
private readonly SemaphoreSlim _semaphore = new(1);
public async Task<IClient> 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<IClient> 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();
}
}

86
src/activities/Elsa.Activities.RabbitMq/Services/Client.cs Normal file
View file

@ -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<TransportMessage, CancellationToken, Task> 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<string, string> messageHeaders) => Configuration.Headers.All(x => messageHeaders.ContainsKey(x.Key) && messageHeaders[x.Key] == x.Value);
}
}

18
src/activities/Elsa.Activities.RabbitMq/Services/IClient.cs Normal file
View file

@ -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<TransportMessage, CancellationToken, Task> handler);
Task PublishMessage(string message);
void StartClient();
void StopClient();
void Dispose();
}
}

12
src/activities/Elsa.Activities.RabbitMq/Services/IMessageReceiverClientFactory.cs Normal file
View file

@ -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<IClient> GetReceiverAsync(RabbitMqBusConfiguration config, CancellationToken cancellationToken = default);
Task DisposeReceiverAsync(IClient receiverClient, CancellationToken cancellationToken = default);
}
}

12
src/activities/Elsa.Activities.RabbitMq/Services/IMessageSenderClientFactory.cs Normal file
View file

@ -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<IClient> GetSenderAsync(RabbitMqBusConfiguration config, CancellationToken cancellationToken = default);
Task DisposeSenderAsync(IClient senderClient, CancellationToken cancellationToken = default);
}
}

16
src/activities/Elsa.Activities.RabbitMq/Services/IRabbitMqQueueStarter.cs Normal file
View file

@ -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<Worker> CreateWorkerAsync(RabbitMqBusConfiguration config, CancellationToken cancellationToken = default);
IAsyncEnumerable<RabbitMqBusConfiguration> GetConfigurationsAsync<T>(Func<IWorkflowBlueprint, bool>? predicate, CancellationToken cancellationToken) where T : IRabbitMqActivity;
}
}

115
src/activities/Elsa.Activities.RabbitMq/Services/RabbitMqQueueStarter.cs Normal file
View file

@ -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<Worker> _workers;
private readonly IMessageReceiverClientFactory _messageReceiverClientFactory;
private readonly ILogger _logger;
public RabbitMqQueueStarter(
IConfiguration configuration,
IMessageReceiverClientFactory messageReceiverClientFactory,
IServiceScopeFactory scopeFactory,
IServiceProvider serviceProvider,
ILogger<RabbitMqQueueStarter> logger)
{
_messageReceiverClientFactory = messageReceiverClientFactory;
_scopeFactory = scopeFactory;
_serviceProvider = serviceProvider;
_logger = logger;
_workers = new List<Worker>();
}
public async Task CreateWorkersAsync(CancellationToken cancellationToken = default)
{
await _semaphore.WaitAsync(cancellationToken);
try
{
await DisposeExistingWorkersAsync();
var receiverConfigs = (await GetConfigurationsAsync<RabbitMqMessageReceived>(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<Worker> CreateWorkerAsync(RabbitMqBusConfiguration config, CancellationToken cancellationToken = default)
{
var receiver = await _messageReceiverClientFactory.GetReceiverAsync(config, cancellationToken);
return ActivatorUtilities.CreateInstance<Worker>(_serviceProvider, (Func<IClient, Task>)DisposeWorkerAsync, receiver);
}
private async Task DisposeWorkerAsync(IClient messageReceiver) => await _messageReceiverClientFactory.DisposeReceiverAsync(messageReceiver);
public async IAsyncEnumerable<RabbitMqBusConfiguration> GetConfigurationsAsync<T>(Func<IWorkflowBlueprint, bool>? predicate, [EnumeratorCancellation] CancellationToken cancellationToken) where T : IRabbitMqActivity
{
using var scope = _scopeFactory.CreateScope();
var workflowRegistry = scope.ServiceProvider.GetRequiredService<IWorkflowRegistry>();
var workflowBlueprintReflector = scope.ServiceProvider.GetRequiredService<IWorkflowBlueprintReflector>();
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<T>())
{
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);
}
}
}
}

60
src/activities/Elsa.Activities.RabbitMq/Services/Worker.cs Normal file
View file

@ -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<IClient, Task> _disposeReceiverAction;
private readonly Scoped<IWorkflowLaunchpad> _workflowLaunchpad;
private string ActivityType => nameof(RabbitMqMessageReceived);
private TimeSpan _delay = TimeSpan.FromMilliseconds(200);
public Worker(
Scoped<IWorkflowLaunchpad> workflowLaunchpad,
IClient client,
Func<IClient, Task> disposeReceiverAction,
ILogger<Worker> 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));
}
}
}

16
src/activities/Elsa.Activities.RabbitMq/Startup.cs Normal file
View file

@ -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();
}
}
}

19
src/activities/Elsa.Activities.RabbitMq/StartupTasks/StartRabbitMqQueues.cs Normal file
View file

@ -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);
}
}

63
src/activities/Elsa.Activities.RabbitMq/Testing/Handlers/ConfigureRabbitMqActivitiesForTestHandler.cs Normal file
View file

@ -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<WorkflowExecuting>, INotificationHandler<WorkflowFaulted>, INotificationHandler<WorkflowCompleted>, INotificationHandler<WorkflowTestExecutionStopped>
{
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;
}
}
}

11
src/activities/Elsa.Activities.RabbitMq/Testing/IRabbitMqTestQueueManager.cs Normal file
View file

@ -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);
}
}

72
src/activities/Elsa.Activities.RabbitMq/Testing/RabbitMqTestQueueManager.cs Normal file
View file

@ -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<string, ICollection<Worker>> _workers;
private readonly IRabbitMqQueueStarter _rabbitMqQueueStarter;
private readonly ILogger _logger;
public RabbitMqTestQueueManager(
IRabbitMqQueueStarter rabbitMqQueueStarter,
ILogger<RabbitMqTestQueueManager> logger)
{
_rabbitMqQueueStarter = rabbitMqQueueStarter;
_logger = logger;
_workers = new Dictionary<string, ICollection<Worker>>();
}
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<Worker>();
var workerConfigs = (await _rabbitMqQueueStarter.GetConfigurationsAsync<RabbitMqMessageReceived>(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();
}
}
}

1
src/core/Elsa.Abstractions/Design/ActivityInputUIHints.cs
View file

@ -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";
/// <summary>
/// An editor that allows the user to write a blob of JSON.

1
src/core/Elsa.Abstractions/Design/PropertyCategories.cs
View file

@ -4,5 +4,6 @@
{
public const string Advanced = "Advanced";
public const string Hosting = "Hosting";
public const string Configuration = "Configuration";
}
}

21
src/designer/elsa-workflows-studio/src/components.d.ts vendored
View file

@ -64,6 +64,12 @@ export namespace Components {
"selectedActivityIds": Array<string>;
"showActivityEditor": (activity: ActivityModel, animate: boolean) => Promise<void>;
}
interface ElsaDictionaryProperty {
"activityModel": ActivityModel;
"propertyDescriptor": ActivityPropertyDescriptor;
"propertyModel": ActivityDefinitionProperty;
"serverUrl": string;
}
interface ElsaDropdownButton {
"icon"?: any;
"items": Array<DropdownButtonItem>;
@ -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<WorkflowModel>) => void;
"selectedActivityIds"?: Array<string>;
}
interface ElsaDictionaryProperty {
"activityModel"?: ActivityModel;
"propertyDescriptor"?: ActivityPropertyDescriptor;
"propertyModel"?: ActivityDefinitionProperty;
"serverUrl"?: string;
}
interface ElsaDropdownButton {
"icon"?: any;
"items"?: Array<DropdownButtonItem>;
@ -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<HTMLElsaControlElement>;
"elsa-copy-button": LocalJSX.ElsaCopyButton & JSXBase.HTMLAttributes<HTMLElsaCopyButtonElement>;
"elsa-designer-tree": LocalJSX.ElsaDesignerTree & JSXBase.HTMLAttributes<HTMLElsaDesignerTreeElement>;
"elsa-dictionary-property": LocalJSX.ElsaDictionaryProperty & JSXBase.HTMLAttributes<HTMLElsaDictionaryPropertyElement>;
"elsa-dropdown-button": LocalJSX.ElsaDropdownButton & JSXBase.HTMLAttributes<HTMLElsaDropdownButtonElement>;
"elsa-dropdown-property": LocalJSX.ElsaDropdownProperty & JSXBase.HTMLAttributes<HTMLElsaDropdownPropertyElement>;
"elsa-expression-editor": LocalJSX.ElsaExpressionEditor & JSXBase.HTMLAttributes<HTMLElsaExpressionEditorElement>;

2
src/designer/elsa-workflows-studio/src/components/editors/elsa-property-editor/readme.md
View file

@ -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

124
src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-dictionary-property/elsa-dictionary-property.tsx Normal file
View file

@ -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 (
<elsa-property-editor propertyDescriptor={propertyDescriptor}
propertyModel={propertyModel}
activityModel={this.activityModel}
onDefaultSyntaxValueChanged={e => 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 (
<div class="elsa-flex elsa-flex-row elsa-justify-between elsa-mb-2">
<input id={keyInputId} type="text" value={key} onChange={(e) => 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"/>
<input id={valueInputId} type="text" value={value} onChange={(e) => 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"/>
<div class="elsa-flex elsa-flex-row elsa-justify-between elsa-w-24">
<button type="button" onClick={() => this.onRowDeleted(index)}>
{iconProvider.getIcon(IconName.TrashBinOutline, {color: IconColor.Gray, hoverColor: IconColor.Red})}
</button>
{isLast && <button type="button" onClick={this.onRowAdded}>
{iconProvider.getIcon(IconName.Plus, {color: IconColor.Gray, hoverColor: IconColor.Green})}
</button>}
</div>
</div>
);
})}
</elsa-property-editor>
);
}
}
Tunnel.injectProps(ElsaDictionaryProperty, ['serverUrl']);

38
src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-dictionary-property/readme.md Normal file
View file

@ -0,0 +1,38 @@
# elsa-dictionary-property
<!-- Auto Generated Below -->
## 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/)*

16
src/designer/elsa-workflows-studio/src/components/editors/properties/elsa-switch-cases-property/elsa-switch-cases-property.tsx
View file

@ -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 {
<td class="elsa-pt-1 elsa-pr-2 elsa-text-right">
<button type="button" onClick={() => this.onDeleteCaseClick(switchCase)}
class="elsa-h-5 elsa-w-5 elsa-mx-auto elsa-outline-none focus:elsa-outline-none">
<svg class="elsa-h-5 elsa-w-5 elsa-text-gray-500" viewBox="0 0 24 24" fill="none" stroke="currentColor"
stroke-width="2" stroke-linecap="round" stroke-linejoin="round">
<polyline points="3 6 5 6 21 6"/>
<path d="M19 6v14a2 2 0 0 1-2 2H7a2 2 0 0 1-2-2V6m3 0V4a2 2 0 0 1 2-2h4a2 2 0 0 1 2 2v2"/>
<line x1="10" y1="11" x2="10" y2="17"/>
<line x1="14" y1="11" x2="14" y2="17"/>
</svg>
{ iconProvider.getIcon(IconName.TrashBinOutline) }
</button>
</td>
</tr>
@ -182,12 +177,7 @@ export class ElsaSwitchCasesProperty {
</table>
<button type="button" onClick={() => this.onAddCaseClick()}
class="elsa-inline-flex elsa-items-center elsa-px-4 elsa-py-2 elsa-border elsa-border-transparent elsa-shadow-sm elsa-text-sm elsa-font-medium elsa-rounded-md elsa-text-white elsa-bg-blue-600 hover:elsa-bg-blue-700 focus:elsa-outline-none focus:elsa-ring-2 focus:elsa-ring-offset-2 focus:elsa-ring-blue-500 elsa-mt-2">
<svg class="-elsa-ml-1 elsa-mr-2 elsa-h-5 elsa-w-5" width="24" height="24" viewBox="0 0 24 24"
stroke-width="2" stroke="currentColor" fill="none" stroke-linecap="round" stroke-linejoin="round">
<path stroke="none" d="M0 0h24v24H0z"/>
<line x1="12" y1="5" x2="12" y2="19"/>
<line x1="5" y1="12" x2="19" y2="12"/>
</svg>
{ iconProvider.getIcon(IconName.Plus) }
Add Case
</button>
</elsa-multi-expression-editor>

1
src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-definition-editor-screen/elsa-workflow-definition-editor-screen.tsx
View file

@ -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);

5
src/designer/elsa-workflows-studio/src/components/screens/workflow-definition-editor/elsa-workflow-test-panel/elsa-workflow-test-panel.tsx
View file

@ -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;

15
src/designer/elsa-workflows-studio/src/drivers/dictionary-driver.tsx Normal file
View file

@ -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 <elsa-dictionary-property activityModel={activity} propertyDescriptor={property} propertyModel={prop} />;
}
update(activity: ActivityModel, property: ActivityPropertyDescriptor, form: FormData) {
}
}

1
src/designer/elsa-workflows-studio/src/drivers/index.ts
View file

@ -9,3 +9,4 @@ export * from './null-property-driver';
export * from './radio-list-driver';
export * from './single-line-driver';
export * from './switch-case-builder-driver';
export * from './dictionary-driver';

8
src/designer/elsa-workflows-studio/src/globals/tailwind.css
View file

@ -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%;
}

2
src/designer/elsa-workflows-studio/src/index.html
View file

@ -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;
}
});

3
src/designer/elsa-workflows-studio/src/plugins/default-drivers-plugin.ts
View file

@ -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<T extends PropertyDisplayDriver>(controlType: string, c: (elsaStudio: ElsaStudio) => T) {

4
src/designer/elsa-workflows-studio/src/services/activity-icon-provider.tsx
View file

@ -60,7 +60,9 @@ export class ActivityIconProvider {
'WriteHttpResponse': () => <WriteHttpResponseIcon/>,
'Redirect': () => <RedirectIcon/>,
'SendEmail': () => <SendEmailIcon/>,
'Webhook': () => <WebhookIcon/>
'Webhook': () => <WebhookIcon/>,
'RabbitMqMessageReceived': () => <SignalReceivedIcon/>,
'SendRabbitMqMessage': () => <SendSignalIcon/>
};
register(activityType: string, icon: string) {

8
src/designer/elsa-workflows-studio/src/services/elsa-client.ts
View file

@ -124,6 +124,9 @@ export const createElsaClient = async function (serverUrl: string): Promise<Elsa
},
restartFromActivity: async (request) => {
await httpClient.post<void>(`v1/workflow-test/restartFromActivity`, request);
},
stop: async (request) => {
await httpClient.post<void>(`v1/workflow-test/stop`, request);
}
},
workflowRegistryApi: {
@ -289,6 +292,7 @@ export interface WorkflowTestApi {
execute(request: WorkflowTestExecuteRequest): Promise<void>;
restartFromActivity(request: WorkflowTestRestartFromActivityRequest): Promise<void>;
stop(request: WorkflowTestStopRequest): Promise<void>;
}
export interface WorkflowRegistryApi {
@ -388,6 +392,10 @@ export interface WorkflowTestRestartFromActivityRequest {
signalRConnectionId: string
}
export interface WorkflowTestStopRequest {
workflowInstanceId: string
}
export interface ExportWorkflowResponse {
fileName: string;
data: Blob;

57
src/designer/elsa-workflows-studio/src/services/icon-provider.tsx Normal file
View file

@ -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) =>
<svg
class={`-elsa-ml-1 elsa-mr-2 elsa-h-5 elsa-w-5 ${options?.color ? `elsa-text-${options.color}-500` : ''} ${options?.hoverColor ? `hover:elsa-text-${options.hoverColor}-500` : ''}`}
width="24" height="24" viewBox="0 0 24 24"
stroke-width="2" stroke="currentColor" fill="transparent" stroke-linecap="round"
stroke-linejoin="round">
<path stroke="none" d="M0 0h24v24H0z"/>
<line x1="12" y1="5" x2="12" y2="19"/>
<line x1="5" y1="12" x2="19" y2="12"/>
</svg>,
'trash-bin-outline': (options?: IconProviderOptions) =>
<svg
class={`elsa-h-5 elsa-w-5 ${options?.color ? `elsa-text-${options.color}-500` : ''} ${options?.hoverColor ? `hover:elsa-text-${options.hoverColor}-500` : ''}`}
width="24" height="24" viewBox="0 0 24 24"
stroke-width="2" stroke="currentColor" fill="transparent" stroke-linecap="round"
stroke-linejoin="round">
<polyline points="3 6 5 6 21 6"/>
<path d="M19 6v14a2 2 0 0 1-2 2H7a2 2 0 0 1-2-2V6m3 0V4a2 2 0 0 1 2-2h4a2 2 0 0 1 2 2v2"/>
<line x1="10" y1="11" x2="10" y2="17"/>
<line x1="14" y1="11" x2="14" y2="17"/>
</svg>
};
getIcon(name: IconName, options?: IconProviderOptions): any {
const provider = this.map[name];
if (!provider)
return undefined;
return provider(options);
}
}
export const iconProvider = new IconProvider();

44
src/modules/Elsa.WorkflowTesting.Api/Endpoints/Stop.cs Normal file
View file

@ -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<IActionResult> Handle([FromBody] WorkflowTestStopRequest request, CancellationToken cancellationToken = default)
{
await _mediator.Publish(new WorkflowTestExecutionStopped(request.WorkflowInstanceId), cancellationToken);
return Ok();
}
}
}

8
src/modules/Elsa.WorkflowTesting.Api/Handlers/ActivityExecutionResultExecutedHandler.cs
View file

@ -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
{

6
src/modules/Elsa.WorkflowTesting.Api/Handlers/ConfigureWorkflowContextForTestHandler.cs
View file

@ -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<WorkflowExecuting>
{

7
src/modules/Elsa.WorkflowTesting.Api/Models/WorkflowTestStopRequest.cs Normal file
View file

@ -0,0 +1,7 @@
namespace Elsa.Testing.Api.Models
{
public class WorkflowTestStopRequest
{
public string WorkflowInstanceId { get; init; }
}
}

4
src/modules/Elsa.WorkflowTesting/Elsa.WorkflowTesting.csproj
View file

@ -6,6 +6,10 @@
<LangVersion>default</LangVersion>
</PropertyGroup>
<ItemGroup>
<FrameworkReference Include="Microsoft.AspNetCore.App" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\core\Elsa.Abstractions\Elsa.Abstractions.csproj" />
<ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />

14
src/modules/Elsa.WorkflowTesting/Events/WorkflowTestExecutionStopped.cs Normal file
View file

@ -0,0 +1,14 @@
using MediatR;
namespace Elsa.Testing.Events
{
public class WorkflowTestExecutionStopped : INotification
{
public string WorkflowInstanceId { get; }
public WorkflowTestExecutionStopped(string workflowInstanceId)
{
WorkflowInstanceId = workflowInstanceId;
}
}
}

2
src/modules/Elsa.WorkflowTesting/Services/IWorkflowTestService.cs
View file

@ -1,5 +1,5 @@
using System.Threading.Tasks;
using Elsa.WorkflowTesting.Messages;
using System.Threading.Tasks;
namespace Elsa.WorkflowTesting.Services
{

1
src/samples/server/Elsa.Samples.Server.Host/Elsa.Samples.Server.Host.csproj
View file

@ -18,6 +18,7 @@
<ProjectReference Include="..\..\..\activities\Elsa.Activities.File\Elsa.Activities.File.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Http\Elsa.Activities.Http.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.MassTransit\Elsa.Activities.MassTransit.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.RabbitMq\Elsa.Activities.RabbitMq.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Rebus\Elsa.Activities.Rebus.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Telnyx\Elsa.Activities.Telnyx.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Temporal.Hangfire\Elsa.Activities.Temporal.Hangfire.csproj" />

1
src/samples/server/Elsa.Samples.Server.Host/Startup.cs
View file

@ -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),

4
src/samples/server/Elsa.Samples.Server.Host/appsettings.json
View file

@ -50,6 +50,10 @@
"Enabled": true,
"Framework": "EntityFrameworkCore",
"ConnectionStringIdentifier": "Sqlite"
},
"RabbitMq": {
"Enabled": true,
"ConnectionStringIdentifier": "RabbitMq"
}
},
"WorkflowChannels": {

45
src/samples/worker/Elsa.Samples.RabbitMqWorker/Elsa.Samples.RabbitMqWorker.csproj Normal file
View file

@ -0,0 +1,45 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net6.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
</PropertyGroup>
<ItemGroup>
<None Remove="appsettings.Development.json" />
<None Remove="appsettings.json" />
</ItemGroup>
<ItemGroup>
<Content Include="appsettings.Development.json">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
<ExcludeFromSingleFile>true</ExcludeFromSingleFile>
<CopyToPublishDirectory>PreserveNewest</CopyToPublishDirectory>
</Content>
<Content Include="appsettings.json">
<CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
<ExcludeFromSingleFile>true</ExcludeFromSingleFile>
<CopyToPublishDirectory>PreserveNewest</CopyToPublishDirectory>
</Content>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\activities\Elsa.Activities.RabbitMq\Elsa.Activities.RabbitMq.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Temporal.Quartz\Elsa.Activities.Temporal.Quartz.csproj" />
<ProjectReference Include="..\..\..\core\Elsa\Elsa.csproj" />
</ItemGroup>
<ItemGroup>
<None Update="Properties\launchSettings.json">
<ExcludeFromSingleFile>true</ExcludeFromSingleFile>
<CopyToPublishDirectory>Never</CopyToPublishDirectory>
</None>
</ItemGroup>
</Project>

30
src/samples/worker/Elsa.Samples.RabbitMqWorker/Program.cs Normal file
View file

@ -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<ConsumerWorkflow>()
.AddWorkflow<ProducerWorkflow>()
.StartWorkflow<SendAndReceiveWorkflow>()
);
});
}
}

11
src/samples/worker/Elsa.Samples.RabbitMqWorker/Properties/launchSettings.json Normal file
View file

@ -0,0 +1,11 @@
{
"profiles": {
"Elsa.Samples.RabbitMqWorker": {
"commandName": "Project",
"dotnetRunMessages": "true",
"environmentVariables": {
"DOTNET_ENVIRONMENT": "Development"
}
}
}
}

30
src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/ConsumerWorkflow.cs Normal file
View file

@ -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<string>();
return $"Received a weather update saying {message}";
});
}
}
}

26
src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/ProducerWorkflow.cs Normal file
View file

@ -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");
}
}
}

29
src/samples/worker/Elsa.Samples.RabbitMqWorker/Workflows/SendAndReceiveWorkflow.cs Normal file
View file

@ -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);
}
}

9
src/samples/worker/Elsa.Samples.RabbitMqWorker/appsettings.Development.json Normal file
View file

@ -0,0 +1,9 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft": "Warning",
"Microsoft.Hosting.Lifetime": "Information"
}
}
}

12
src/samples/worker/Elsa.Samples.RabbitMqWorker/appsettings.json Normal file
View file

@ -0,0 +1,12 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft": "Warning",
"Microsoft.Hosting.Lifetime": "Information"
}
},
"ConnectionStrings": {
"RabbitMq": ""
}
}

3
src/server/Elsa.Server.Api/Extensions/ServiceCollectionExtensions.cs
View file

@ -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<IEndpointContentSerializerSettingsProvider, EndpointContentSerializerSettingsProvider>()
.AddAutoMapperProfile<AutoMapperProfile>()
.AddSignalR();
return services;
}