diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusSendActivity.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusSendActivity.cs new file mode 100644 index 000000000..7f9e2d696 --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusSendActivity.cs @@ -0,0 +1,100 @@ +using Elsa.ActivityResults; +using Elsa.Attributes; +using Elsa.Design; +using Elsa.Expressions; +using Elsa.Serialization; +using Elsa.Services; +using Elsa.Services.Models; +using Microsoft.Azure.ServiceBus; +using Microsoft.Azure.ServiceBus.Core; +using System; +using System.Collections.Generic; +using System.Text; +using System.Threading.Tasks; + +namespace Elsa.Activities.AzureServiceBus +{ + public abstract class AzureServiceBusSendActivity : Activity + { + protected ISenderClient Sender; + + protected IContentSerializer _serializer; + + public AzureServiceBusSendActivity(IContentSerializer serializer) + { + _serializer = serializer; + } + + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid, SyntaxNames.Json })] + public object Message { get; set; } = default!; + + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string? CorrelationId { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string ContentType { get; set; } = default!; + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string? Label { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string? To { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string MessageId { get; set; } = default!; + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string? PartitionKey { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string? ViaPartitionKey { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string? ReplyTo { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string? SessionId { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public DateTime ExpiresAtUtc { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public TimeSpan TimeToLive { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public string? ReplyToSessionId { get; set; } + [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] + public DateTime ScheduledEnqueueTimeUtc { get; set; } + [ActivityInput(DefaultSyntax = SyntaxNames.Json, SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid,SyntaxNames.Json },UIHint = ActivityInputUIHints.MultiLine)] + public IDictionary UserProperties { get; set; } = new Dictionary(); + + protected override async ValueTask OnExecuteAsync(ActivityExecutionContext context) + { + var message = CreateMessage(Message); + + if (!string.IsNullOrWhiteSpace(context.WorkflowExecutionContext.CorrelationId)) + message.CorrelationId = context.WorkflowExecutionContext.CorrelationId; + + await Sender.SendAsync(message); + return Done(); + } + + protected Message CreateMessage(Object input) + { + var message = Extensions.MessageBodyExtensions.CreateMessage(_serializer, input); + + message.CorrelationId = CorrelationId; + message.ContentType = ContentType; + message.Label = Label; + message.To = To; + message.PartitionKey = PartitionKey; + message.ViaPartitionKey = ViaPartitionKey; + message.ReplyTo = ReplyTo; + message.SessionId = SessionId; + message.ReplyToSessionId = ReplyToSessionId; + + if (MessageId != null) + message.MessageId = MessageId; + if (TimeToLive != null && TimeToLive > TimeSpan.Zero) + message.TimeToLive = TimeToLive; + if (ScheduledEnqueueTimeUtc != null) + message.ScheduledEnqueueTimeUtc = ScheduledEnqueueTimeUtc; + + if (UserProperties != null) + foreach (var props in UserProperties) + message.UserProperties.Add(props.Key, props.Value); + + return message; + } + + } +} diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusQueueMessage.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusQueueMessage.cs index 342ce1214..2c8265f0d 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusQueueMessage.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusQueueMessage.cs @@ -1,4 +1,4 @@ -using System.Threading.Tasks; +using System.Threading.Tasks; using Elsa.Activities.AzureServiceBus.Services; using Elsa.ActivityResults; using Elsa.Attributes; @@ -10,33 +10,23 @@ using Elsa.Services.Models; namespace Elsa.Activities.AzureServiceBus { [Action(Category = "Azure Service Bus", DisplayName = "Send Service Bus Message", Description = "Sends a message to the specified queue", Outcomes = new[] { OutcomeNames.Done })] - public class SendAzureServiceBusQueueMessage : Activity + public class SendAzureServiceBusQueueMessage : AzureServiceBusSendActivity { private readonly IQueueMessageSenderFactory _queueMessageSenderFactory; - private readonly IContentSerializer _serializer; public SendAzureServiceBusQueueMessage(IQueueMessageSenderFactory queueMessageSenderFactory, IContentSerializer serializer) + :base(serializer) { _queueMessageSenderFactory = queueMessageSenderFactory; - _serializer = serializer; } [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid })] public string QueueName { get; set; } = default!; - [ActivityInput(SupportedSyntaxes = new[] { SyntaxNames.JavaScript, SyntaxNames.Liquid, SyntaxNames.Json })] - public object Message { get; set; } = default!; - protected override async ValueTask OnExecuteAsync(ActivityExecutionContext context) { - var sender = await _queueMessageSenderFactory.GetSenderAsync(QueueName, context.CancellationToken); - var message = Extensions.MessageBodyExtensions.CreateMessage(_serializer, Message); - - if (!string.IsNullOrWhiteSpace(context.WorkflowExecutionContext.CorrelationId)) - message.CorrelationId = context.WorkflowExecutionContext.CorrelationId; - - await sender.SendAsync(message); - return Done(); + Sender = await _queueMessageSenderFactory.GetSenderAsync(QueueName, context.CancellationToken); + return await base.OnExecuteAsync(context); } } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusTopicMessage.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusTopicMessage.cs index ec9b6d21e..f14da10dc 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusTopicMessage.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusTopicMessage.cs @@ -9,31 +9,23 @@ using Elsa.Services.Models; namespace Elsa.Activities.AzureServiceBus { [Action(Category = "Azure Service Bus", DisplayName = "Send Service Bus Topic Message", Description = "Sends a message to the specified topic", Outcomes = new[] { OutcomeNames.Done })] - public class SendAzureServiceBusTopicMessage : Activity + public class SendAzureServiceBusTopicMessage : AzureServiceBusSendActivity { private readonly ITopicMessageSenderFactory _messageSenderFactory; - private readonly IContentSerializer _serializer; public SendAzureServiceBusTopicMessage(ITopicMessageSenderFactory messageSenderFactory, IContentSerializer serializer) + :base(serializer) { - _messageSenderFactory = messageSenderFactory; - _serializer = serializer; + _messageSenderFactory = messageSenderFactory; } - [ActivityInput] public string TopicName { get; set; } = default!; - [ActivityInput] public object Message { get; set; } = default!; + [ActivityInput] public string TopicName { get; set; } = default!; protected override async ValueTask OnExecuteAsync(ActivityExecutionContext context) { - var sender = await _messageSenderFactory.GetTopicSenderAsync(TopicName, context.CancellationToken); + Sender = await _messageSenderFactory.GetTopicSenderAsync(TopicName, context.CancellationToken); - var message = Extensions.MessageBodyExtensions.CreateMessage(_serializer,Message); - - if (!string.IsNullOrWhiteSpace(context.WorkflowExecutionContext.CorrelationId)) - message.CorrelationId = context.WorkflowExecutionContext.CorrelationId; - - await sender.SendAsync(message); - return Done(); + return await base.OnExecuteAsync(context); } } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/MessageBodyExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/MessageBodyExtensions.cs index 24300f3a4..ae53eb4da 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/MessageBodyExtensions.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/MessageBodyExtensions.cs @@ -24,12 +24,7 @@ namespace Elsa.Activities.AzureServiceBus.Extensions { byte[] messageBytes; - if(message is MessageModel) - { - var messageModel = (MessageModel)message; - return CreateMessageFromMessageModel(messageModel); - } - else if (message is string s) + if (message is string s) messageBytes = Encoding.UTF8.GetBytes(s); else { @@ -39,35 +34,5 @@ namespace Elsa.Activities.AzureServiceBus.Extensions return new Message(messageBytes); } - - private static Message CreateMessageFromMessageModel(MessageModel message) - { - var returnMessage = new Message(message.Body) - { - CorrelationId = message.CorrelationId, - ContentType = message.ContentType, - Label = message.Label, - To = message.To, - PartitionKey = message.PartitionKey, - ViaPartitionKey = message.ViaPartitionKey, - ReplyTo = message.ReplyTo, - SessionId = message.SessionId, - ReplyToSessionId = message.ReplyToSessionId, - }; - - if(message.MessageId != null) - returnMessage.MessageId = message.MessageId; - if(message.TimeToLive != null && message.TimeToLive > TimeSpan.Zero ) - returnMessage.TimeToLive = message.TimeToLive; - if(message.ScheduledEnqueueTimeUtc != null) - returnMessage.ScheduledEnqueueTimeUtc = message.ScheduledEnqueueTimeUtc; - - if (message.UserProperties != null) - foreach (var props in message.UserProperties) - returnMessage.UserProperties.Add(props.Key, props.Value); - - return returnMessage; - } - } } \ No newline at end of file