From e7178ca28ecbb9b6b34bedbb29cac2304378615b Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Mon, 23 Nov 2020 16:46:42 +0100 Subject: [PATCH] Add Azure ServiceBus activities --- .../AzureServiceBusMessageReceived.cs | 7 ++- ...viceBusMessageReceivedBuilderExtensions.cs | 11 ++++ ...zureServiceBusMessageReceivedExtensions.cs | 23 +++++++ ...viceBusMessageReceivedBuilderExtensions.cs | 16 +++++ ...zureServiceBusMessageReceivedExtensions.cs | 22 +++++++ .../SendAzureServiceBusMessage.cs | 12 ++-- .../Elsa.Activities.AzureServiceBus.csproj | 23 ++++++- ...ivities.AzureServiceBus.csproj.DotSettings | 4 ++ .../Extensions/ManagementClientExtensions.cs | 17 ++++++ .../Extensions/ServiceCollectionExtensions.cs | 9 +-- ...BusFactory.cs => IMessageSenderFactory.cs} | 6 +- .../Services/MessageReceiverFactory.cs | 45 ++++++++++++++ .../Services/MessageSenderFactory.cs | 44 ++++++++++++++ .../Services/QueueWorker.cs | 56 ++++++++++++------ .../Services/ServiceBusFactory.cs | 54 ----------------- .../StartupTasks/StartServiceBusQueues.cs | 20 +++---- .../Triggers/MessageReceivedTrigger.cs | 1 - .../Elsa.Activities.AzureServiceBus/icon.png | Bin 0 -> 9825 bytes 18 files changed, 268 insertions(+), 102 deletions(-) rename src/activities/Elsa.Activities.AzureServiceBus/Activities/{ => AzureServiceBusMessageReceived}/AzureServiceBusMessageReceived.cs (75%) create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceivedBuilderExtensions.cs create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceivedExtensions.cs create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusMessageReceivedBuilderExtensions.cs create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusMessageReceivedExtensions.cs rename src/activities/Elsa.Activities.AzureServiceBus/Activities/{ => SendAzureServiceBusMessage}/SendAzureServiceBusMessage.cs (74%) create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Elsa.Activities.AzureServiceBus.csproj.DotSettings create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Extensions/ManagementClientExtensions.cs rename src/activities/Elsa.Activities.AzureServiceBus/Services/{IServiceBusFactory.cs => IMessageSenderFactory.cs} (78%) create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Services/MessageReceiverFactory.cs create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Services/MessageSenderFactory.cs delete mode 100644 src/activities/Elsa.Activities.AzureServiceBus/Services/ServiceBusFactory.cs create mode 100644 src/activities/Elsa.Activities.AzureServiceBus/icon.png diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceived.cs similarity index 75% rename from src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived.cs rename to src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceived.cs index 696426a3f..dd83385c1 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceived.cs @@ -7,7 +7,7 @@ using Elsa.Services; using Elsa.Services.Models; using Microsoft.Azure.ServiceBus; -namespace Elsa.Activities.AzureServiceBus.Activities +namespace Elsa.Activities.AzureServiceBus { [Trigger(Category = "Azure Service Bus", DisplayName = "Service Bus Message Received", Description = "Triggered when a message is received on the specified queue", Outcomes = new[] { OutcomeNames.Done })] public class AzureServiceBusMessageReceived : Activity @@ -22,9 +22,10 @@ namespace Elsa.Activities.AzureServiceBus.Activities [ActivityProperty] public string QueueName { get; set; } = default!; [ActivityProperty] public Type MessageType { get; set; } = default!; - protected override IActivityExecutionResult OnExecute() => Suspend(); + protected override IActivityExecutionResult OnExecute(ActivityExecutionContext context) => context.WorkflowExecutionContext.IsFirstPass ? ExecuteInternal(context) : Suspend(); + protected override IActivityExecutionResult OnResume(ActivityExecutionContext context) => ExecuteInternal(context); - protected override IActivityExecutionResult OnResume(ActivityExecutionContext context) + protected IActivityExecutionResult ExecuteInternal(ActivityExecutionContext context) { var message = (Message) context.Input!; var bytes = message.Body; diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceivedBuilderExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceivedBuilderExtensions.cs new file mode 100644 index 000000000..179865f9e --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceivedBuilderExtensions.cs @@ -0,0 +1,11 @@ +using System; +using Elsa.Builders; + +namespace Elsa.Activities.AzureServiceBus +{ + public static class AzureServiceBusMessageReceivedBuilderExtensions + { + public static IActivityBuilder MessageReceived(this IBuilder builder, Action> setup) => builder.Then(setup); + public static IActivityBuilder MessageReceived(this IBuilder builder, string queueName) => builder.MessageReceived(setup => setup.WithQueueName(queueName).WithMessageType()); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceivedExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceivedExtensions.cs new file mode 100644 index 000000000..f091195ce --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/AzureServiceBusMessageReceived/AzureServiceBusMessageReceivedExtensions.cs @@ -0,0 +1,23 @@ +using System; +using System.Threading.Tasks; +using Elsa.Builders; +using Elsa.Services.Models; + +// ReSharper disable once CheckNamespace +namespace Elsa.Activities.AzureServiceBus +{ + public static class AzureServiceBusMessageReceivedExtensions + { + public static ISetupActivity WithQueueName(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.QueueName, value!); + public static ISetupActivity WithQueueName(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.QueueName, value!); + public static ISetupActivity WithQueueName(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.QueueName, value!); + public static ISetupActivity WithQueueName(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.QueueName, value!); + public static ISetupActivity WithQueueName(this ISetupActivity messageReceived, string value) => messageReceived.Set(x => x.QueueName, value!); + + public static ISetupActivity WithMessageType(this ISetupActivity messageReceived, Func> value) => messageReceived.Set(x => x.MessageType, value!); + public static ISetupActivity WithMessageType(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.MessageType, value!); + public static ISetupActivity WithMessageType(this ISetupActivity messageReceived, Func value) => messageReceived.Set(x => x.MessageType, value!); + public static ISetupActivity WithMessageType(this ISetupActivity messageReceived, Type value) => messageReceived.Set(x => x.MessageType, value!); + public static ISetupActivity WithMessageType(this ISetupActivity messageReceived) => messageReceived.WithMessageType(typeof(T)); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusMessageReceivedBuilderExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusMessageReceivedBuilderExtensions.cs new file mode 100644 index 000000000..123ffdd76 --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusMessageReceivedBuilderExtensions.cs @@ -0,0 +1,16 @@ +using System; +using System.Threading.Tasks; +using Elsa.Builders; +using Elsa.Services.Models; + +namespace Elsa.Activities.AzureServiceBus +{ + public static class SendAzureServiceBusMessageBuilderExtensions + { + public static IActivityBuilder SendMessage(this IBuilder builder, Action> setup) => builder.Then(setup); + public static IActivityBuilder SendMessage(this IBuilder builder, string queueName, Func> message) => builder.SendMessage(setup => setup.WithQueueName(queueName).WithMessage(message)); + public static IActivityBuilder SendMessage(this IBuilder builder, string queueName, Func message) => builder.SendMessage(setup => setup.WithQueueName(queueName).WithMessage(message)); + public static IActivityBuilder SendMessage(this IBuilder builder, string queueName, Func message) => builder.SendMessage(setup => setup.WithQueueName(queueName).WithMessage(message)); + public static IActivityBuilder SendMessage(this IBuilder builder, string queueName, object message) => builder.SendMessage(setup => setup.WithQueueName(queueName).WithMessage(message)); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusMessageReceivedExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusMessageReceivedExtensions.cs new file mode 100644 index 000000000..49e336f76 --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/AzureServiceBusMessageReceivedExtensions.cs @@ -0,0 +1,22 @@ +using System; +using System.Threading.Tasks; +using Elsa.Builders; +using Elsa.Services.Models; + +// ReSharper disable once CheckNamespace +namespace Elsa.Activities.AzureServiceBus +{ + public static class SendAzureServiceBusMessageExtensions + { + public static ISetupActivity WithQueueName(this ISetupActivity activity, Func> value) => activity.Set(x => x.QueueName, value!); + public static ISetupActivity WithQueueName(this ISetupActivity activity, Func> value) => activity.Set(x => x.QueueName, value!); + public static ISetupActivity WithQueueName(this ISetupActivity activity, Func value) => activity.Set(x => x.QueueName, value!); + public static ISetupActivity WithQueueName(this ISetupActivity activity, Func value) => activity.Set(x => x.QueueName, value!); + public static ISetupActivity WithQueueName(this ISetupActivity activity, string value) => activity.Set(x => x.QueueName, value!); + + public static ISetupActivity WithMessage(this ISetupActivity activity, Func> value) => activity.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity activity, Func value) => activity.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity activity, Func value) => activity.Set(x => x.Message, value!); + public static ISetupActivity WithMessage(this ISetupActivity activity, object value) => activity.Set(x => x.Message, value!); + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage.cs b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusMessage.cs similarity index 74% rename from src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage.cs rename to src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusMessage.cs index 8fe9f0a7d..a1f5e8e22 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Activities/SendAzureServiceBusMessage/SendAzureServiceBusMessage.cs @@ -1,25 +1,25 @@ using System.Text; using System.Threading; using System.Threading.Tasks; +using Elsa.Activities.AzureServiceBus.Services; using Elsa.ActivityResults; using Elsa.Attributes; using Elsa.Serialization; using Elsa.Services; using Elsa.Services.Models; using Microsoft.Azure.ServiceBus; -using IServiceBusFactory = Elsa.Activities.AzureServiceBus.Services.IServiceBusFactory; -namespace Elsa.Activities.AzureServiceBus.Activities +namespace Elsa.Activities.AzureServiceBus { [Trigger(Category = "Azure Service Bus", DisplayName = "Send Service Bus Message", Description = "Sends a message to the specified queue", Outcomes = new[] { OutcomeNames.Done })] public class SendAzureServiceBusMessage : Activity { - private readonly IServiceBusFactory _serviceBusFactory; + private readonly IMessageSenderFactory _messageSenderFactory; private readonly IContentSerializer _serializer; - public SendAzureServiceBusMessage(IServiceBusFactory serviceBusFactory, IContentSerializer serializer) + public SendAzureServiceBusMessage(IMessageSenderFactory messageSenderFactory, IContentSerializer serializer) { - _serviceBusFactory = serviceBusFactory; + _messageSenderFactory = messageSenderFactory; _serializer = serializer; } @@ -28,7 +28,7 @@ namespace Elsa.Activities.AzureServiceBus.Activities protected override async ValueTask OnExecuteAsync(ActivityExecutionContext context, CancellationToken cancellationToken) { - var sender = await _serviceBusFactory.GetSenderAsync(QueueName, cancellationToken); + var sender = await _messageSenderFactory.GetSenderAsync(QueueName, cancellationToken); var json = _serializer.Serialize(Message); var bytes = Encoding.UTF8.GetBytes(json); var message = new Message(bytes); diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Elsa.Activities.AzureServiceBus.csproj b/src/activities/Elsa.Activities.AzureServiceBus/Elsa.Activities.AzureServiceBus.csproj index d064f2ff9..1e03d9a0d 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Elsa.Activities.AzureServiceBus.csproj +++ b/src/activities/Elsa.Activities.AzureServiceBus/Elsa.Activities.AzureServiceBus.csproj @@ -2,17 +2,34 @@ netstandard2.0 + 1.0.0 latest + Elsa Contributors + + 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 Azure Service Bus. + + 2020 + https://github.com/elsa-workflows/elsa-core + https://github.com/elsa-workflows/elsa-core + GitHub + elsa, workflows + icon.png enable - - + - + + + + True + + + diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Elsa.Activities.AzureServiceBus.csproj.DotSettings b/src/activities/Elsa.Activities.AzureServiceBus/Elsa.Activities.AzureServiceBus.csproj.DotSettings new file mode 100644 index 000000000..5433deafa --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Elsa.Activities.AzureServiceBus.csproj.DotSettings @@ -0,0 +1,4 @@ + + True + True + True \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ManagementClientExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ManagementClientExtensions.cs new file mode 100644 index 000000000..02f43aae2 --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ManagementClientExtensions.cs @@ -0,0 +1,17 @@ +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Azure.ServiceBus.Management; + +namespace Elsa.Activities.AzureServiceBus.Extensions +{ + public static class ManagementClientExtensions + { + public static async Task EnsureQueueExistsAsync(this ManagementClient managementClient, string queueName, CancellationToken cancellationToken) + { + if (await managementClient.QueueExistsAsync(queueName, cancellationToken)) + return; + + await managementClient.CreateQueueAsync(queueName, cancellationToken); + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs index c46d48006..86604cd56 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Extensions/ServiceCollectionExtensions.cs @@ -1,9 +1,8 @@ using System; -using Elsa.Activities.AzureServiceBus.Activities; using Elsa.Activities.AzureServiceBus.Options; using Elsa.Activities.AzureServiceBus.Services; using Elsa.Activities.AzureServiceBus.StartupTasks; -using Elsa.Runtime; +using Elsa.Activities.AzureServiceBus.Triggers; using Microsoft.Azure.ServiceBus; using Microsoft.Azure.ServiceBus.Management; using Microsoft.Extensions.DependencyInjection; @@ -23,8 +22,10 @@ namespace Elsa.Activities.AzureServiceBus.Extensions return services .AddSingleton(CreateServiceBusConnection) .AddSingleton(CreateServiceBusManagementClient) - .AddSingleton() - .AddStartupTask() + .AddSingleton() + .AddSingleton() + .AddHostedService() + .AddTriggerProvider() .AddActivity() .AddActivity(); } diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/IServiceBusFactory.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/IMessageSenderFactory.cs similarity index 78% rename from src/activities/Elsa.Activities.AzureServiceBus/Services/IServiceBusFactory.cs rename to src/activities/Elsa.Activities.AzureServiceBus/Services/IMessageSenderFactory.cs index 4d0e0dd55..c56792c64 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/IServiceBusFactory.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/IMessageSenderFactory.cs @@ -4,9 +4,13 @@ using Microsoft.Azure.ServiceBus.Core; namespace Elsa.Activities.AzureServiceBus.Services { - public interface IServiceBusFactory + public interface IMessageSenderFactory { Task GetSenderAsync(string queueName, CancellationToken cancellationToken = default); + } + + public interface IMessageReceiverFactory + { Task GetReceiverAsync(string queueName, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/MessageReceiverFactory.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/MessageReceiverFactory.cs new file mode 100644 index 000000000..4ef78a7c2 --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/MessageReceiverFactory.cs @@ -0,0 +1,45 @@ +using System; +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Elsa.Activities.AzureServiceBus.Extensions; +using Microsoft.Azure.ServiceBus; +using Microsoft.Azure.ServiceBus.Core; +using Microsoft.Azure.ServiceBus.Management; + +namespace Elsa.Activities.AzureServiceBus.Services +{ + public class MessageReceiverFactory : IMessageReceiverFactory + { + private readonly ServiceBusConnection _connection; + private readonly ManagementClient _managementClient; + private readonly IDictionary _receivers = new Dictionary(); + private readonly SemaphoreSlim _semaphore = new(1); + + public MessageReceiverFactory(ServiceBusConnection connection, ManagementClient managementClient) + { + _connection = connection; + _managementClient = managementClient; + } + + public async Task GetReceiverAsync(string queueName, CancellationToken cancellationToken) + { + if (_receivers.TryGetValue(queueName, out var messageReceiver)) + return messageReceiver; + + await _semaphore.WaitAsync(cancellationToken); + + try + { + await _managementClient.EnsureQueueExistsAsync(queueName, cancellationToken); + var newMessageReceiver = new MessageReceiver(_connection, queueName); + _receivers.Add(queueName, newMessageReceiver); + return newMessageReceiver; + } + finally + { + _semaphore.Release(); + } + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/MessageSenderFactory.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/MessageSenderFactory.cs new file mode 100644 index 000000000..ebfd37c49 --- /dev/null +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/MessageSenderFactory.cs @@ -0,0 +1,44 @@ +using System.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; +using Elsa.Activities.AzureServiceBus.Extensions; +using Microsoft.Azure.ServiceBus; +using Microsoft.Azure.ServiceBus.Core; +using Microsoft.Azure.ServiceBus.Management; + +namespace Elsa.Activities.AzureServiceBus.Services +{ + public class MessageSenderFactory : IMessageSenderFactory + { + private readonly ServiceBusConnection _connection; + private readonly ManagementClient _managementClient; + private readonly IDictionary _senders = new Dictionary(); + private readonly SemaphoreSlim _semaphore = new(1); + + public MessageSenderFactory(ServiceBusConnection connection, ManagementClient managementClient) + { + _connection = connection; + _managementClient = managementClient; + } + + public async Task GetSenderAsync(string queueName, CancellationToken cancellationToken) + { + if (_senders.TryGetValue(queueName, out var messageSender)) + return messageSender; + + await _semaphore.WaitAsync(cancellationToken); + + try + { + await _managementClient.EnsureQueueExistsAsync(queueName, cancellationToken); + var newMessageSender = new MessageSender(_connection, queueName); + _senders.Add(queueName, newMessageSender); + return newMessageSender; + } + finally + { + _semaphore.Release(); + } + } + } +} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs index 14089b2f9..5dec1fea1 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Services/QueueWorker.cs @@ -1,40 +1,58 @@ -using System.Threading; +using System; +using System.Threading; using System.Threading.Tasks; using Elsa.Activities.AzureServiceBus.Triggers; using Elsa.Services; +using Microsoft.Azure.ServiceBus; using Microsoft.Azure.ServiceBus.Core; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; namespace Elsa.Activities.AzureServiceBus.Services { public class QueueWorker { private readonly IMessageReceiver _messageReceiver; - private readonly IWorkflowScheduler _workflowScheduler; + private readonly IServiceProvider _serviceProvider; + private readonly ILogger _logger; - public QueueWorker(IMessageReceiver messageReceiver, IWorkflowScheduler workflowScheduler) + public QueueWorker(IMessageReceiver messageReceiver, IServiceProvider serviceProvider, ILogger logger) { _messageReceiver = messageReceiver; - _workflowScheduler = workflowScheduler; - } - - public async Task StartAsync(CancellationToken cancellationToken) - { - var cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); - await Task.Factory.StartNew(() => ReadQueueAsync(cancellationTokenSource.Token), cancellationToken); - } - - private async Task ReadQueueAsync(CancellationToken cancellationToken) - { - while (!cancellationToken.IsCancellationRequested) + _serviceProvider = serviceProvider; + _logger = logger; + + _messageReceiver.RegisterMessageHandler(OnMessageReceived, new MessageHandlerOptions(ExceptionReceivedHandler) { - var message = await _messageReceiver.ReceiveAsync(); + AutoComplete = false, + MaxConcurrentCalls = 10 + }); + } - if(message == null) - continue; + private async Task OnMessageReceived(Message message, CancellationToken cancellationToken) + { + using (var scope = _serviceProvider.CreateScope()) + { + var workflowScheduler = scope.ServiceProvider.GetRequiredService(); - await _workflowScheduler.TriggerWorkflowsAsync(x => x.QueueName == _messageReceiver.Path && (string.IsNullOrWhiteSpace(x.CorrelationId) || x.CorrelationId == message.CorrelationId), message, + await workflowScheduler.TriggerWorkflowsAsync(x => x.QueueName == _messageReceiver.Path && (string.IsNullOrWhiteSpace(x.CorrelationId) || x.CorrelationId == message.CorrelationId), message, message.CorrelationId, cancellationToken: cancellationToken); } + + await _messageReceiver.CompleteAsync(message.SystemProperties.LockToken); + } + + private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs e) + { + var context = e.ExceptionReceivedContext; + + _logger.LogError("Message handler encountered an exception {Exception}.", e.Exception); + _logger.LogError("Exception context for troubleshooting:"); + _logger.LogError("- Endpoint: {Endpoint}", context.Endpoint); + _logger.LogError("- Entity Path: {EntityPath}", context.EntityPath); + _logger.LogError("- Executing Action: {Action}", context.Action); + + return Task.CompletedTask; } } } \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Services/ServiceBusFactory.cs b/src/activities/Elsa.Activities.AzureServiceBus/Services/ServiceBusFactory.cs deleted file mode 100644 index bfec991c2..000000000 --- a/src/activities/Elsa.Activities.AzureServiceBus/Services/ServiceBusFactory.cs +++ /dev/null @@ -1,54 +0,0 @@ -using System.Collections.Generic; -using System.Threading; -using System.Threading.Tasks; -using Microsoft.Azure.ServiceBus; -using Microsoft.Azure.ServiceBus.Core; -using Microsoft.Azure.ServiceBus.Management; - -namespace Elsa.Activities.AzureServiceBus.Services -{ - public class ServiceBusFactory : IServiceBusFactory - { - private readonly ServiceBusConnection _connection; - private readonly ManagementClient _managementClient; - private readonly IDictionary _senders = new Dictionary(); - private readonly IDictionary _receivers = new Dictionary(); - - public ServiceBusFactory(ServiceBusConnection connection, ManagementClient managementClient) - { - _connection = connection; - _managementClient = managementClient; - } - - public async Task GetSenderAsync(string queueName, CancellationToken cancellationToken) - { - if (_senders.TryGetValue(queueName, out var messageSender)) - return messageSender; - - await EnsureQueueExistsAsync(queueName, cancellationToken); - - var newMessageSender = new MessageSender(_connection, queueName); - _senders.Add(queueName, newMessageSender); - return newMessageSender; - } - - public async Task GetReceiverAsync(string queueName, CancellationToken cancellationToken) - { - if (_receivers.TryGetValue(queueName, out var messageReceiver)) - return messageReceiver; - - await EnsureQueueExistsAsync(queueName, cancellationToken); - var newMessageReceiver = new MessageReceiver(_connection, queueName); - _receivers.Add(queueName, newMessageReceiver); - return newMessageReceiver; - } - - private async Task EnsureQueueExistsAsync(string queueName, CancellationToken cancellationToken) - { - if (await _managementClient.QueueExistsAsync(queueName, cancellationToken)) - return; - - await _managementClient.CreateQueueAsync(queueName, cancellationToken); - } - } -} \ No newline at end of file diff --git a/src/activities/Elsa.Activities.AzureServiceBus/StartupTasks/StartServiceBusQueues.cs b/src/activities/Elsa.Activities.AzureServiceBus/StartupTasks/StartServiceBusQueues.cs index 7f51ae4b0..a4a50a3a4 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/StartupTasks/StartServiceBusQueues.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/StartupTasks/StartServiceBusQueues.cs @@ -4,39 +4,37 @@ using System.Linq; using System.Runtime.CompilerServices; using System.Threading; using System.Threading.Tasks; -using Elsa.Activities.AzureServiceBus.Activities; using Elsa.Activities.AzureServiceBus.Services; using Elsa.Services; using Microsoft.Extensions.DependencyInjection; -using IServiceBusFactory = Elsa.Activities.AzureServiceBus.Services.IServiceBusFactory; +using Microsoft.Extensions.Hosting; namespace Elsa.Activities.AzureServiceBus.StartupTasks { - public class StartServiceBusQueues : IStartupTask + public class StartServiceBusQueues : BackgroundService { private readonly IWorkflowRegistry _workflowRegistry; private readonly IWorkflowBlueprintReflector _workflowBlueprintReflector; - private readonly IServiceBusFactory _serviceBusFactory; + private readonly IMessageReceiverFactory _messageReceiverFactory; private readonly IServiceProvider _serviceProvider; - public StartServiceBusQueues(IWorkflowRegistry workflowRegistry, IWorkflowBlueprintReflector workflowBlueprintReflector, IServiceBusFactory serviceBusFactory, IServiceProvider serviceProvider) + public StartServiceBusQueues(IWorkflowRegistry workflowRegistry, IWorkflowBlueprintReflector workflowBlueprintReflector, IMessageReceiverFactory messageReceiverFactory, IServiceProvider serviceProvider) { _workflowRegistry = workflowRegistry; _workflowBlueprintReflector = workflowBlueprintReflector; - _serviceBusFactory = serviceBusFactory; + _messageReceiverFactory = messageReceiverFactory; _serviceProvider = serviceProvider; } - public async Task ExecuteAsync(CancellationToken cancellationToken = default) + protected override async Task ExecuteAsync(CancellationToken stoppingToken) { + var cancellationToken = stoppingToken; var queueNames = await GetQueueNamesAsync(cancellationToken).ToListAsync(cancellationToken); foreach (var queueName in queueNames) { - var receiver = await _serviceBusFactory.GetReceiverAsync(queueName, cancellationToken); - var worker = ActivatorUtilities.CreateInstance(_serviceProvider, receiver); - - await worker.StartAsync(cancellationToken); + var receiver = await _messageReceiverFactory.GetReceiverAsync(queueName, cancellationToken); + ActivatorUtilities.CreateInstance(_serviceProvider, receiver); } } diff --git a/src/activities/Elsa.Activities.AzureServiceBus/Triggers/MessageReceivedTrigger.cs b/src/activities/Elsa.Activities.AzureServiceBus/Triggers/MessageReceivedTrigger.cs index 602cfe0c0..46d352288 100644 --- a/src/activities/Elsa.Activities.AzureServiceBus/Triggers/MessageReceivedTrigger.cs +++ b/src/activities/Elsa.Activities.AzureServiceBus/Triggers/MessageReceivedTrigger.cs @@ -1,6 +1,5 @@ using System.Threading; using System.Threading.Tasks; -using Elsa.Activities.AzureServiceBus.Activities; using Elsa.Triggers; namespace Elsa.Activities.AzureServiceBus.Triggers diff --git a/src/activities/Elsa.Activities.AzureServiceBus/icon.png b/src/activities/Elsa.Activities.AzureServiceBus/icon.png new file mode 100644 index 0000000000000000000000000000000000000000..28dbaafbfe82119161a9b48c7e1fb59fef019c2f GIT binary patch literal 9825 zcmaKSbyytF((Yivf&~i%S=`+TvdBVkhsAXX?(QKF2*HB~cNX^m0Rn*pcPF^J1cH;h z`F-b{@BVSm-RIfe>7MtkuIj3u?y2tg>Z9KM^5v< z7)~EY7X%vs#3X%O%q{I<9v}%M;o|1y;^pN)AUNE8ojlBaIGo(+ z{w+Zk=5Fa`>*8VS>;(EN(cHq>(?gsVq3Qpa;OO!nu}<#)>Lx^oarv0LaPe?*|265~ zKoyn$Kh)9jKWKLkP1yhH`~MW|uI=jrduo3rO?uK%}L3czGd0f^yva`~t$lGLV0BWt}ZO9brx$|K?i#FPG=Pa{tDH zqYI*CS(uxx7tBh|&Djz3kIkaC|5X=R$bXghFRs;p)kRkBzjC<{%5eSl?EmHIe^(*G z=kMiz;ucZ(Pvpa#5aI5Ii0hjdd369lJEb5irR}qDn1u<|>Am~0;@=X2+!DI(WU!s# z&uPw?*0R-Mn4Z|SotR>sa(BGK;VvYgA{(!*O*HHFgF}O`l1P?8N;YiJi|b0#{mXAU zRJ6_P%RArf=C`#&nZ7##CqtRu%i{^5E}rh*)~Wlt<$V=m^=?B^29@}HC`drIxdqAK zAx*uF_1cgV-`&S~VyAJ91i7N0(W=|XI*G;xyuWG;U@fy`K)AX2Z}j>NU2F?v63niF zj#txWsZ>J}R=R`_n80{TAA88WAft&M4RrKF5?kAa<#ulc)-5(rM-w(L=cN24Uw{Lu zTxwT8NIFB1EPne0sAC9E$E!zE{v0%$WX6+c*T>11=9^Cq^JLvTW-* zkCBhWz56&`B1pB!z!%gif*+L`G>;gAqIK&QIwTzcVD*vPnyQ7;c1y9Q!d{n&rw$p| zbn}~I7P2=G$$nwEqeK zn%@0@>6mMMi)f)V9}Yn%C)+&?7ZIU0*T;)qMQ5c&e+huuePXW{^eG+A@(E@FB8dU8 z)#vS@!Eua1Kc`ie21)=-OAhloHrJ)AAhY2DLc)gnW)SxR?$H`Z0kFk>jmAuGYD~@d zyet6VHT#Jz51ru6- zHW9`h!H^YgXmatu{+P9O<&hq^lDK!po_?>RgnN7g>&%+&bpo@cuP>EBk^*YO;LTlL70yosrZO(w}V%Z&!xmqLawKvGc}i zN}tsK_?e#~Ker55z^;GB&ef~xh2Bu)#&MAN&0ba3>o$1FzJF+b^r7=L<<$-P+>&G6g#X zBr_?toWHG{^xqwh_+qMsKV+dQk~MvraQe=?rwey+^dotd5fnBpgV`8d?q~@Yuc6$d zJmiZ;2LSzG@?GYl3TzFqa~)qvLg1>SB6*t{OXJSMMW$IG^A7p!?sDkzV<~ZYEfp== z*YfC&Lyx4N$DR|7lyb35$6AT0oj=Qm8f|(1NIPfYrdXwBxK-Gm*%6n<{b7gCR}1VN z#vc-ee0kwXz#zaPbgfHX0;<_DZ#05rs7QG z6>?^29&Sb>jn4ZUQ#dBEQzA0$bQ`1+VapitJpC zCoPz|?P`k}(^bQD(c#FzTuKj*XMgSVLDGWnr$bbw(d~C5hX(q%0(Rno_&OVi0TZ9J zxetj9Mei$9Om5LVV=)&cKi;0;iUc&` zPsv}V~7D1G*A|M&j0iWEZhujNswT0W<4(KtQW^u&;wYpG%Y z@U$@JlQtG{&365Sox+xbYf(!H=oFY(s46}?!2dR@TOR6%B25mAVUc7Vx!Wy5 zYw!h-LcE9ue?!0;;FMi{by{^21QbBU2BNF~{sLXd0DwUP0N@$`z^@QP{ZKmam^QI>zRxAfEr4!OJ;YpEaTZAv2l&G*FzM}IKWJyB~Ps|kW9a^EAVG!)fFI8 z?C7JG*pi>CP_qF)|=tWMq)DKc6ZQTURqc3}SjqD2@!ITjZD(Z^B-KpraURU!HXp0)3u> za|RMSynWlHx4F)<`L6*GpAqsmh*EF`b62bX4P);E<$EgR5ak*r#e$v$e=8lP_Dt3d zNjn;S<;Y=VFv*urh?ma9V5Kxt0ck-20OA?>r_q+`Rj~@=Q=fS6=q7H_@!I*>Tto%-pNwQ zSOH?SaC7{;e{MyMz;l`{FOf<2X0jmOa91DLv*kk{E6(RB#M;0dtNi~ZU(z1)8CKAEXjxlpZ?tscAbec%T zp@Og}P8c-ZmXhJ0N!!Vt*86_u><(nK@1_G*U-y)W$P7z&gvHwpoalT0Jg21SNMari zHNgVtR?d}(W=G|9t_t-ghh;+KFvQDDTE*E-y;1VNBZI&VJKN3H`Di zON^5RALa&78X~?G5Gj_mW@-j?mJHlYmQTB)GgjuW`YwhM+9Qo2tF}35Nf%7RJLKCl zoR3cmNc&5^S@F{FN_c}0h>_x|d7GCMYfaHgA5AXdoic)upX1)5OFsip3pjE`&(S48 zmcv_C9hd5aSLL%&mRyeWYK!wWY+E_P7jC)t8aj#Tnm48Ugw!fa%_dsjgK~y#5?Lx; zlg)M2u8%b8+x$!8J5Ly_9;F^od=g-BXbi z&gdt#p>Ilh@OB6Z$s{j}!Va#%;byR}R=ndT)i-M4_Pxw*%!WiU$c4Vl)Fqmsv$%@a z8`S(yP}yE|3%KN+Htu}s+C2Bt7o7SEMVX`e(f0nLW|X8(_Qm*feX~h{3Ql8Dt()^L z;zi+z4kO4*L$&iz3(~bashdF%2^Rv91 z?dTpBkq`MmxE>hk6?{!exy$oOla7i8ApUYY#V8m)RDitas=}yE=f!+$@)`Q@WsAvf zEcZh85+)YGY-Duy+$HJ&nbcyxk2eJGjzQ?O$u^8_IC{ITmkbwUmS_wV zSUUIO;)LalDq=mhGCWwmj0Rbv-bIGZQwfl@ku&@*VIBX*+;eg9S3Bz=CC~2B+miSv zs9R+47_ZqR#9ie~{Eq|{>U;?W?55Pk#!1UToCbHlkBdjL+QzA8tIG#mXTRWKe=id# z6;D2d#`O`L@HIr$aC$x>L1O1rQ|LXb0!cWHBAc9+q&*L&2)vV6t<51HNQy#2vZl(f zzEa;Vd>Y2a_D?4uJDCAlXLpOx&5)NCL z1jgpi=1lcGbezdS&dn`!ij~Q17Yw7Mf9;H_eG+_pWrA{R-`&p>9HymyHQ%t}&fA;y zxvTi8tRU}NbL4dI)s?@i|WVD zENNTY1)oT@z`Ktw9D5pB@WAQxzIiI-+@$fO_m>)-2Ohy*x!cF>_OqM~9{m<>F7ZvB ziXFUSHe+DFdQAo|O6A(x1(Be~(|UF3OvWex?VVt0!kL`XApX6XHlQO#Rs@rDLZR&1P$j z!uF+}J#1X^75(%r?7!pEE(KMZG$v*UIWtltB`ZOE%xuk8srIMqBdn?8H*%^AWwKYs zH|X-ZHB`Xlht86DeR4{>s^wc5#N`W(_Ty_ZQgzSR^f5_(^!L9{;V_i`K*~WxYtuN; zVk^nFmLXkxCk-;9UH=-kPell?87bF!A(>1m@rxXc)aI?w`nd|D%TI2lBXHMjEC_DU z2rs72#r_qr$4yxE8D+L`zO$5|(gVX^+B;o+$yyphP6DS5$DTwuYV2^HkHOSPFUibx z*{}Q=8s!`l`+$AculnZk1QIRg8%@-AQofPK9?zDX&%~GZ3mx;;`2$bHMZFHW8=W5B zR2;Az5cT8Y;_|zH=RaRrV;fv6o696`Pl;{HRNStSM>WZtWJ+U(-^pRwUy7$&e&k;w z2_SnHLF%}2ta5yOe14VnjC}grhVkvuEZxQYYm;t!T@}l^v9lm0=Fg*=6AXn54a|-0 z6`TI|iPltkHH;rv_NtkOv4uu!zUHt|eXZ1bm8jS&_AoLs!tX!xbfH-Y{L9{uVx#@J z=!_HNQud3G@S8)IU6Sx=(^ncnI|3_20hLjmKhJ+%=GDOlS?xV48opO)l?YObe$MH* zbK-6J{Mn}CSNO?G`5zf-1)CqBD+5p4(&({*y$QmD_aeTVT^uicP|XtM48z+aO1m{o z_{EluY2I^c?y-G=sq(|IXyZ zbuSU3UHFQicR3=TG%Sjkjq#~t3*EABNyqj}n4N>z!;Bw^_<39Ik;e2ev{fe&Pe8`#YPN8VJ$r#F2BL=*uTa&c5et(0`|4+N_@^a18)7iEO} z&dU_6dR3WC@`g%YS}qYc`#8}J_`N~(sZMqQ+P&d+eeqHrXVTm%*2e0;xK1j4I0nU;kuO>)-T=d9B8www0fqLyr-vqH#GOE<$kd+o<>;*9&B&RnW|-i zJ55``lO-xT7on}iXa{8XF|?9nmwjuX^W|1Z!$#z0U(4w1cx8VOTv%`X*)`MBg%LgK zgdIV)S{7HaX{%4{_h>SP!q9Q#43UNKADKRWP56g?Hq4cVZP@`$cZuVrmGx+JaVAB( zx+ee=BcrBE0|9UEH5v=KUGHO z`0Fv9(?pmlj2VO!@*Ry$JU}C8W@yjja^PfhV?(9g?~+KhN#|uA+tuW7f@HsiFHkDtcO<*&PX>&aj&oMAMc8Y{2o{<_1hB-8f-Wz=8&^PH{SE9zJ%_qifp%V7;&o! zzZ`dtkVHeHEZns6d}8t4s8xk@fjzes$uzzfR6SY~VTwNX?X8~9($4;Dxx8_!SFtMP z%Xq~6OC^n=d%Sw{d7;^vY3?JFn$75P;O0?#jB2L(iik(?HNEJh1ET;&CfJsfcbCML z1nZR7>8)LYZ&J1RyB^6I2a$9RUJkB_h;th=|DjaJ&YR^_$9 zXr6a~{KsD7R^{idQ~X$$MzV+*(Q$LTD~ocT8Sb33JeAw^(x=po&ryzWlj8CH^>pgS z-hE4b1MC;0-=BvRxSzdy)1pOPm_*@b#94ldy(&+_vQ7Sh%1*grh0hRCFx&8~{7n;2 zB(WTHqO@Ir^W9=rBnC~B{jU^`i)+`+^^>E_S*)|yZ{9Pu&N#eby?tVe?YP(!A!vy_k{DHK`BZ!4UhDEO z{bPgQG?vPV`P8$3`m-KP)A6DYJK=A&6jM3F<=&XF*tgELxHFmKZOr5dmDBo?_ZgLT z+-%#oy3nM3kEK)z$0n670M{g8n?^g&PiXZ;wYqM^X}62Y8QBrAW`7UIo-Q4g$D*?b zx-^L$7aNMdMTCZ7SobV56J|F>o*caN|J@MJj3VfRBS8spzEz- z6-lg!o^$%Q;4t*Aaj@f^XOVG7&{V~qzYRVHNgssD@#?{DulQJRyK>4y*YAI3$EpeJnS!8TUUESPu>Tb* z2ANvduerrV?0KGI>$>qBzMt*$JIdF6=W7c-d#>E~xVum>al_Jelk4|_2eXx3 z+s+7ZW&Nd}!DllnVlG3?v#VOf*rI8r2G@i{vFN~wVm=9HKfR$O^2f9M9^IYt0)u1> zE63{yLV3%|_l7cVmHMg|DohZH*3dI__HlQ(bzp({Xmu6L7IdBc^a%&$uWLEhRwY;Wy)H1y*%yf5t zBkAJUT2esQglO`hULjSW-#}~ZQmZFHs4Nm9W0b+~!LL(kpp)XcfqnF`Lr%gA(IvXq zwdMg*hOs`rmn9QLN?i&-DB7d^ry1%#&aX6zLVkR>Y7k-A4@6v z74t6!U2;PW&Z5889Udmb2jCmmObl5!)JWU1XdGV^t9-&@iGOp>{YiO%q~iM8$44V? zs*G(Ib3pj1P76IfJ+ooWvyX|Zknr>E7tZRB_csu(X1ioon}8y$l<`uBF5V~?l&^xP z%^sSv?Rr&4WHHt1iL%dg=d^2%S32>ySkt7#v7$_@yF$^%PO-L@#TO+{1c@smBQX$p zzZe(2u$`siP~sPNY3u9js|o=B*P|L27&ghy)e6_Qk3eZ=QW6ms4vx+nGvjw=NV#5B zChc;JML*14N3t>B2_dPAydI7EmAPnC6ZCK-q^P12E{~+n6M;*q8DVy|t5-Wk3VBP6RRAnlYgL9X?&o!oi^u80vZBhCz9Espsk$*(fC8wd^i;RVHOq^h49e#*t!)&4Dt}jI8ZI+KouQ> zNbSAZn|wW?knsDdwG=yPMTBND6$OQ|Oau<8!0A&U-hHOZP;D?6O5-U&<<=FIQuibP z$$3I(g;oL^NIxIsdTBdR*zZ=V$WL0~Qf6Q4btHOwK8T6W0a(4OKEWjyvr03vmRd8a zaT?5Hd?DzRPe3Q}BZCCs&3J-K&7-NRB-q?y_P~(3aP4qa#A)27;o-sWhYdXI;zRvK zHrn(4<7QLGJt=7gzR2|MPW}wU9%S<)!pzGh-NId2azC z;x%?Sm%?UCU)`gJ`0!;Gs~C7BKBVcELW)A$|V_}I# zE~AFTVBoaVJjH*9N3EciEtX&}@MqQF(!qYjwPl8vojv;Y;vvoNVwb_n9{@IS#0+;o zpP6dbZ)bblUt_lD>5VbZW6$u#Qi#hToK6~M|K|~%BEk3dbzutg?O!(X|$WB&r*%Z#sHI?_Fjl1`P;TtSI0l- zy}=;AJkdJd-wt|XY{zz{`g3WiOn>sK)TG@{yEDibu3h#Fk?Xmb&?~eo;mq%>jPY}! z#>qkd+2(CJcl$k6Lp4)afrdseDc~g{nMr$4NNw(loWu$hJ3x`N+nslnF{)5ZW`%4G zCR?QfZ*CrjWoLd=+d6%jpDN<~h|C{=`$|DIucwfH%?kdbHg3-g`aC=$ zqMSovyHnn&h}vAoGL>x?k!?@c$cm})xt{*>>{|biJjTKk07NI9{eC(ab#Oqg0S^Gn z-OGREC1ejvK>;?w1K4OEaN4h{HXLo3fRLwi*3*UT^3crie@Z^8AF>JAi)}jV2ZI_A zyPDF!4I3j(1iROjrTmcqdU`FK0qvT!x-9Eda03|tMnX&^WT{JWgaOp>{NXMiKxJe6LC1c;{Nc|! za|0a!MwJLofVO*tB|8;1zBLvtKtnKEIw$OHJhH zvKY9EHVB%f`{pHb zGpJmnTBk&8Jv}x8gy@BpU{HN~A~XLOPLcC-b$>fM$M{<6P0aW*5MtDgF3B;!%gV#l znN(h2oo3TAK56OAnzw`mh|nkL6Ef;*h&(_7ofPXyHv12`8pHt*`aO(NAF{QbdfCE} zO#Hlyvy#RD2%(77VJr9AjM9?`Oj5y9svo3L#zO+Q(w{U15ZC$@5$L{4NI@4jmPG+V zqFo$q&fylxOrbDh4hib5Gku9u8~{XVL@4IoW}QsX_643i{@mHk&Fjq4^n5mna-woHenpYO*l$Z8uc{rpsq9cRLU*@=`Za=i@hSlWJN^2fZL!_K~PcX8-` z?IpmEmIG^d#ICKI$6)(rQ_j@JPD77`3LZY)^ruT#cWj2ycr|rk7w|m7Kh_WW{ob-4 z+Ml!2MhD_iv)?|TTQ8+Svzk54^EspI$8|lu($&;ZRFQxIYrOhulccxC%M2v6F2h(OuM$RZEuweX}1jYA>#MSUo-b?{j4P%EB^0O zBex*nPj_=~5AQ~V*eQ##aQ))YV}t5d74I8f)-CHt(-0(}K(Ww}Ou3Nd@1ZXRh^lOr Iv{~r?0hzVtHvj+t literal 0 HcmV?d00001