diff --git a/Elsa.sln b/Elsa.sln index 760e774dd..27af6f4f3 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -381,6 +381,10 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.RabbitMq", 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 +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Activities.Mqtt", "src\activities\Elsa.Activities.Mqtt\Elsa.Activities.Mqtt.csproj", "{791D50F6-EAF2-4E20-9EDF-3BAF9CB7F796}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Elsa.Samples.MqttWorker", "src\samples\worker\Elsa.Samples.MqttWorker\Elsa.Samples.MqttWorker.csproj", "{8B58C3EA-20C2-491A-ADEC-E41F0EC787FB}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -903,6 +907,14 @@ Global {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 + {791D50F6-EAF2-4E20-9EDF-3BAF9CB7F796}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {791D50F6-EAF2-4E20-9EDF-3BAF9CB7F796}.Debug|Any CPU.Build.0 = Debug|Any CPU + {791D50F6-EAF2-4E20-9EDF-3BAF9CB7F796}.Release|Any CPU.ActiveCfg = Release|Any CPU + {791D50F6-EAF2-4E20-9EDF-3BAF9CB7F796}.Release|Any CPU.Build.0 = Release|Any CPU + {8B58C3EA-20C2-491A-ADEC-E41F0EC787FB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {8B58C3EA-20C2-491A-ADEC-E41F0EC787FB}.Debug|Any CPU.Build.0 = Debug|Any CPU + {8B58C3EA-20C2-491A-ADEC-E41F0EC787FB}.Release|Any CPU.ActiveCfg = Release|Any CPU + {8B58C3EA-20C2-491A-ADEC-E41F0EC787FB}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -1073,6 +1085,8 @@ Global {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} + {791D50F6-EAF2-4E20-9EDF-3BAF9CB7F796} = {B43B546E-23F3-46E8-ACB7-D04F05CDA180} + {8B58C3EA-20C2-491A-ADEC-E41F0EC787FB} = {E42743A0-FBDD-4150-9D53-6000496D9B87} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {8B0975FD-7050-48B0-88C5-48C33378E158} diff --git a/src/activities/Elsa.Activities.Mqtt/Activities/MqttBaseActivity.cs b/src/activities/Elsa.Activities.Mqtt/Activities/MqttBaseActivity.cs new file mode 100644 index 000000000..cd29f76c1 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Activities/MqttBaseActivity.cs @@ -0,0 +1,53 @@ +using Elsa.Attributes; +using Elsa.Design; +using Elsa.Expressions; +using Elsa.Services; +using System.Net.Mqtt; + +namespace Elsa.Activities.Mqtt +{ + public abstract class MqttBaseActivity : Activity + { + [ActivityInput( + Hint = "Topic", + Order = 1, + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string Topic { get; set; } = default!; + + + [ActivityInput( + Hint = "MQTT broker hostname", + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid }, + Order = 1, + Category = PropertyCategories.Configuration)] + public string Host { get; set; } = default!; + + [ActivityInput( + Hint = "MQTT broker port", + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid }, + Order = 2, + Category = PropertyCategories.Configuration)] + public int Port { get; set; } = default!; + + [ActivityInput( + Hint = "Username", + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid }, + Order = 3, + Category = PropertyCategories.Configuration)] + public string Username { get; set; } = default!; + + [ActivityInput( + Hint = "Password", + SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid }, + Order = 4, + Category = PropertyCategories.Configuration)] + public string Password { get; set; } = default!; + + [ActivityInput( + Hint = "Quality Of Service", + UIHint = ActivityInputUIHints.Dropdown, + Order = 5, + Category = PropertyCategories.Configuration)] + public MqttQualityOfService QualityOfService { get; set; } = default; + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceived.cs b/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceived.cs new file mode 100644 index 000000000..9f8d4d001 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceived.cs @@ -0,0 +1,57 @@ +using Elsa.Activities.Mqtt.Options; +using Elsa.Activities.Mqtt.Services; +using Elsa.ActivityResults; +using Elsa.Attributes; +using Elsa.Services.Models; +using System.Net.Mqtt; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt +{ + [Trigger( + Category = "MQTT", + DisplayName = "MQTT Message Received", + Description = "Triggers when MQTT message matching specified topic is received", + Outcomes = new[] { OutcomeNames.Done } + )] + public class MqttMessageReceived : MqttBaseActivity + { + private readonly IMessageReceiverClientFactory _messageReceiver; + + public MqttMessageReceived(IMessageReceiverClientFactory messageReceiver) + { + _messageReceiver = messageReceiver; + } + + [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 IActivityExecutionResult ExecuteInternalAsync(ActivityExecutionContext context) + { + if (context.Input != null) + { + var message = (MqttApplicationMessage)context.Input; + Output = System.Text.Encoding.UTF8.GetString(message.Payload); + } + + context.LogOutputProperty(this, nameof(Output), Output); + + return Done(); + } + + private async ValueTask SuspendInternalAsync() + { + var options = new MqttClientOptions(Topic, Host, Port, Username, Password, QualityOfService); + + var receiver = await _messageReceiver.GetReceiverAsync(options); + + await receiver.SubscribeAsync(Topic); + + return Suspend(); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceivedBuilderExtensions.cs b/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceivedBuilderExtensions.cs new file mode 100644 index 000000000..044388c89 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceivedBuilderExtensions.cs @@ -0,0 +1,16 @@ +using Elsa.Builders; +using System; +using System.Net.Mqtt; +using System.Runtime.CompilerServices; + +namespace Elsa.Activities.Mqtt +{ + public static class MqttMessageReceivedBuilderExtensions + { + 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 topic, string host, int port, string username, string password, MqttQualityOfService qos, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.MessageReceived(setup => setup.WithTopic(topic).WithHost(host).WithPort(port).WithUsername(username).WithPassword(password).WithQualityOfService(qos), lineNumber, sourceFile); + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceivedExtensions.cs b/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceivedExtensions.cs new file mode 100644 index 000000000..514a1e9bf --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceivedExtensions.cs @@ -0,0 +1,47 @@ +using Elsa.Builders; +using Elsa.Services.Models; +using System; +using System.Net.Mqtt; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt +{ + public static class MqttMessageReceivedExtensions + { + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Topic, value!); + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Topic, value!); + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Topic, value!); + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Topic, value!); + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Topic, value!); + + public static ISetupActivity WithHost(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Host, value!); + public static ISetupActivity WithHost(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Host, value!); + public static ISetupActivity WithHost(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Host, value!); + public static ISetupActivity WithHost(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Host, value!); + public static ISetupActivity WithHost(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Host, value!); + + public static ISetupActivity WithPort(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Port, value!); + public static ISetupActivity WithPort(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Port, value!); + public static ISetupActivity WithPort(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Port, value!); + public static ISetupActivity WithPort(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Port, value!); + public static ISetupActivity WithPort(this ISetupActivity messageReceived, int value) => messageReceived.Set(x => x.Port, value!); + + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Username, value!); + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Username, value!); + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Username, value!); + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Username, value!); + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Username, value!); + + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Password, value!); + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Password, value!); + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Password, value!); + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Password, value!); + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Password, value!); + + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.QualityOfService, value!); + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.QualityOfService, value!); + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.QualityOfService, value!); + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.QualityOfService, value!); + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, MqttQualityOfService value) => messageReceived.Set(x => x.QualityOfService, value!); + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessage.cs b/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessage.cs new file mode 100644 index 000000000..748dc2f32 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessage.cs @@ -0,0 +1,48 @@ +using Elsa.Activities.Mqtt.Options; +using Elsa.Activities.Mqtt.Services; +using Elsa.ActivityResults; +using Elsa.Attributes; +using Elsa.Design; +using Elsa.Expressions; +using Elsa.Services.Models; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt +{ + [Trigger( + Category = "MQTT", + DisplayName = "Send MQTT Message", + Description = "Sends MQTT message matching with topic", + Outcomes = new[] { OutcomeNames.Done } + )] + public class SendMqttMessage : MqttBaseActivity + { + private readonly IMessageSenderClientFactory _messageSenderClientFactory; + + public SendMqttMessage(IMessageSenderClientFactory messageSenderClientFactory) + { + _messageSenderClientFactory = messageSenderClientFactory; + } + + [ActivityInput( + Hint = "Message body", + Order = 2, + UIHint = ActivityInputUIHints.MultiLine, + SupportedSyntaxes = new[] { SyntaxNames.Json })] + public string Message { get; set; } = default!; + + [ActivityOutput(Hint = "Received message")] + public object? Output { get; set; } + + protected override async ValueTask OnExecuteAsync(ActivityExecutionContext context) + { + var options = new MqttClientOptions(Topic, Host, Port, Username, Password, QualityOfService); + + var client = await _messageSenderClientFactory.GetSenderAsync(options); + + await client.PublishMessageAsync(Topic, Message); + + return Done(); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessageBuilderExtensions.cs b/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessageBuilderExtensions.cs new file mode 100644 index 000000000..ed6411d20 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessageBuilderExtensions.cs @@ -0,0 +1,27 @@ +using Elsa.Builders; +using Elsa.Services.Models; +using System; +using System.Net.Mqtt; +using System.Runtime.CompilerServices; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt +{ + public static class SendMqttMessageBuilderExtensions + { + public static IActivityBuilder SendMessage(this IBuilder builder, Action> setup, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.Then(setup, null, lineNumber, sourceFile); + + public static IActivityBuilder SendMessage(this IBuilder builder, string topic, string host, int port, string username, string password, MqttQualityOfService qos, Func> message, [CallerLineNumber] int lineNumber = default, + [CallerFilePath] string? sourceFile = default) => builder.SendMessage(setup => setup.WithTopic(topic).WithHost(host).WithPort(port).WithUsername(username).WithPassword(password).WithQualityOfService(qos).WithMessage(message), lineNumber, sourceFile); + + public static IActivityBuilder SendMessage(this IBuilder builder, string topic, string host, int port, string username, string password, MqttQualityOfService qos, Func message, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.SendMessage(setup => setup.WithTopic(topic).WithHost(host).WithPort(port).WithUsername(username).WithPassword(password).WithQualityOfService(qos).WithMessage(message), lineNumber, sourceFile); + + public static IActivityBuilder SendMessage(this IBuilder builder, string topic, string host, int port, string username, string password, MqttQualityOfService qos, Func message, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.SendMessage(setup => setup.WithTopic(topic).WithHost(host).WithPort(port).WithUsername(username).WithPassword(password).WithQualityOfService(qos).WithMessage(message), lineNumber, sourceFile); + + public static IActivityBuilder SendMessage(this IBuilder builder, string topic, string host, int port, string username, string password, MqttQualityOfService qos, string message, [CallerLineNumber] int lineNumber = default, [CallerFilePath] string? sourceFile = default) => + builder.SendMessage(setup => setup.WithTopic(topic).WithHost(host).WithPort(port).WithUsername(username).WithPassword(password).WithQualityOfService(qos).WithMessage(message), lineNumber, sourceFile); + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessageExtensions.cs b/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessageExtensions.cs new file mode 100644 index 000000000..1528f0282 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessageExtensions.cs @@ -0,0 +1,53 @@ +using Elsa.Builders; +using Elsa.Services.Models; +using System; +using System.Net.Mqtt; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt +{ + public static class SendMqttMessageExtensions + { + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Topic, value!); + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Topic, value!); + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Topic, value!); + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Topic, value!); + public static ISetupActivity WithTopic(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Topic, value!); + + public static ISetupActivity WithHost(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Host, value!); + public static ISetupActivity WithHost(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Host, value!); + public static ISetupActivity WithHost(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Host, value!); + public static ISetupActivity WithHost(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Host, value!); + public static ISetupActivity WithHost(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Host, value!); + + public static ISetupActivity WithPort(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Port, value!); + public static ISetupActivity WithPort(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Port, value!); + public static ISetupActivity WithPort(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Port, value!); + public static ISetupActivity WithPort(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Port, value!); + public static ISetupActivity WithPort(this ISetupActivity messageReceived, int value) => messageReceived.Set(x => x.Port, value!); + + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Username, value!); + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Username, value!); + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Username, value!); + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Username, value!); + public static ISetupActivity WithUsername(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Username, value!); + + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Password, value!); + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Password, value!); + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Password, value!); + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Password, value!); + public static ISetupActivity WithPassword(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Password, value!); + + public static ISetupActivity WithMessage(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.Message, value!); + + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.QualityOfService, value!); + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.QualityOfService, value!); + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.QualityOfService, value!); + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.QualityOfService, value!); + public static ISetupActivity WithQualityOfService(this ISetupActivity messageReceived, MqttQualityOfService value) => messageReceived.Set(x => x.QualityOfService, value!); + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Bookmarks/MessageReceivedBookmark.cs b/src/activities/Elsa.Activities.Mqtt/Bookmarks/MessageReceivedBookmark.cs new file mode 100644 index 000000000..5cbfe38a2 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Bookmarks/MessageReceivedBookmark.cs @@ -0,0 +1,45 @@ +using Elsa.Services; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Bookmarks +{ + public class MessageReceivedBookmark : IBookmark + { + public MessageReceivedBookmark() + { + } + + public MessageReceivedBookmark(string topic, string host, int port, string username, string password) + { + Topic = topic; + Host = host; + Port = port; + Username = username; + Password = password; + } + + public string Topic { get; set; } = default!; + public string Host { get; set; } = default!; + public int Port { get; set; } = default!; + public string Username { get; set; } = default!; + public string Password { get; set; } = default!; + } + + public class MessageReceivedBookmarkProvider : BookmarkProvider + { + public override async ValueTask> GetBookmarksAsync(BookmarkProviderContext context, CancellationToken cancellationToken) => + new[] + { + Result(new MessageReceivedBookmark + { + Topic = (await context.ReadActivityPropertyAsync(x => x.Topic, cancellationToken))!, + Host = (await context.ReadActivityPropertyAsync(x => x.Host, cancellationToken))!, + Port = (await context.ReadActivityPropertyAsync(x => x.Port, cancellationToken))!, + Username = (await context.ReadActivityPropertyAsync(x => x.Username, cancellationToken))!, + Password = (await context.ReadActivityPropertyAsync(x => x.Password, cancellationToken))! + }) + }; + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Consumers/RestartMqttTopicsConsumer.cs b/src/activities/Elsa.Activities.Mqtt/Consumers/RestartMqttTopicsConsumer.cs new file mode 100644 index 000000000..92b435429 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Consumers/RestartMqttTopicsConsumer.cs @@ -0,0 +1,16 @@ +using System.Threading.Tasks; +using Elsa.Activities.Mqtt.Services; +using Elsa.Events; +using Rebus.Handlers; + +namespace Elsa.Activities.Mqtt.Consumers +{ + public class RestartMqttTopicsConsumer: IHandleMessages, IHandleMessages, IHandleMessages + { + private readonly IMqttTopicsStarter _mqttTopicsStarter; + public RestartMqttTopicsConsumer(IMqttTopicsStarter mqttTopicsStarter) => _mqttTopicsStarter = mqttTopicsStarter; + public Task Handle(WorkflowDefinitionPublished message) => _mqttTopicsStarter.CreateWorkersAsync(); + public Task Handle(WorkflowDefinitionRetracted message) => _mqttTopicsStarter.CreateWorkersAsync(); + public Task Handle(WorkflowDefinitionDeleted message) => _mqttTopicsStarter.CreateWorkersAsync(); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Elsa.Activities.Mqtt.csproj b/src/activities/Elsa.Activities.Mqtt/Elsa.Activities.Mqtt.csproj new file mode 100644 index 000000000..86aec4bd3 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Elsa.Activities.Mqtt.csproj @@ -0,0 +1,25 @@ + + + + + + + + net5.0 + Elsa.Activities.Mqtt + Elsa is a set of workflow libraries and tools that enable lean and mean workflowing capabilities in any .NET Core application. +This package provides activities to send and receive messages using MQTT. + elsa, workflows + + + + + + + + + + + + + diff --git a/src/activities/Elsa.Activities.Mqtt/Extensions/ServiceCollectionExtensions.cs b/src/activities/Elsa.Activities.Mqtt/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 000000000..4ec9dde43 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Extensions/ServiceCollectionExtensions.cs @@ -0,0 +1,40 @@ +using Elsa.Activities.Mqtt.Bookmarks; +using Elsa.Activities.Mqtt.Consumers; +using Elsa.Activities.Mqtt.Services; +using Elsa.Activities.Mqtt.StartupTasks; +using Elsa.Activities.Mqtt.Testing; +using Elsa.Events; +using Elsa.Options; +using Elsa.Runtime; +using Elsa.Services; +using Microsoft.Extensions.DependencyInjection; + +namespace Elsa.Activities.Mqtt +{ + public static class ServiceCollectionExtensions + { + public static ElsaOptionsBuilder AddMqttActivities(this ElsaOptionsBuilder options) + { + options.Services + .AddSingleton() + .AddSingleton(sp => sp.GetRequiredService()) + .AddSingleton(sp => sp.GetRequiredService()) + .AddSingleton() + .AddSingleton>() + .AddStartupTask() + .AddBookmarkProvider() + .AddSingleton() + .AddNotificationHandlersFrom(); + + options.AddPubSubConsumer("WorkflowDefinitionEvents"); + options.AddPubSubConsumer("WorkflowDefinitionEvents"); + options.AddPubSubConsumer("WorkflowDefinitionEvents"); + + options + .AddActivity() + .AddActivity(); + + return options; + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/FodyWeavers.xml b/src/activities/Elsa.Activities.Mqtt/FodyWeavers.xml new file mode 100644 index 000000000..00e1d9a1c --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/FodyWeavers.xml @@ -0,0 +1,3 @@ + + + \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Options/MqttClientOptions.cs b/src/activities/Elsa.Activities.Mqtt/Options/MqttClientOptions.cs new file mode 100644 index 000000000..e28deb2cf --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Options/MqttClientOptions.cs @@ -0,0 +1,35 @@ +using System; +using System.Net.Mqtt; + +namespace Elsa.Activities.Mqtt.Options +{ + public class MqttClientOptions + { + public string Topic { get; } + public string Host { get; } + public int Port { get; } + public string Username { get; } + public string Password { get; } + public MqttQualityOfService QualityOfService { get; } + + public MqttClientOptions(string topic, string host, int port, string username, string password, MqttQualityOfService qos) + { + Topic = topic; + Host = host; + Port = port; + Username = username; + Password = password; + QualityOfService = qos; + } + + public MqttClientCredentials GenerateMqttClientCredentials() + { + return new MqttClientCredentials($"Elsa{ Guid.NewGuid():N}", Username, Password); + } + + public override int GetHashCode() + { + return HashCode.Combine(Topic, Host, Port, Username, Password); + } + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Services/BusClientFactory.cs b/src/activities/Elsa.Activities.Mqtt/Services/BusClientFactory.cs new file mode 100644 index 000000000..3196fc00a --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Services/BusClientFactory.cs @@ -0,0 +1,80 @@ +using Elsa.Activities.Mqtt.Options; +using Microsoft.Extensions.DependencyInjection; +using System; +using System.Collections.Generic; +using System.Net.Mqtt; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Services +{ + public class BusClientFactory : IMessageReceiverClientFactory, IMessageSenderClientFactory + { + private readonly IServiceProvider _serviceProvider; + private readonly IDictionary _senders = new Dictionary(); + private readonly IDictionary _receivers = new Dictionary(); + private readonly SemaphoreSlim _semaphore = new(1); + + public BusClientFactory(IServiceProvider serviceProvider) + { + _serviceProvider = serviceProvider; + } + + public async Task GetSenderAsync(MqttClientOptions options, CancellationToken cancellationToken) + { + await _semaphore.WaitAsync(cancellationToken); + + try + { + if (_senders.TryGetValue(options.GetHashCode(), out var messageSender)) + return messageSender; + + var newClient = await MqttClient.CreateAsync(options.Host, options.Port); + var newMessageSender = ActivatorUtilities.CreateInstance(_serviceProvider, newClient, options); + + _senders.Add(newMessageSender.Options.GetHashCode(), newMessageSender); + return newMessageSender; + } + finally + { + _semaphore.Release(); + } + } + + public async Task GetReceiverAsync(MqttClientOptions options, CancellationToken cancellationToken) + { + await _semaphore.WaitAsync(cancellationToken); + + try + { + if (_receivers.TryGetValue(options.GetHashCode(), out var messageReceiver)) + return messageReceiver; + + var newClient = await MqttClient.CreateAsync(options.Host, options.Port); + var newMessageReceiver = ActivatorUtilities.CreateInstance(_serviceProvider, newClient, options); + + _receivers.Add(options.GetHashCode(), newMessageReceiver); + return newMessageReceiver; + } + finally + { + _semaphore.Release(); + } + } + + public async Task DisposeReceiverAsync(IMqttClientWrapper receiverClient, CancellationToken cancellationToken = default) + { + await _semaphore.WaitAsync(cancellationToken); + + try + { + _receivers.Remove(receiverClient.Options.GetHashCode()); + receiverClient.Dispose(); + } + finally + { + _semaphore.Release(); + } + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Services/IMessageReceiverClientFactory.cs b/src/activities/Elsa.Activities.Mqtt/Services/IMessageReceiverClientFactory.cs new file mode 100644 index 000000000..d16b24da2 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Services/IMessageReceiverClientFactory.cs @@ -0,0 +1,12 @@ +using Elsa.Activities.Mqtt.Options; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Services +{ + public interface IMessageReceiverClientFactory + { + Task GetReceiverAsync(MqttClientOptions configuration, CancellationToken cancellationToken = default); + Task DisposeReceiverAsync(IMqttClientWrapper receiverClient, CancellationToken cancellationToken = default); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Services/IMessageSenderClientFactory.cs b/src/activities/Elsa.Activities.Mqtt/Services/IMessageSenderClientFactory.cs new file mode 100644 index 000000000..187e590f2 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Services/IMessageSenderClientFactory.cs @@ -0,0 +1,11 @@ +using Elsa.Activities.Mqtt.Options; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Services +{ + public interface IMessageSenderClientFactory + { + Task GetSenderAsync(MqttClientOptions configuration, CancellationToken cancellationToken = default); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Services/IMqttClientWrapper.cs b/src/activities/Elsa.Activities.Mqtt/Services/IMqttClientWrapper.cs new file mode 100644 index 000000000..0ce61bd64 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Services/IMqttClientWrapper.cs @@ -0,0 +1,17 @@ +using Elsa.Activities.Mqtt.Options; +using System; +using System.Net.Mqtt; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Services +{ + public interface IMqttClientWrapper : IDisposable + { + IMqttClient Client { get; } + MqttClientOptions Options { get; } + + Task SubscribeAsync(string topic); + Task PublishMessageAsync(string topic, string message); + void SetMessageHandler(Func handler); + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Services/IMqttTopicsStarter.cs b/src/activities/Elsa.Activities.Mqtt/Services/IMqttTopicsStarter.cs new file mode 100644 index 000000000..72f7f9f63 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Services/IMqttTopicsStarter.cs @@ -0,0 +1,16 @@ +using Elsa.Activities.Mqtt.Options; +using Elsa.Services.Models; +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Services +{ + public interface IMqttTopicsStarter + { + Task CreateWorkersAsync(CancellationToken cancellationToken = default); + Task CreateWorkerAsync(MqttClientOptions config, CancellationToken cancellationToken); + IAsyncEnumerable GetConfigurationsAsync(Func? predicate, CancellationToken cancellationToken); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Services/MqttClientWrapper.cs b/src/activities/Elsa.Activities.Mqtt/Services/MqttClientWrapper.cs new file mode 100644 index 000000000..d7ae45290 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Services/MqttClientWrapper.cs @@ -0,0 +1,78 @@ +using Elsa.Activities.Mqtt.Options; +using Microsoft.Extensions.Logging; +using System; +using System.Net.Mqtt; +using System.Text; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Services +{ + public class MqttClientWrapper : IMqttClientWrapper + { + private readonly ILogger _logger; + private Func? _messageHandler; + + public IMqttClient Client { get; } + public MqttClientOptions Options { get; } + + + public MqttClientWrapper(IMqttClient client, MqttClientOptions options, ILogger logger) + { + Client = client; + Options = options; + _logger = logger; + } + + public async Task SubscribeAsync(string topic) + { + await ConnectAsync(); + + await Client.SubscribeAsync(topic, Options.QualityOfService); + + Client.MessageStream.Subscribe(async message => + { + if (_messageHandler != null) + await _messageHandler(message); + else + _logger.LogWarning("Attempted to subscribe to topic {Topic}, but no message handler was set.", Options.Topic); + + await Client.UnsubscribeAsync(topic); + await DisconnectAsync(); + }); + } + + public async Task PublishMessageAsync(string topic, string message) + { + await ConnectAsync(); + + var applicationMessage = CreateApplicationMessage(topic, message); + + await Client.PublishAsync(applicationMessage, Options.QualityOfService); + + await DisconnectAsync(); + } + + public void SetMessageHandler(Func handler) + { + _messageHandler = handler; + } + + public void Dispose() + { + if (Client.IsConnected) DisconnectAsync().Wait(); + Client.Dispose(); + } + + private async Task ConnectAsync() + { + if (!Client.IsConnected) await Client.ConnectAsync(Options.GenerateMqttClientCredentials(), null, true); + } + + private async Task DisconnectAsync() + { + if (Client.IsConnected) await Client.DisconnectAsync(); + } + + private MqttApplicationMessage CreateApplicationMessage(string topic, string message) => new(topic, Encoding.UTF8.GetBytes(message)); + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Services/MqttTopicsStarter.cs b/src/activities/Elsa.Activities.Mqtt/Services/MqttTopicsStarter.cs new file mode 100644 index 000000000..2dc3b111e --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Services/MqttTopicsStarter.cs @@ -0,0 +1,118 @@ +using Elsa.Activities.Mqtt.Options; +using Elsa.Persistence; +using Elsa.Persistence.Specifications.WorkflowInstances; +using Elsa.Services; +using Elsa.Services.Models; +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.Mqtt.Services +{ + public class MqttTopicsStarter : IMqttTopicsStarter + { + private readonly IMessageReceiverClientFactory _receiverFactory; + private readonly IServiceScopeFactory _scopeFactory; + private readonly IServiceProvider _serviceProvider; + private readonly ILogger _logger; + private readonly ICollection _workers; + + public MqttTopicsStarter( + IMessageReceiverClientFactory receiverFactory, + IServiceScopeFactory scopeFactory, + IServiceProvider serviceProvider, + ILogger logger) + { + _receiverFactory = receiverFactory; + _scopeFactory = scopeFactory; + _serviceProvider = serviceProvider; + _logger = logger; + _workers = new List(); + } + + public async Task CreateWorkersAsync(CancellationToken cancellationToken) + { + await DisposeExistingWorkersAsync(); + var configs = (await GetConfigurationsAsync(null, cancellationToken).ToListAsync(cancellationToken)).Distinct(); + + foreach (var config in configs) + { + try + { + _workers.Add(await CreateWorkerAsync(config, cancellationToken)); + } + catch (Exception e) + { + _logger.LogWarning(e, "Failed to create a receiver for topic {Topic}", config.Topic); + } + } + } + + public async Task CreateWorkerAsync(MqttClientOptions config, CancellationToken cancellationToken) + { + var receiver = await _receiverFactory.GetReceiverAsync(config, cancellationToken); + return ActivatorUtilities.CreateInstance(_serviceProvider, receiver, (Func)DisposeReceiverAsync); + } + + public async IAsyncEnumerable GetConfigurationsAsync(Func? predicate, [EnumeratorCancellation] CancellationToken cancellationToken) + { + using var scope = _scopeFactory.CreateScope(); + var workflowRegistry = scope.ServiceProvider.GetRequiredService(); + var workflowBlueprintReflector = scope.ServiceProvider.GetRequiredService(); + var workflowInstanceStore = scope.ServiceProvider.GetRequiredService(); + var workflows = await workflowRegistry.ListActiveAsync(cancellationToken); + + var query = + from workflow in workflows + from activity in workflow.Activities + where activity.Type == nameof(MqttMessageReceived) + select workflow; + + var filteredQuery = predicate == null ? query : query.Where(predicate); + + foreach (var workflow in query) + { + // If a workflow is not published, only consider it for processing if it has at least one non-ended workflow instance. + if (!workflow.IsPublished && !await WorkflowHasNonFinishedWorkflowsAsync(workflow, workflowInstanceStore, cancellationToken)) + continue; + + var workflowBlueprintWrapper = await workflowBlueprintReflector.ReflectAsync(scope.ServiceProvider, workflow, cancellationToken); + + foreach (var activity in workflowBlueprintWrapper.Filter()) + { + var topic = await activity.EvaluatePropertyValueAsync(x => x.Topic, cancellationToken); + var host = await activity.EvaluatePropertyValueAsync(x => x.Host, cancellationToken); + var port = await activity.EvaluatePropertyValueAsync(x => x.Port, cancellationToken); + var username = await activity.EvaluatePropertyValueAsync(x => x.Username, cancellationToken); + var password = await activity.EvaluatePropertyValueAsync(x => x.Password, cancellationToken); + var qos = await activity.EvaluatePropertyValueAsync(x => x.QualityOfService, cancellationToken); + + yield return new MqttClientOptions(topic!, host!, port!, username!, password!, qos); + } + } + } + + private async Task DisposeExistingWorkersAsync() + { + foreach (var worker in _workers.ToList()) + { + await worker.DisposeAsync(); + _workers.Remove(worker); + } + } + + private async Task DisposeReceiverAsync(IMqttClientWrapper messageReceiver) => await _receiverFactory.DisposeReceiverAsync(messageReceiver); + + + private static async Task WorkflowHasNonFinishedWorkflowsAsync(IWorkflowBlueprint workflowBlueprint, IWorkflowInstanceStore workflowInstanceStore, CancellationToken cancellationToken) + { + var count = await workflowInstanceStore.CountAsync(new UnfinishedWorkflowSpecification().WithWorkflowDefinition(workflowBlueprint.Id), cancellationToken); + return count > 0; + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Services/Worker.cs b/src/activities/Elsa.Activities.Mqtt/Services/Worker.cs new file mode 100644 index 000000000..0619e21e8 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Services/Worker.cs @@ -0,0 +1,46 @@ +using Elsa.Activities.Mqtt.Bookmarks; +using Elsa.Activities.Mqtt.Options; +using Elsa.Models; +using Elsa.Services; +using Elsa.Services.Models; +using System; +using System.Net.Mqtt; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Services +{ + public class Worker + { + private readonly Scoped _workflowLaunchpad; + private readonly Func _disposeReceiverAction; + private readonly IMqttClientWrapper _receiverClient; + + public Worker( + IMqttClientWrapper receiverClient, + Scoped workflowLaunchpad, + Func disposeReceiverAction) + { + _receiverClient = receiverClient; + _workflowLaunchpad = workflowLaunchpad; + _disposeReceiverAction = disposeReceiverAction; + + _receiverClient.SetMessageHandler(OnMessageReceived); + } + + private string ActivityType => nameof(MqttMessageReceived); + public async ValueTask DisposeAsync() => await _disposeReceiverAction(_receiverClient); + + private IBookmark CreateBookmark(MqttClientOptions options) => new MessageReceivedBookmark(options.Topic, options.Host, options.Port, options.Username, options.Password); + + private async Task TriggerWorkflowsAsync(MqttApplicationMessage message, CancellationToken cancellationToken) + { + var bookmark = CreateBookmark(_receiverClient.Options); + var launchContext = new WorkflowsQuery(ActivityType, bookmark); + + await _workflowLaunchpad.UseServiceAsync(service => service.CollectAndDispatchWorkflowsAsync(launchContext, new WorkflowInput(message), cancellationToken)); + } + + private async Task OnMessageReceived(MqttApplicationMessage message) => await TriggerWorkflowsAsync(message, CancellationToken.None); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Startup.cs b/src/activities/Elsa.Activities.Mqtt/Startup.cs new file mode 100644 index 000000000..0e36e98af --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Startup.cs @@ -0,0 +1,16 @@ +using Elsa.Attributes; +using Elsa.Options; +using Elsa.Services.Startup; +using Microsoft.Extensions.Configuration; + +namespace Elsa.Activities.Mqtt +{ + [Feature("Mqtt")] + public class Startup : StartupBase + { + public override void ConfigureElsa(ElsaOptionsBuilder elsa, IConfiguration configuration) + { + elsa.AddMqttActivities(); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/StartupTasks/StartMqttTopics.cs b/src/activities/Elsa.Activities.Mqtt/StartupTasks/StartMqttTopics.cs new file mode 100644 index 000000000..89cff2b1f --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/StartupTasks/StartMqttTopics.cs @@ -0,0 +1,15 @@ +using System.Threading; +using System.Threading.Tasks; +using Elsa.Activities.Mqtt.Services; +using Elsa.Services; + +namespace Elsa.Activities.Mqtt.StartupTasks +{ + public class StartMqttTopics : IStartupTask + { + private readonly IMqttTopicsStarter _mqttTopicsStarter; + public StartMqttTopics(IMqttTopicsStarter mqttTopicsStarter) => _mqttTopicsStarter = mqttTopicsStarter; + public int Order => 2000; + public Task ExecuteAsync(CancellationToken stoppingToken) => _mqttTopicsStarter.CreateWorkersAsync(stoppingToken); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Testing/Handlers/ConfigureMqttActivitiesForTestHandler.cs b/src/activities/Elsa.Activities.Mqtt/Testing/Handlers/ConfigureMqttActivitiesForTestHandler.cs new file mode 100644 index 000000000..336456737 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Testing/Handlers/ConfigureMqttActivitiesForTestHandler.cs @@ -0,0 +1,59 @@ +using Elsa.Events; +using Elsa.WorkflowTesting.Events; +using MediatR; +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Testing +{ + public class ConfigureMqttActivitiesForTestHandler : INotificationHandler, INotificationHandler, INotificationHandler, INotificationHandler + { + private readonly IMqttTestClientManager _mqttTestClientManager; + + public ConfigureMqttActivitiesForTestHandler(IMqttTestClientManager mqttTestClientManager) + { + _mqttTestClientManager = mqttTestClientManager; + } + + 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 _mqttTestClientManager.CreateTestWorkersAsync(workflowId, workflowInstanceId, cancellationToken); + } + + public async Task Handle(WorkflowFaulted notification, CancellationToken cancellationToken) + { + await HandleTestWorkflowExecutionFinished(notification, cancellationToken); + } + + public async Task Handle(WorkflowCompleted notification, CancellationToken cancellationToken) + { + await HandleTestWorkflowExecutionFinished(notification, cancellationToken); + } + + public async Task Handle(WorkflowTestExecutionStopped notification, CancellationToken cancellationToken) + { + await _mqttTestClientManager.DisposeTestWorkersAsync(notification.WorkflowInstanceId); + } + + private async Task HandleTestWorkflowExecutionFinished(WorkflowNotification notification, CancellationToken cancellationToken) + { + var isTest = Convert.ToBoolean(notification.WorkflowExecutionContext.WorkflowInstance.GetMetadata("isTest")); + + if (!isTest) return; + + var workflowInstanceId = notification.WorkflowExecutionContext.WorkflowInstance.Id; + + await _mqttTestClientManager.DisposeTestWorkersAsync(workflowInstanceId); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.Mqtt/Testing/IMqttTestClientManager.cs b/src/activities/Elsa.Activities.Mqtt/Testing/IMqttTestClientManager.cs new file mode 100644 index 000000000..4c6ea15b6 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Testing/IMqttTestClientManager.cs @@ -0,0 +1,11 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Testing +{ + public interface IMqttTestClientManager + { + Task CreateTestWorkersAsync(string workflowId, string workflowInstanceId, CancellationToken cancellationToken = default); + Task DisposeTestWorkersAsync(string workflowInstanceId); + } +} diff --git a/src/activities/Elsa.Activities.Mqtt/Testing/MqttTestClientManager.cs b/src/activities/Elsa.Activities.Mqtt/Testing/MqttTestClientManager.cs new file mode 100644 index 000000000..a24111402 --- /dev/null +++ b/src/activities/Elsa.Activities.Mqtt/Testing/MqttTestClientManager.cs @@ -0,0 +1,68 @@ +using Elsa.Activities.Mqtt.Services; +using Microsoft.Extensions.Logging; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +namespace Elsa.Activities.Mqtt.Testing +{ + public class MqttTestClientManager : IMqttTestClientManager + { + private readonly SemaphoreSlim _semaphore = new(1); + private readonly IDictionary> _workers; + private readonly IMqttTopicsStarter _mqttTopicsStarter; + private readonly ILogger _logger; + + public MqttTestClientManager( + IMqttTopicsStarter mqttTopicsStarter, + ILogger logger) + { + _mqttTopicsStarter = mqttTopicsStarter; + _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)) + { + _workers[workflowInstanceId] = new List(); + } + + var receiverConfigs = (await _mqttTopicsStarter.GetConfigurationsAsync(x => x.Id == workflowId, cancellationToken).ToListAsync(cancellationToken)).Distinct(); + + foreach (var config in receiverConfigs) + { + try + { + _workers[workflowInstanceId].Add(await _mqttTopicsStarter.CreateWorkerAsync(config, cancellationToken)); + } + catch (Exception e) + { + _logger.LogWarning(e, "Failed to create a test receiver for topic {Topic}", config.Topic); + } + } + } + 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/PropertyCategories.cs b/src/core/Elsa.Abstractions/Design/PropertyCategories.cs index e7c5abccf..b1613878c 100644 --- a/src/core/Elsa.Abstractions/Design/PropertyCategories.cs +++ b/src/core/Elsa.Abstractions/Design/PropertyCategories.cs @@ -1,4 +1,4 @@ -namespace Elsa.Design +namespace Elsa.Design { public static class PropertyCategories { 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 ba1921b20..81643b189 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 @@ -19,6 +19,7 @@ + diff --git a/src/samples/server/Elsa.Samples.Server.Host/Startup.cs b/src/samples/server/Elsa.Samples.Server.Host/Startup.cs index 3f89dcdd1..c000fcc09 100644 --- a/src/samples/server/Elsa.Samples.Server.Host/Startup.cs +++ b/src/samples/server/Elsa.Samples.Server.Host/Startup.cs @@ -44,6 +44,7 @@ namespace Elsa.Samples.Server.Host typeof(Elsa.Activities.Telnyx.Startup), typeof(Elsa.Activities.File.Startup), typeof(Elsa.Activities.RabbitMq.Startup), + typeof(Elsa.Activities.Mqtt.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 6dbd59621..b31894c6a 100644 --- a/src/samples/server/Elsa.Samples.Server.Host/appsettings.json +++ b/src/samples/server/Elsa.Samples.Server.Host/appsettings.json @@ -54,6 +54,9 @@ "RabbitMq": { "Enabled": true, "ConnectionStringIdentifier": "RabbitMq" + }, + "Mqtt": { + "Enabled": true } }, "WorkflowChannels": { diff --git a/src/samples/worker/Elsa.Samples.MqttWorker/Elsa.Samples.MqttWorker.csproj b/src/samples/worker/Elsa.Samples.MqttWorker/Elsa.Samples.MqttWorker.csproj new file mode 100644 index 000000000..8a37f6adc --- /dev/null +++ b/src/samples/worker/Elsa.Samples.MqttWorker/Elsa.Samples.MqttWorker.csproj @@ -0,0 +1,20 @@ + + + + Exe + net5.0 + enable + latest + + + + + + + + + + + + + diff --git a/src/samples/worker/Elsa.Samples.MqttWorker/Program.cs b/src/samples/worker/Elsa.Samples.MqttWorker/Program.cs new file mode 100644 index 000000000..8ee2bcf5c --- /dev/null +++ b/src/samples/worker/Elsa.Samples.MqttWorker/Program.cs @@ -0,0 +1,29 @@ +using Elsa.Activities.Mqtt; +using Elsa.Samples.MqttWorker.Workflows; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; + +namespace Elsa.Samples.MqttWorker +{ + 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() + .AddMqttActivities() + .AddWorkflow() + .AddWorkflow() + ); + }); + } +} \ No newline at end of file diff --git a/src/samples/worker/Elsa.Samples.MqttWorker/Properties/launchSettings.json b/src/samples/worker/Elsa.Samples.MqttWorker/Properties/launchSettings.json new file mode 100644 index 000000000..a509725e9 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.MqttWorker/Properties/launchSettings.json @@ -0,0 +1,11 @@ +{ + "profiles": { + "Elsa.Samples.MqttWorker": { + "commandName": "Project", + "dotnetRunMessages": "true", + "environmentVariables": { + "DOTNET_ENVIRONMENT": "Development" + } + } + } +} diff --git a/src/samples/worker/Elsa.Samples.MqttWorker/Workflows/ConsumerWorkflow.cs b/src/samples/worker/Elsa.Samples.MqttWorker/Workflows/ConsumerWorkflow.cs new file mode 100644 index 000000000..b7c439c07 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.MqttWorker/Workflows/ConsumerWorkflow.cs @@ -0,0 +1,41 @@ +using Elsa.Activities.Console; +using Elsa.Activities.Mqtt; +using Elsa.Activities.Temporal; +using Elsa.Builders; +using Microsoft.Extensions.Configuration; +using NodaTime; +using System.Net.Mqtt; + +namespace Elsa.Samples.MqttWorker.Workflows +{ + public class ConsumerWorkflow : IWorkflow + { + private readonly string _host; + private readonly int _port; + private readonly string _username; + private readonly string _password; + private readonly MqttQualityOfService _qos; + public ConsumerWorkflow(IConfiguration configuration) + { + var section = configuration.GetSection("Mqtt"); + + _host = section.GetValue("Host"); + _port = section.GetValue("Port"); + _username = section.GetValue("Username"); + _password = section.GetValue("Password"); + _qos = section.GetValue("QualityOfService"); + } + + public void Build(IWorkflowBuilder builder) + { + builder + .Timer(Duration.FromSeconds(5)) + .MessageReceived("/temperature", _host, _port, _username, _password, _qos) + .WriteLine(context => + { + var message = context.GetInput(); + return $"Received a temperature update saying {message}!"; + }); + } + } +} \ No newline at end of file diff --git a/src/samples/worker/Elsa.Samples.MqttWorker/Workflows/ProducerWorkflow.cs b/src/samples/worker/Elsa.Samples.MqttWorker/Workflows/ProducerWorkflow.cs new file mode 100644 index 000000000..27eb02a62 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.MqttWorker/Workflows/ProducerWorkflow.cs @@ -0,0 +1,42 @@ +using Elsa.Activities.Console; +using Elsa.Activities.Mqtt; +using Elsa.Activities.Temporal; +using Elsa.Builders; +using Microsoft.Extensions.Configuration; +using NodaTime; +using System; +using System.Net.Mqtt; + +namespace Elsa.Samples.MqttWorker.Workflows +{ + public class ProducerWorkflow : IWorkflow + { + private readonly string _host; + private readonly int _port; + private readonly string _username; + private readonly string _password; + private readonly MqttQualityOfService _qos; + private readonly Random _random; + public ProducerWorkflow(IConfiguration configuration) + { + var section = configuration.GetSection("Mqtt"); + + _host = section.GetValue("Host"); + _port = section.GetValue("Port"); + _username = section.GetValue("Username"); + _password = section.GetValue("Password"); + _qos = section.GetValue("QualityOfService"); + _random = new(); + } + + public void Build(IWorkflowBuilder builder) + { + builder + .Timer(Duration.FromSeconds(5)) + .WriteLine("Sending a temperature update with the \"/temperature\" topic.") + .SendMessage("/temperature", _host, _port, _username, _password, _qos, GetRandomTemperature); + } + + private string GetRandomTemperature() => $"{_random.Next(4, 32)} degrees Celsium"; + } +} \ No newline at end of file diff --git a/src/samples/worker/Elsa.Samples.MqttWorker/appsettings.Development.json b/src/samples/worker/Elsa.Samples.MqttWorker/appsettings.Development.json new file mode 100644 index 000000000..8983e0fc1 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.MqttWorker/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.MqttWorker/appsettings.json b/src/samples/worker/Elsa.Samples.MqttWorker/appsettings.json new file mode 100644 index 000000000..de241be54 --- /dev/null +++ b/src/samples/worker/Elsa.Samples.MqttWorker/appsettings.json @@ -0,0 +1,17 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Information", + "Microsoft": "Warning", + "Microsoft.Hosting.Lifetime": "Information" + } + }, + "Mqtt": { + "Host": "", + "Port": 1883, + "Username": "", + "Password": "", + "_comment": "QualityOfService: [0: AtMostOnce, 1: AtLeastOnce, 2: ExactlyOnce]", + "QualityOfService": 1 + } +}