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

Closes #2489, MQTT activity (#2615)

Browse source 
* added MQTT activity

* Added common.props to Elsa.Activities.Mqtt, downgraded to .NET 5

* updated project settings for MQTT

* removed implicit usings from MQTT projects
This commit is contained in:
Konstantins Vedenins 2021-12-22 12:49:57 +02:00 committed by GitHub
parent 8675baf1ae
commit 5c6dc307f5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
38 changed files with 1201 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
Elsa.sln
View file

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

53
src/activities/Elsa.Activities.Mqtt/Activities/MqttBaseActivity.cs Normal file
View file

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

57
src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceived.cs Normal file
View file

@ -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<IActivityExecutionResult> 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<IActivityExecutionResult> SuspendInternalAsync()
{
var options = new MqttClientOptions(Topic, Host, Port, Username, Password, QualityOfService);
var receiver = await _messageReceiver.GetReceiverAsync(options);
await receiver.SubscribeAsync(Topic);
return Suspend();
}
}
}

16
src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceivedBuilderExtensions.cs Normal file
View file

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

47
src/activities/Elsa.Activities.Mqtt/Activities/MqttMessageReceived/MqttMessageReceivedExtensions.cs Normal file
View file

@ -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<MqttMessageReceived> WithTopic(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<MqttMessageReceived> WithTopic(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<MqttMessageReceived> WithTopic(this ISetupActivity<MqttMessageReceived> messageReceived, Func<string> value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<MqttMessageReceived> WithTopic(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<MqttMessageReceived> WithTopic(this ISetupActivity<MqttMessageReceived> messageReceived, string value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<MqttMessageReceived> WithHost(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<MqttMessageReceived> WithHost(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<MqttMessageReceived> WithHost(this ISetupActivity<MqttMessageReceived> messageReceived, Func<string> value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<MqttMessageReceived> WithHost(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<MqttMessageReceived> WithHost(this ISetupActivity<MqttMessageReceived> messageReceived, string value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<MqttMessageReceived> WithPort(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<int>> value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<MqttMessageReceived> WithPort(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ValueTask<int>> value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<MqttMessageReceived> WithPort(this ISetupActivity<MqttMessageReceived> messageReceived, Func<int> value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<MqttMessageReceived> WithPort(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, int> value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<MqttMessageReceived> WithPort(this ISetupActivity<MqttMessageReceived> messageReceived, int value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<MqttMessageReceived> WithUsername(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<MqttMessageReceived> WithUsername(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<MqttMessageReceived> WithUsername(this ISetupActivity<MqttMessageReceived> messageReceived, Func<string> value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<MqttMessageReceived> WithUsername(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<MqttMessageReceived> WithUsername(this ISetupActivity<MqttMessageReceived> messageReceived, string value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<MqttMessageReceived> WithPassword(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<MqttMessageReceived> WithPassword(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<MqttMessageReceived> WithPassword(this ISetupActivity<MqttMessageReceived> messageReceived, Func<string> value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<MqttMessageReceived> WithPassword(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<MqttMessageReceived> WithPassword(this ISetupActivity<MqttMessageReceived> messageReceived, string value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<MqttMessageReceived> WithQualityOfService(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, ValueTask<MqttQualityOfService>> value) => messageReceived.Set(x => x.QualityOfService, value!);
public static ISetupActivity<MqttMessageReceived> WithQualityOfService(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ValueTask<MqttQualityOfService>> value) => messageReceived.Set(x => x.QualityOfService, value!);
public static ISetupActivity<MqttMessageReceived> WithQualityOfService(this ISetupActivity<MqttMessageReceived> messageReceived, Func<MqttQualityOfService> value) => messageReceived.Set(x => x.QualityOfService, value!);
public static ISetupActivity<MqttMessageReceived> WithQualityOfService(this ISetupActivity<MqttMessageReceived> messageReceived, Func<ActivityExecutionContext, MqttQualityOfService> value) => messageReceived.Set(x => x.QualityOfService, value!);
public static ISetupActivity<MqttMessageReceived> WithQualityOfService(this ISetupActivity<MqttMessageReceived> messageReceived, MqttQualityOfService value) => messageReceived.Set(x => x.QualityOfService, value!);
}
}

48
src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessage.cs Normal file
View file

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

27
src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessageBuilderExtensions.cs Normal file
View file

@ -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<ISetupActivity<SendMqttMessage>> 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<ActivityExecutionContext, ValueTask<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);
public static IActivityBuilder SendMessage(this IBuilder builder, string topic, string host, int port, string username, string password, MqttQualityOfService qos, Func<ActivityExecutionContext, 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);
public static IActivityBuilder SendMessage(this IBuilder builder, string topic, string host, int port, string username, string password, MqttQualityOfService qos, Func<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);
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);
}
}

53
src/activities/Elsa.Activities.Mqtt/Activities/SendMqttMessage/SendMqttMessageExtensions.cs Normal file
View file

@ -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<SendMqttMessage> WithTopic(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<SendMqttMessage> WithTopic(this ISetupActivity<SendMqttMessage> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<SendMqttMessage> WithTopic(this ISetupActivity<SendMqttMessage> messageReceived, Func<string> value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<SendMqttMessage> WithTopic(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<SendMqttMessage> WithTopic(this ISetupActivity<SendMqttMessage> messageReceived, string value) => messageReceived.Set(x => x.Topic, value!);
public static ISetupActivity<SendMqttMessage> WithHost(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<SendMqttMessage> WithHost(this ISetupActivity<SendMqttMessage> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<SendMqttMessage> WithHost(this ISetupActivity<SendMqttMessage> messageReceived, Func<string> value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<SendMqttMessage> WithHost(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<SendMqttMessage> WithHost(this ISetupActivity<SendMqttMessage> messageReceived, string value) => messageReceived.Set(x => x.Host, value!);
public static ISetupActivity<SendMqttMessage> WithPort(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, ValueTask<int>> value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<SendMqttMessage> WithPort(this ISetupActivity<SendMqttMessage> messageReceived, Func<ValueTask<int>> value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<SendMqttMessage> WithPort(this ISetupActivity<SendMqttMessage> messageReceived, Func<int> value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<SendMqttMessage> WithPort(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, int> value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<SendMqttMessage> WithPort(this ISetupActivity<SendMqttMessage> messageReceived, int value) => messageReceived.Set(x => x.Port, value!);
public static ISetupActivity<SendMqttMessage> WithUsername(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<SendMqttMessage> WithUsername(this ISetupActivity<SendMqttMessage> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<SendMqttMessage> WithUsername(this ISetupActivity<SendMqttMessage> messageReceived, Func<string> value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<SendMqttMessage> WithUsername(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<SendMqttMessage> WithUsername(this ISetupActivity<SendMqttMessage> messageReceived, string value) => messageReceived.Set(x => x.Username, value!);
public static ISetupActivity<SendMqttMessage> WithPassword(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<SendMqttMessage> WithPassword(this ISetupActivity<SendMqttMessage> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<SendMqttMessage> WithPassword(this ISetupActivity<SendMqttMessage> messageReceived, Func<string> value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<SendMqttMessage> WithPassword(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<SendMqttMessage> WithPassword(this ISetupActivity<SendMqttMessage> messageReceived, string value) => messageReceived.Set(x => x.Password, value!);
public static ISetupActivity<SendMqttMessage> WithMessage(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, ValueTask<string>> value) => messageReceived.Set(x => x.Message, value!);
public static ISetupActivity<SendMqttMessage> WithMessage(this ISetupActivity<SendMqttMessage> messageReceived, Func<ValueTask<string>> value) => messageReceived.Set(x => x.Message, value!);
public static ISetupActivity<SendMqttMessage> WithMessage(this ISetupActivity<SendMqttMessage> messageReceived, Func<string> value) => messageReceived.Set(x => x.Message, value!);
public static ISetupActivity<SendMqttMessage> WithMessage(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, string> value) => messageReceived.Set(x => x.Message, value!);
public static ISetupActivity<SendMqttMessage> WithMessage(this ISetupActivity<SendMqttMessage> messageReceived, string value) => messageReceived.Set(x => x.Message, value!);
public static ISetupActivity<SendMqttMessage> WithQualityOfService(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, ValueTask<MqttQualityOfService>> value) => messageReceived.Set(x => x.QualityOfService, value!);
public static ISetupActivity<SendMqttMessage> WithQualityOfService(this ISetupActivity<SendMqttMessage> messageReceived, Func<ValueTask<MqttQualityOfService>> value) => messageReceived.Set(x => x.QualityOfService, value!);
public static ISetupActivity<SendMqttMessage> WithQualityOfService(this ISetupActivity<SendMqttMessage> messageReceived, Func<MqttQualityOfService> value) => messageReceived.Set(x => x.QualityOfService, value!);
public static ISetupActivity<SendMqttMessage> WithQualityOfService(this ISetupActivity<SendMqttMessage> messageReceived, Func<ActivityExecutionContext, MqttQualityOfService> value) => messageReceived.Set(x => x.QualityOfService, value!);
public static ISetupActivity<SendMqttMessage> WithQualityOfService(this ISetupActivity<SendMqttMessage> messageReceived, MqttQualityOfService value) => messageReceived.Set(x => x.QualityOfService, value!);
}
}

45
src/activities/Elsa.Activities.Mqtt/Bookmarks/MessageReceivedBookmark.cs Normal file
View file

@ -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<MessageReceivedBookmark, MqttMessageReceived>
{
public override async ValueTask<IEnumerable<BookmarkResult>> GetBookmarksAsync(BookmarkProviderContext<MqttMessageReceived> 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))!
})
};
}
}

16
src/activities/Elsa.Activities.Mqtt/Consumers/RestartMqttTopicsConsumer.cs Normal file
View file

@ -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<WorkflowDefinitionPublished>, IHandleMessages<WorkflowDefinitionRetracted>, IHandleMessages<WorkflowDefinitionDeleted>
{
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();
}
}

25
src/activities/Elsa.Activities.Mqtt/Elsa.Activities.Mqtt.csproj Normal file
View file

@ -0,0 +1,25 @@
<Project Sdk="Microsoft.NET.Sdk">
<Import Project="..\..\..\common.props" />
<Import Project="..\..\..\configureawait.props" />
<PropertyGroup>
<TargetFramework>net5.0</TargetFramework>
<RootNamespace>Elsa.Activities.Mqtt</RootNamespace>
<Description>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.</Description>
<PackageTags>elsa, workflows</PackageTags>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\core\Elsa.Abstractions\Elsa.Abstractions.csproj" />
<ProjectReference Include="..\..\core\Elsa.Core\Elsa.Core.csproj" />
<ProjectReference Include="..\..\modules\workflowtesting\Elsa.WorkflowTesting\Elsa.WorkflowTesting.csproj" />
</ItemGroup>
<ItemGroup>
<PackageReference Include="System.Net.Mqtt" Version="0.6.16-beta" />
</ItemGroup>
</Project>

40
src/activities/Elsa.Activities.Mqtt/Extensions/ServiceCollectionExtensions.cs Normal file
View file

@ -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<BusClientFactory>()
.AddSingleton<IMessageReceiverClientFactory>(sp => sp.GetRequiredService<BusClientFactory>())
.AddSingleton<IMessageSenderClientFactory>(sp => sp.GetRequiredService<BusClientFactory>())
.AddSingleton<IMqttTopicsStarter, MqttTopicsStarter>()
.AddSingleton<Scoped<IWorkflowLaunchpad>>()
.AddStartupTask<StartMqttTopics>()
.AddBookmarkProvider<MessageReceivedBookmarkProvider>()
.AddSingleton<IMqttTestClientManager, MqttTestClientManager>()
.AddNotificationHandlersFrom<ConfigureMqttActivitiesForTestHandler>();
options.AddPubSubConsumer<RestartMqttTopicsConsumer, WorkflowDefinitionPublished>("WorkflowDefinitionEvents");
options.AddPubSubConsumer<RestartMqttTopicsConsumer, WorkflowDefinitionRetracted>("WorkflowDefinitionEvents");
options.AddPubSubConsumer<RestartMqttTopicsConsumer, WorkflowDefinitionDeleted>("WorkflowDefinitionEvents");
options
.AddActivity<MqttMessageReceived>()
.AddActivity<SendMqttMessage>();
return options;
}
}
}

3
src/activities/Elsa.Activities.Mqtt/FodyWeavers.xml Normal file
View file

@ -0,0 +1,3 @@
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd">
<ConfigureAwait />
</Weavers>

35
src/activities/Elsa.Activities.Mqtt/Options/MqttClientOptions.cs Normal file
View file

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

80
src/activities/Elsa.Activities.Mqtt/Services/BusClientFactory.cs Normal file
View file

@ -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<int, IMqttClientWrapper> _senders = new Dictionary<int, IMqttClientWrapper>();
private readonly IDictionary<int, IMqttClientWrapper> _receivers = new Dictionary<int, IMqttClientWrapper>();
private readonly SemaphoreSlim _semaphore = new(1);
public BusClientFactory(IServiceProvider serviceProvider)
{
_serviceProvider = serviceProvider;
}
public async Task<IMqttClientWrapper> 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<MqttClientWrapper>(_serviceProvider, newClient, options);
_senders.Add(newMessageSender.Options.GetHashCode(), newMessageSender);
return newMessageSender;
}
finally
{
_semaphore.Release();
}
}
public async Task<IMqttClientWrapper> 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<MqttClientWrapper>(_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();
}
}
}
}

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

@ -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<IMqttClientWrapper> GetReceiverAsync(MqttClientOptions configuration, CancellationToken cancellationToken = default);
Task DisposeReceiverAsync(IMqttClientWrapper receiverClient, CancellationToken cancellationToken = default);
}
}

11
src/activities/Elsa.Activities.Mqtt/Services/IMessageSenderClientFactory.cs Normal file
View file

@ -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<IMqttClientWrapper> GetSenderAsync(MqttClientOptions configuration, CancellationToken cancellationToken = default);
}
}

17
src/activities/Elsa.Activities.Mqtt/Services/IMqttClientWrapper.cs Normal file
View file

@ -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<MqttApplicationMessage, Task> handler);
}
}

16
src/activities/Elsa.Activities.Mqtt/Services/IMqttTopicsStarter.cs Normal file
View file

@ -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<Worker> CreateWorkerAsync(MqttClientOptions config, CancellationToken cancellationToken);
IAsyncEnumerable<MqttClientOptions> GetConfigurationsAsync(Func<IWorkflowBlueprint, bool>? predicate, CancellationToken cancellationToken);
}
}

78
src/activities/Elsa.Activities.Mqtt/Services/MqttClientWrapper.cs Normal file
View file

@ -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<MqttApplicationMessage, Task>? _messageHandler;
public IMqttClient Client { get; }
public MqttClientOptions Options { get; }
public MqttClientWrapper(IMqttClient client, MqttClientOptions options, ILogger<MqttClientWrapper> 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<MqttApplicationMessage, Task> 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));
}
}

118
src/activities/Elsa.Activities.Mqtt/Services/MqttTopicsStarter.cs Normal file
View file

@ -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<MqttTopicsStarter> _logger;
private readonly ICollection<Worker> _workers;
public MqttTopicsStarter(
IMessageReceiverClientFactory receiverFactory,
IServiceScopeFactory scopeFactory,
IServiceProvider serviceProvider,
ILogger<MqttTopicsStarter> logger)
{
_receiverFactory = receiverFactory;
_scopeFactory = scopeFactory;
_serviceProvider = serviceProvider;
_logger = logger;
_workers = new List<Worker>();
}
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<Worker> CreateWorkerAsync(MqttClientOptions config, CancellationToken cancellationToken)
{
var receiver = await _receiverFactory.GetReceiverAsync(config, cancellationToken);
return ActivatorUtilities.CreateInstance<Worker>(_serviceProvider, receiver, (Func<IMqttClientWrapper, Task>)DisposeReceiverAsync);
}
public async IAsyncEnumerable<MqttClientOptions> GetConfigurationsAsync(Func<IWorkflowBlueprint, bool>? predicate, [EnumeratorCancellation] CancellationToken cancellationToken)
{
using var scope = _scopeFactory.CreateScope();
var workflowRegistry = scope.ServiceProvider.GetRequiredService<IWorkflowRegistry>();
var workflowBlueprintReflector = scope.ServiceProvider.GetRequiredService<IWorkflowBlueprintReflector>();
var workflowInstanceStore = scope.ServiceProvider.GetRequiredService<IWorkflowInstanceStore>();
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<MqttMessageReceived>())
{
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<bool> WorkflowHasNonFinishedWorkflowsAsync(IWorkflowBlueprint workflowBlueprint, IWorkflowInstanceStore workflowInstanceStore, CancellationToken cancellationToken)
{
var count = await workflowInstanceStore.CountAsync(new UnfinishedWorkflowSpecification().WithWorkflowDefinition(workflowBlueprint.Id), cancellationToken);
return count > 0;
}
}
}

46
src/activities/Elsa.Activities.Mqtt/Services/Worker.cs Normal file
View file

@ -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<IWorkflowLaunchpad> _workflowLaunchpad;
private readonly Func<IMqttClientWrapper, Task> _disposeReceiverAction;
private readonly IMqttClientWrapper _receiverClient;
public Worker(
IMqttClientWrapper receiverClient,
Scoped<IWorkflowLaunchpad> workflowLaunchpad,
Func<IMqttClientWrapper, Task> 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);
}
}

16
src/activities/Elsa.Activities.Mqtt/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.Mqtt
{
[Feature("Mqtt")]
public class Startup : StartupBase
{
public override void ConfigureElsa(ElsaOptionsBuilder elsa, IConfiguration configuration)
{
elsa.AddMqttActivities();
}
}
}

15
src/activities/Elsa.Activities.Mqtt/StartupTasks/StartMqttTopics.cs Normal file
View file

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

59
src/activities/Elsa.Activities.Mqtt/Testing/Handlers/ConfigureMqttActivitiesForTestHandler.cs Normal file
View file

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

11
src/activities/Elsa.Activities.Mqtt/Testing/IMqttTestClientManager.cs Normal file
View file

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

68
src/activities/Elsa.Activities.Mqtt/Testing/MqttTestClientManager.cs Normal file
View file

@ -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<string, ICollection<Worker>> _workers;
private readonly IMqttTopicsStarter _mqttTopicsStarter;
private readonly ILogger _logger;
public MqttTestClientManager(
IMqttTopicsStarter mqttTopicsStarter,
ILogger<MqttTestClientManager> logger)
{
_mqttTopicsStarter = mqttTopicsStarter;
_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))
{
_workers[workflowInstanceId] = new List<Worker>();
}
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();
}
}
}

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

@ -1,4 +1,4 @@
namespace Elsa.Design
namespace Elsa.Design
{
public static class PropertyCategories
{

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

@ -19,6 +19,7 @@
<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.Mqtt\Elsa.Activities.Mqtt.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

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

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

@ -54,6 +54,9 @@
"RabbitMq": {
"Enabled": true,
"ConnectionStringIdentifier": "RabbitMq"
},
"Mqtt": {
"Enabled": true
}
},
"WorkflowChannels": {

20
src/samples/worker/Elsa.Samples.MqttWorker/Elsa.Samples.MqttWorker.csproj Normal file
View file

@ -0,0 +1,20 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net5.0</TargetFramework>
<Nullable>enable</Nullable>
<LangVersion>latest</LangVersion>
</PropertyGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Hosting" Version="6.0.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Mqtt\Elsa.Activities.Mqtt.csproj" />
<ProjectReference Include="..\..\..\activities\Elsa.Activities.Temporal.Quartz\Elsa.Activities.Temporal.Quartz.csproj" />
<ProjectReference Include="..\..\..\core\Elsa\Elsa.csproj" />
</ItemGroup>
</Project>

29
src/samples/worker/Elsa.Samples.MqttWorker/Program.cs Normal file
View file

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

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

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

41
src/samples/worker/Elsa.Samples.MqttWorker/Workflows/ConsumerWorkflow.cs Normal file
View file

@ -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<string>("Host");
_port = section.GetValue<int>("Port");
_username = section.GetValue<string>("Username");
_password = section.GetValue<string>("Password");
_qos = section.GetValue<MqttQualityOfService>("QualityOfService");
}
public void Build(IWorkflowBuilder builder)
{
builder
.Timer(Duration.FromSeconds(5))
.MessageReceived("/temperature", _host, _port, _username, _password, _qos)
.WriteLine(context =>
{
var message = context.GetInput<string>();
return $"Received a temperature update saying {message}!";
});
}
}
}

42
src/samples/worker/Elsa.Samples.MqttWorker/Workflows/ProducerWorkflow.cs Normal file
View file

@ -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<string>("Host");
_port = section.GetValue<int>("Port");
_username = section.GetValue<string>("Username");
_password = section.GetValue<string>("Password");
_qos = section.GetValue<MqttQualityOfService>("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";
}
}

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

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

17
src/samples/worker/Elsa.Samples.MqttWorker/appsettings.json Normal file
View file

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