From 43157fb6d6b59c100b417d98e2942010defa3533 Mon Sep 17 00:00:00 2001 From: Sipke Schoorstra Date: Sun, 23 Feb 2025 23:43:15 +0100 Subject: [PATCH] Remove Azure Service Bus integration and related tests Deleted the entire Azure Service Bus module, including activities, configurations, tests, and supporting code. This removal eliminates features like message handling workflows, queue/topic providers, and unit testing utilities for Azure Service Bus. --- Elsa.sln | 14 -- .../Elsa.Server.Web/Elsa.Server.Web.csproj | 1 - src/apps/Elsa.Server.Web/Program.cs | 11 +- .../Activities/MessageReceived.cs | 126 ------------------ .../Activities/SendMessage.cs | 104 --------------- .../Contracts/IQueueProvider.cs | 14 -- .../Contracts/IServiceBusInitializer.cs | 12 -- .../Contracts/ISubscriptionProvider.cs | 16 --- .../Contracts/ITopicProvider.cs | 14 -- .../Contracts/IWorkerManager.cs | 27 ---- .../Elsa.AzureServiceBus.csproj | 23 ---- .../Extensions/ModuleExtensions.cs | 25 ---- .../Features/AzureServiceBusFeature.cs | 94 ------------- .../Elsa.AzureServiceBus/FodyWeavers.xml | 3 - .../Handlers/UpdateWorkers.cs | 39 ------ .../CreateQueuesTopicsAndSubscriptions.cs | 24 ---- .../Models/MessageReceivedStimulus.cs | 47 ------- .../Models/QueueDefinition.cs | 6 - .../Models/ReceivedServiceBusMessageModel.cs | 36 ----- .../Models/SubscriptionDefinition.cs | 18 --- .../Models/TopicDefinition.cs | 17 --- .../Options/AzureServiceBusOptions.cs | 30 ----- ...rationQueueTopicAndSubscriptionProvider.cs | 28 ---- .../Services/ServiceBusInitializer.cs | 90 ------------- .../Elsa.AzureServiceBus/Services/Worker.cs | 122 ----------------- .../Services/WorkerManager.cs | 43 ------ .../Tasks/StartWorkers.cs | 48 ------- .../Abstractions/AppComponentTest.cs | 24 ---- .../AzureServiceBusTests.cs | 83 ------------ ...Elsa.AzureServiceBus.ComponentTests.csproj | 29 ---- ...rviceBus.ComponentTests.csproj.DotSettings | 4 - ...reServiceBusServiceCollectionExtensions.cs | 82 ------------ .../Fixtures/App.cs | 24 ---- .../Fixtures/Infrastructure.cs | 32 ----- .../Fixtures/ServiceBusAppCollection.cs | 9 -- .../Fixtures/WorkflowServer.cs | 98 -------------- .../Mocks/MockServiceBusProcessor.cs | 23 ---- .../MessageReceivedTriggerWorkflow.cs | 36 ----- ...SendOneMessageWithCorrelationIdWorkflow.cs | 34 ----- .../Workflows/SendOneMessageWorkflow.cs | 35 ----- 40 files changed, 1 insertion(+), 1544 deletions(-) delete mode 100644 src/modules/Elsa.AzureServiceBus/Activities/MessageReceived.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Activities/SendMessage.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Contracts/IQueueProvider.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Contracts/IServiceBusInitializer.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Contracts/ISubscriptionProvider.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Contracts/ITopicProvider.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Contracts/IWorkerManager.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Elsa.AzureServiceBus.csproj delete mode 100644 src/modules/Elsa.AzureServiceBus/Extensions/ModuleExtensions.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Features/AzureServiceBusFeature.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/FodyWeavers.xml delete mode 100644 src/modules/Elsa.AzureServiceBus/Handlers/UpdateWorkers.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/HostedServices/CreateQueuesTopicsAndSubscriptions.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Models/MessageReceivedStimulus.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Models/QueueDefinition.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Models/ReceivedServiceBusMessageModel.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Models/SubscriptionDefinition.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Models/TopicDefinition.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Options/AzureServiceBusOptions.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Providers/ConfigurationQueueTopicAndSubscriptionProvider.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Services/ServiceBusInitializer.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Services/Worker.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Services/WorkerManager.cs delete mode 100644 src/modules/Elsa.AzureServiceBus/Tasks/StartWorkers.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Abstractions/AppComponentTest.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/AzureServiceBusTests.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj.DotSettings delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Extensions/AzureServiceBusServiceCollectionExtensions.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/App.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/Infrastructure.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/ServiceBusAppCollection.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Mocks/MockServiceBusProcessor.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/MessageReceivedTriggerWorkflow.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/SendOneMessageWithCorrelationIdWorkflow.cs delete mode 100644 test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/SendOneMessageWorkflow.cs diff --git a/Elsa.sln b/Elsa.sln index d0ea54438..b15113b2f 100644 --- a/Elsa.sln +++ b/Elsa.sln @@ -33,8 +33,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Http", "src\modules\El EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Scheduling", "src\modules\Elsa.Scheduling\Elsa.Scheduling.csproj", "{ACD65CE5-3CC2-47B1-BFAC-72443D764F6E}" EndProject -Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.AzureServiceBus", "src\modules\Elsa.AzureServiceBus\Elsa.AzureServiceBus.csproj", "{24C7095B-5CDD-4369-9F34-5565A019B195}" -EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.JavaScript", "src\modules\Elsa.JavaScript\Elsa.JavaScript.csproj", "{D31581AB-A6C1-4B73-AB63-45667F6C82AE}" EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Hangfire", "src\modules\Elsa.Hangfire\Elsa.Hangfire.csproj", "{0601A2A6-2C62-418B-9104-8CDE497E5283}" @@ -246,8 +244,6 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.Activities.Integration EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.JavaScript.IntegrationTests", "test\integration\Elsa.JavaScript.IntegrationTests\Elsa.JavaScript.IntegrationTests.csproj", "{4332A6BC-434A-4AF5-A075-F1BBCDD28F5D}" EndProject -Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Elsa.AzureServiceBus.ComponentTests", "test\component\Elsa.AzureServiceBus.ComponentTests\Elsa.AzureServiceBus.ComponentTests.csproj", "{7F5A304F-7DD9-4A64-8FCD-9B1FF06735B9}" -EndProject Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "apps", "apps", "{D92BEAB2-60D6-4BB4-885A-6BA681C6CCF1}" ProjectSection(SolutionItems) = preProject src\apps\Directory.Build.props = src\apps\Directory.Build.props @@ -424,10 +420,6 @@ Global {ACD65CE5-3CC2-47B1-BFAC-72443D764F6E}.Debug|Any CPU.Build.0 = Debug|Any CPU {ACD65CE5-3CC2-47B1-BFAC-72443D764F6E}.Release|Any CPU.ActiveCfg = Release|Any CPU {ACD65CE5-3CC2-47B1-BFAC-72443D764F6E}.Release|Any CPU.Build.0 = Release|Any CPU - {24C7095B-5CDD-4369-9F34-5565A019B195}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {24C7095B-5CDD-4369-9F34-5565A019B195}.Debug|Any CPU.Build.0 = Debug|Any CPU - {24C7095B-5CDD-4369-9F34-5565A019B195}.Release|Any CPU.ActiveCfg = Release|Any CPU - {24C7095B-5CDD-4369-9F34-5565A019B195}.Release|Any CPU.Build.0 = Release|Any CPU {D31581AB-A6C1-4B73-AB63-45667F6C82AE}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {D31581AB-A6C1-4B73-AB63-45667F6C82AE}.Debug|Any CPU.Build.0 = Debug|Any CPU {D31581AB-A6C1-4B73-AB63-45667F6C82AE}.Release|Any CPU.ActiveCfg = Release|Any CPU @@ -706,10 +698,6 @@ Global {4332A6BC-434A-4AF5-A075-F1BBCDD28F5D}.Debug|Any CPU.Build.0 = Debug|Any CPU {4332A6BC-434A-4AF5-A075-F1BBCDD28F5D}.Release|Any CPU.ActiveCfg = Release|Any CPU {4332A6BC-434A-4AF5-A075-F1BBCDD28F5D}.Release|Any CPU.Build.0 = Release|Any CPU - {7F5A304F-7DD9-4A64-8FCD-9B1FF06735B9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {7F5A304F-7DD9-4A64-8FCD-9B1FF06735B9}.Debug|Any CPU.Build.0 = Debug|Any CPU - {7F5A304F-7DD9-4A64-8FCD-9B1FF06735B9}.Release|Any CPU.ActiveCfg = Release|Any CPU - {7F5A304F-7DD9-4A64-8FCD-9B1FF06735B9}.Release|Any CPU.Build.0 = Release|Any CPU {5ADDDFB1-E59B-4097-97B7-8C24E2D60463}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {5ADDDFB1-E59B-4097-97B7-8C24E2D60463}.Debug|Any CPU.Build.0 = Debug|Any CPU {5ADDDFB1-E59B-4097-97B7-8C24E2D60463}.Release|Any CPU.ActiveCfg = Release|Any CPU @@ -888,7 +876,6 @@ Global {7D5A49B4-9A9B-496E-803B-DEB85B2C3132} = {AB797AF0-C12C-46DE-A157-7E25625C6200} {82E26BF9-5F3A-4365-899A-AB1FFD54AA45} = {EBD0CF78-C5D7-4B4B-94C9-C5D8C20B6F59} {ACD65CE5-3CC2-47B1-BFAC-72443D764F6E} = {AB797AF0-C12C-46DE-A157-7E25625C6200} - {24C7095B-5CDD-4369-9F34-5565A019B195} = {DD089B8B-DA73-492A-9010-F772D1C178DA} {D31581AB-A6C1-4B73-AB63-45667F6C82AE} = {6EF07978-A6D2-40EB-891D-7D70C5F37E76} {0601A2A6-2C62-418B-9104-8CDE497E5283} = {AB797AF0-C12C-46DE-A157-7E25625C6200} {302BFC43-ED2F-43AE-8AD4-FCD481B0AC67} = {B08B4E00-C2AB-48F3-8389-449F42AEF179} @@ -975,7 +962,6 @@ Global {4B598AF7-BD7D-4544-A274-2CDDD98F4167} = {C6658DE0-2B2F-47F0-BB61-2CA66D435C09} {99B171E6-0248-4402-836D-98947CD63772} = {1B8D5897-902E-4632-8698-E89CAF3DDF54} {4332A6BC-434A-4AF5-A075-F1BBCDD28F5D} = {1B8D5897-902E-4632-8698-E89CAF3DDF54} - {7F5A304F-7DD9-4A64-8FCD-9B1FF06735B9} = {08B41FFA-CEE3-46A7-B5C0-3EB65D37A16C} {D92BEAB2-60D6-4BB4-885A-6BA681C6CCF1} = {61017E64-6D00-49CB-9E81-5002DC8F7D5F} {5ADDDFB1-E59B-4097-97B7-8C24E2D60463} = {D92BEAB2-60D6-4BB4-885A-6BA681C6CCF1} {97C7E531-9D5F-43FD-AA19-BF24DA13B612} = {5BA4A8FA-F7F4-45B3-AEC8-8886D35AAC79} diff --git a/src/apps/Elsa.Server.Web/Elsa.Server.Web.csproj b/src/apps/Elsa.Server.Web/Elsa.Server.Web.csproj index 033581426..6c3f0e76b 100644 --- a/src/apps/Elsa.Server.Web/Elsa.Server.Web.csproj +++ b/src/apps/Elsa.Server.Web/Elsa.Server.Web.csproj @@ -5,7 +5,6 @@ - diff --git a/src/apps/Elsa.Server.Web/Program.cs b/src/apps/Elsa.Server.Web/Program.cs index b0b5bbfea..7f0887c07 100644 --- a/src/apps/Elsa.Server.Web/Program.cs +++ b/src/apps/Elsa.Server.Web/Program.cs @@ -84,7 +84,6 @@ const bool useZipCompression = false; const bool runEFCoreMigrations = true; const bool useMemoryStores = false; const bool useCaching = true; -const bool useAzureServiceBus = false; const bool useKafka = false; const bool useReadOnlyMode = false; const bool useSignalR = false; // Disabled until Elsa Studio sends authenticated requests. @@ -544,15 +543,7 @@ services } }); } - - if (useAzureServiceBus) - { - elsa.UseAzureServiceBus(azureServiceBusConnectionString, asb => - { - asb.AzureServiceBusOptions = options => configuration.GetSection("AzureServiceBus").Bind(options); - }); - } - + if (useKafka) { elsa.UseKafka(kafka => diff --git a/src/modules/Elsa.AzureServiceBus/Activities/MessageReceived.cs b/src/modules/Elsa.AzureServiceBus/Activities/MessageReceived.cs deleted file mode 100644 index 48a9c9f18..000000000 --- a/src/modules/Elsa.AzureServiceBus/Activities/MessageReceived.cs +++ /dev/null @@ -1,126 +0,0 @@ -using System.Runtime.CompilerServices; -using Elsa.AzureServiceBus.Models; -using Elsa.Common; -using Elsa.Expressions.Models; -using Elsa.Extensions; -using Elsa.Workflows; -using Elsa.Workflows.Attributes; -using Elsa.Workflows.Models; - -namespace Elsa.AzureServiceBus.Activities; - -/// -/// Triggered when a message is received on a specified queue or topic and subscription. -/// -[Activity("Elsa.AzureServiceBus", "Azure Service Bus", "Executes when a message is received from the configured queue or topic and subscription")] -public class MessageReceived : Trigger -{ - internal const string InputKey = "TransportMessage"; - - /// - public MessageReceived([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) - { - } - - /// - public MessageReceived(Input queue) - { - QueueOrTopic = queue; - } - - /// - public MessageReceived(string queue) : this(new Input(queue)) - { - } - - /// - public MessageReceived(Input topic, Input subscription) - { - QueueOrTopic = topic; - Subscription = subscription; - } - - /// - public MessageReceived(string topic, string subscription) : this(new Input(topic), new Input(subscription)) - { - } - - /// - /// The name of the queue or topic to read from. - /// - [Input(Description = "The name of the queue or topic to read from.")] - public Input QueueOrTopic { get; set; } = default!; - - /// - /// The name of the subscription to read from. - /// - [Input(Description = "The name of the subscription to read from.")] - public Input? Subscription { get; set; } - - /// - /// The .NET type to deserialize the message into. Defaults to . - /// - [Input(Description = "The .NET type to deserialize the message into.")] - public Input MessageType { get; set; } = new(typeof(string)); - - /// - /// The received transport message. - /// - [Output(Description = "The received transport message.")] - public Output TransportMessage { get; set; } = default!; - - /// - /// The received transport message. - /// - [Output(Description = "The received message.")] - public Output Message { get; set; } = default!; - - /// - /// The formatter to use to parse the message. - /// - [Input(Description = "The formatter to use to serialize the message.")] - public Input Formatter { get; set; } = default!; - - /// - protected override object GetTriggerPayload(TriggerIndexingContext context) => GetStimulus(context.ExpressionExecutionContext); - - /// - protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) - { - // If we did not receive external input, it means we are just now encountering this activity. - if (context.IsTriggerOfWorkflow()) - { - await Resume(context); - } - else - { - // Create bookmarks for when we receive the expected HTTP request. - context.CreateBookmark(GetStimulus(context.ExpressionExecutionContext), Resume,false); - } - } - - private async ValueTask Resume(ActivityExecutionContext context) - { - var receivedMessage = context.GetWorkflowInput(InputKey); - await SetResultAsync(receivedMessage, context); - await context.CompleteActivityAsync(); - } - - private async Task SetResultAsync(ReceivedServiceBusMessageModel receivedMessage, ActivityExecutionContext context) - { - var bodyAsString = new BinaryData(receivedMessage.Body).ToString(); - var targetType = context.Get(MessageType); - var formatter = Formatter.GetOrDefault(context); - var body = formatter == null ? bodyAsString : await formatter.FromStringAsync(bodyAsString, targetType, context.CancellationToken); - - context.Set(TransportMessage, receivedMessage); - context.Set(Message, body); - } - - private object GetStimulus(ExpressionExecutionContext context) - { - var queueOrTopic = context.Get(QueueOrTopic)!; - var subscription = context.Get(Subscription); - return new MessageReceivedStimulus(queueOrTopic, subscription); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Activities/SendMessage.cs b/src/modules/Elsa.AzureServiceBus/Activities/SendMessage.cs deleted file mode 100644 index d9f94204a..000000000 --- a/src/modules/Elsa.AzureServiceBus/Activities/SendMessage.cs +++ /dev/null @@ -1,104 +0,0 @@ -using System.Runtime.CompilerServices; -using System.Text.Json; -using Azure.Messaging.ServiceBus; -using Elsa.Common; -using Elsa.Common.Services; -using Elsa.Extensions; -using Elsa.Workflows; -using Elsa.Workflows.Attributes; -using Elsa.Workflows.Models; -using Elsa.Workflows.UIHints; -using JetBrains.Annotations; - -namespace Elsa.AzureServiceBus.Activities; - -/// -/// Sends a message to a queue or topic in Azure Service Bus. -/// -[Activity("Elsa.AzureServiceBus.Send", "Azure Service Bus", "Send a message to a queue or topic")] -[PublicAPI] -public class SendMessage : CodeActivity -{ - /// - public SendMessage([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line) - { - } - - /// - /// The contents of the message to send. - /// - [Input(Description = "The contents of the message to send.")] - public Input MessageBody { get; set; } = default!; - - /// - /// The queue or topic to send the message to. - /// - public Input QueueOrTopic { get; set; } = default!; - - /// - /// The content type of the message. - /// - public Input? ContentType { get; set; } - - /// - /// The subject of the message. - /// - public Input? Subject { get; set; } - - /// - /// The correlation ID of the message. - /// - public Input? CorrelationId { get; set; } - - /// - /// The formatter to use when serializing the message body. - /// - public Input FormatterType { get; set; } = default!; - - /// - /// The application properties to embed with the Service Bus Message - /// - [Input(Category = "Advanced", - DefaultSyntax = "Json", - SupportedSyntaxes = ["JavaScript", "Json"], - UIHint = InputUIHints.MultiLine) - ] - public Input?> ApplicationProperties { get; set; } = default!; - - /// - protected override async ValueTask ExecuteAsync(ActivityExecutionContext context) - { - var queueOrTopic = context.Get(QueueOrTopic); - var messageBody = context.Get(MessageBody); - var cancellationToken = context.CancellationToken; - var serializedMessageBody = await SerializeMessageBodyAsync(context, messageBody!, cancellationToken); - - var message = new ServiceBusMessage(serializedMessageBody) - { - ContentType = context.Get(ContentType), - Subject = context.Get(Subject), - CorrelationId = context.Get(CorrelationId) - }; - - var applicationProperties = ApplicationProperties.GetOrDefault(context); - - if (applicationProperties != null) - foreach (var property in applicationProperties) - message.ApplicationProperties.Add(property.Key, ((JsonElement)property.Value).GetString()); - - var client = context.GetRequiredService(); - await using var sender = client.CreateSender(queueOrTopic); - await sender.SendMessageAsync(message, cancellationToken); - } - - private async ValueTask SerializeMessageBodyAsync(ActivityExecutionContext context, object value, CancellationToken cancellationToken) - { - if (value is string s) return BinaryData.FromString(s); - - var formatterType = FormatterType.GetOrDefault(context) ?? typeof(JsonFormatter); - var formatter = context.GetServices().First(x => x.GetType() == formatterType); - var data = await formatter.ToStringAsync(value, cancellationToken); - - return BinaryData.FromString(data); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Contracts/IQueueProvider.cs b/src/modules/Elsa.AzureServiceBus/Contracts/IQueueProvider.cs deleted file mode 100644 index 9171e9d01..000000000 --- a/src/modules/Elsa.AzureServiceBus/Contracts/IQueueProvider.cs +++ /dev/null @@ -1,14 +0,0 @@ -using Elsa.AzureServiceBus.Models; - -namespace Elsa.AzureServiceBus.Contracts; - -/// -/// Provides queue definitions to the system. -/// -public interface IQueueProvider -{ - /// - /// Returns a list of s. - /// - ValueTask> GetQueuesAsync(CancellationToken cancellationToken); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Contracts/IServiceBusInitializer.cs b/src/modules/Elsa.AzureServiceBus/Contracts/IServiceBusInitializer.cs deleted file mode 100644 index dfecf946e..000000000 --- a/src/modules/Elsa.AzureServiceBus/Contracts/IServiceBusInitializer.cs +++ /dev/null @@ -1,12 +0,0 @@ -namespace Elsa.AzureServiceBus.Contracts; - -/// -/// Creates queues, topics and subscriptions provided by , and implementations. -/// -public interface IServiceBusInitializer -{ - /// - /// Creates queues, topics and subscriptions provided by , and implementations. - /// - Task InitializeAsync(CancellationToken cancellationToken = default); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Contracts/ISubscriptionProvider.cs b/src/modules/Elsa.AzureServiceBus/Contracts/ISubscriptionProvider.cs deleted file mode 100644 index 167f7f3fd..000000000 --- a/src/modules/Elsa.AzureServiceBus/Contracts/ISubscriptionProvider.cs +++ /dev/null @@ -1,16 +0,0 @@ -using Elsa.AzureServiceBus.Models; - -namespace Elsa.AzureServiceBus.Contracts; - -/// -/// Provides subscription definitions to the system. -/// -[Obsolete("Use AzureServiceBusOptions.Topics instead.")] - -public interface ISubscriptionProvider -{ - /// - /// Return a list of s. - /// - ValueTask> GetSubscriptionsAsync(CancellationToken cancellationToken); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Contracts/ITopicProvider.cs b/src/modules/Elsa.AzureServiceBus/Contracts/ITopicProvider.cs deleted file mode 100644 index d1ac1d575..000000000 --- a/src/modules/Elsa.AzureServiceBus/Contracts/ITopicProvider.cs +++ /dev/null @@ -1,14 +0,0 @@ -using Elsa.AzureServiceBus.Models; - -namespace Elsa.AzureServiceBus.Contracts; - -/// -/// Provides topic definitions to the system. -/// -public interface ITopicProvider -{ - /// - /// Returns a list of s. - /// - ValueTask> GetTopicsAsync(CancellationToken cancellationToken); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Contracts/IWorkerManager.cs b/src/modules/Elsa.AzureServiceBus/Contracts/IWorkerManager.cs deleted file mode 100644 index 989f65fa5..000000000 --- a/src/modules/Elsa.AzureServiceBus/Contracts/IWorkerManager.cs +++ /dev/null @@ -1,27 +0,0 @@ -using Elsa.AzureServiceBus.Services; - -namespace Elsa.AzureServiceBus.Contracts; - -/// -/// Manages message workers. -/// -public interface IWorkerManager -{ - /// - /// A list of workers under management. - /// - IEnumerable Workers { get; } - - /// - /// Ensures that at least one worker exists for the specified queue/topic and subscription. - /// - Task StartWorkerAsync(string queueOrTopic, string? subscription, CancellationToken cancellationToken = default); - - /// - /// Finds a worker for the specified queue or topic and subscription. - /// - /// The name of the queue or topic. - /// The name of the subscription. - /// The worker, or null if no worker was found. - Worker? FindWorkerFor(string queueOrTopic, string? subscription); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Elsa.AzureServiceBus.csproj b/src/modules/Elsa.AzureServiceBus/Elsa.AzureServiceBus.csproj deleted file mode 100644 index 931768726..000000000 --- a/src/modules/Elsa.AzureServiceBus/Elsa.AzureServiceBus.csproj +++ /dev/null @@ -1,23 +0,0 @@ - - - - - Provides Azure Service Bus integration and activities. - - elsa module azure-service-bus service-bus - - - - - - - - - - - - - - - - diff --git a/src/modules/Elsa.AzureServiceBus/Extensions/ModuleExtensions.cs b/src/modules/Elsa.AzureServiceBus/Extensions/ModuleExtensions.cs deleted file mode 100644 index 4643e121d..000000000 --- a/src/modules/Elsa.AzureServiceBus/Extensions/ModuleExtensions.cs +++ /dev/null @@ -1,25 +0,0 @@ -using Elsa.AzureServiceBus.Features; -using Elsa.Features.Services; - -// ReSharper disable once CheckNamespace -namespace Elsa.Extensions; - -/// -/// Adds extension methods to to register Azure Service Bus related services. -/// -public static class ModuleExtensions -{ - /// - /// Enable and configure the feature. - /// - public static IModule UseAzureServiceBus(this IModule module, string connectionStringOrName, Action? setup = null) - { - setup += feature => feature.AzureServiceBusOptions += options => options.ConnectionStringOrName = connectionStringOrName; - return module.Use(setup); - } - - /// - /// Enable and configure the feature. - /// - public static IModule UseAzureServiceBus(this IModule module, Action? setup = null) => module.Use(setup); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Features/AzureServiceBusFeature.cs b/src/modules/Elsa.AzureServiceBus/Features/AzureServiceBusFeature.cs deleted file mode 100644 index 27a6cc52d..000000000 --- a/src/modules/Elsa.AzureServiceBus/Features/AzureServiceBusFeature.cs +++ /dev/null @@ -1,94 +0,0 @@ -using Azure.Messaging.ServiceBus; -using Azure.Messaging.ServiceBus.Administration; -using Elsa.AzureServiceBus.Contracts; -using Elsa.AzureServiceBus.Handlers; -using Elsa.AzureServiceBus.HostedServices; -using Elsa.AzureServiceBus.Options; -using Elsa.AzureServiceBus.Providers; -using Elsa.AzureServiceBus.Services; -using Elsa.AzureServiceBus.Tasks; -using Elsa.Extensions; -using Elsa.Features.Abstractions; -using Elsa.Features.Services; -using Microsoft.Extensions.Configuration; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Options; - -namespace Elsa.AzureServiceBus.Features; - -/// -/// Enables and configures the Azure Service Bus feature. -/// -public class AzureServiceBusFeature : FeatureBase -{ - /// - public AzureServiceBusFeature(IModule module) : base(module) - { - } - - /// - /// A value controlling whether queues, topics and subscriptions should be created automatically. - /// - public bool CreateQueuesTopicsAndSubscriptions { get; set; } = true; - - /// - /// A delegate to configure . - /// - public Action AzureServiceBusOptions { get; set; } = _ => { }; - - /// - /// A delegate to create a instance. - /// - public Func ServiceBusClientFactory { get; set; } = sp => new(GetConnectionString(sp)); - - /// - /// A delegate to create a instance. - /// - public Func ServiceBusAdministrationClientFactory { get; set; } = sp => new(GetConnectionString(sp)); - - /// - public override void ConfigureHostedServices() - { - if (CreateQueuesTopicsAndSubscriptions) - Module.ConfigureHostedService(); - } - - /// - public override void Configure() - { - // Activities. - Module.AddActivitiesFrom(); - } - - /// - public override void Apply() - { - Services.Configure(AzureServiceBusOptions); - - Services - .AddSingleton(ServiceBusAdministrationClientFactory) - .AddSingleton(ServiceBusClientFactory) - .AddSingleton() - .AddSingleton() - .AddScoped(); - - // Tasks. - Services.AddBackgroundTask(); - - // Definition providers. - Services - .AddSingleton(sp => sp.GetRequiredService()) - .AddSingleton(sp => sp.GetRequiredService()) - .AddSingleton(sp => sp.GetRequiredService()); - - // Handlers. - Services.AddHandlersFrom(); - } - - private static string GetConnectionString(IServiceProvider serviceProvider) - { - var options = serviceProvider.GetRequiredService>().Value; - var configuration = serviceProvider.GetRequiredService(); - return configuration.GetConnectionString(options.ConnectionStringOrName) ?? options.ConnectionStringOrName; - } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/FodyWeavers.xml b/src/modules/Elsa.AzureServiceBus/FodyWeavers.xml deleted file mode 100644 index 00e1d9a1c..000000000 --- a/src/modules/Elsa.AzureServiceBus/FodyWeavers.xml +++ /dev/null @@ -1,3 +0,0 @@ - - - \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Handlers/UpdateWorkers.cs b/src/modules/Elsa.AzureServiceBus/Handlers/UpdateWorkers.cs deleted file mode 100644 index c3f43fe7c..000000000 --- a/src/modules/Elsa.AzureServiceBus/Handlers/UpdateWorkers.cs +++ /dev/null @@ -1,39 +0,0 @@ -using Elsa.AzureServiceBus.Activities; -using Elsa.AzureServiceBus.Contracts; -using Elsa.AzureServiceBus.Models; -using Elsa.Extensions; -using Elsa.Mediator.Contracts; -using Elsa.Workflows.Runtime.Notifications; -using JetBrains.Annotations; - -namespace Elsa.AzureServiceBus.Handlers; - -/// -/// Creates workers for each trigger & bookmark in response to updated workflow trigger indexes and bookmarks. -/// -[UsedImplicitly] -public class UpdateWorkers(IWorkerManager workerManager) : INotificationHandler, INotificationHandler -{ - /// - /// Adds, updates and removes workers based on added and removed triggers. - /// - public async Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken) - { - var added = notification.IndexedWorkflowTriggers.AddedTriggers.Filter().Select(x => x.GetPayload()); - await StartWorkersAsync(added, cancellationToken); - } - - /// - /// Adds, updates and removes workers based on added and removed bookmarks. - /// - public async Task HandleAsync(WorkflowBookmarksIndexed notification, CancellationToken cancellationToken) - { - var added = notification.IndexedWorkflowBookmarks.AddedBookmarks.Filter().Select(x => x.GetPayload()); - await StartWorkersAsync(added, cancellationToken); - } - - private async Task StartWorkersAsync(IEnumerable payloads, CancellationToken cancellationToken) - { - foreach (var payload in payloads) await workerManager.StartWorkerAsync(payload.QueueOrTopic, payload.Subscription, cancellationToken); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/HostedServices/CreateQueuesTopicsAndSubscriptions.cs b/src/modules/Elsa.AzureServiceBus/HostedServices/CreateQueuesTopicsAndSubscriptions.cs deleted file mode 100644 index 3e488421e..000000000 --- a/src/modules/Elsa.AzureServiceBus/HostedServices/CreateQueuesTopicsAndSubscriptions.cs +++ /dev/null @@ -1,24 +0,0 @@ -using Elsa.AzureServiceBus.Contracts; -using JetBrains.Annotations; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Hosting; - -namespace Elsa.AzureServiceBus.HostedServices; - -/// -/// A blocking hosted service that creates queues, topics and subscriptions. -/// -[UsedImplicitly] -public class CreateQueuesTopicsAndSubscriptions(IServiceScopeFactory scopeFactory) : IHostedService -{ - /// - public async Task StartAsync(CancellationToken cancellationToken) - { - await using var scope = scopeFactory.CreateAsyncScope(); - var initializer = scope.ServiceProvider.GetRequiredService(); - await initializer.InitializeAsync(cancellationToken); - } - - /// - public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask; -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Models/MessageReceivedStimulus.cs b/src/modules/Elsa.AzureServiceBus/Models/MessageReceivedStimulus.cs deleted file mode 100644 index 3ff8f1507..000000000 --- a/src/modules/Elsa.AzureServiceBus/Models/MessageReceivedStimulus.cs +++ /dev/null @@ -1,47 +0,0 @@ -using System.Text.Json.Serialization; - -namespace Elsa.AzureServiceBus.Models; - -/// -/// A bookmark payload model for triggering workflows when messages come in at a given queue or topic and subscription. -/// -public record MessageReceivedStimulus -{ - private readonly string _queueOrTopic = default!; - private readonly string? _subscription; - - /// - /// Constructor. - /// - [JsonConstructor] - public MessageReceivedStimulus() - { - } - - /// - /// Constructor. - /// - public MessageReceivedStimulus(string queueOrTopic, string? subscription) - { - QueueOrTopic = queueOrTopic; - Subscription = subscription; - } - - /// - /// The queue or topic to trigger from. - /// - public string QueueOrTopic - { - get => _queueOrTopic; - init => _queueOrTopic = value.ToLowerInvariant(); - } - - /// - /// The subscription to trigger from. - /// - public string? Subscription - { - get => _subscription; - init => _subscription = value?.ToLowerInvariant(); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Models/QueueDefinition.cs b/src/modules/Elsa.AzureServiceBus/Models/QueueDefinition.cs deleted file mode 100644 index 4c85efdc3..000000000 --- a/src/modules/Elsa.AzureServiceBus/Models/QueueDefinition.cs +++ /dev/null @@ -1,6 +0,0 @@ -namespace Elsa.AzureServiceBus.Models; - -/// -/// Represents a queue that is available to the system. -/// -public record QueueDefinition(string Name); \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Models/ReceivedServiceBusMessageModel.cs b/src/modules/Elsa.AzureServiceBus/Models/ReceivedServiceBusMessageModel.cs deleted file mode 100644 index 5b3889653..000000000 --- a/src/modules/Elsa.AzureServiceBus/Models/ReceivedServiceBusMessageModel.cs +++ /dev/null @@ -1,36 +0,0 @@ -using JetBrains.Annotations; - -namespace Elsa.AzureServiceBus.Models; - -/// -/// A serializable version of . -/// -// Needs to be a class and not a record, because of the polymorphic serialization that cannot deal with $type properties. -[PublicAPI] -public class ReceivedServiceBusMessageModel -{ - public byte[] Body { get; init; } = default!; - public string? Subject { get; init; } - public string? ContentType { get; init; } - public string? To { get; init; } - public string? CorrelationId { get; init; } - public int DeliveryCount { get; init; } - public DateTimeOffset EnqueuedTime { get; init; } - public DateTimeOffset ScheduledEnqueuedTime { get; init; } - public DateTimeOffset ExpiresAt { get; init; } - public DateTimeOffset LockedUntil { get; init; } - public TimeSpan TimeToLive { get; init; } - public string? LockToken { get; init; } - public string? MessageId { get; init; } - public string? PartitionKey { get; init; } - public string? TransactionPartitionKey { get; init; } - public string? ReplyTo { get; init; } - public long SequenceNumber { get; init; } - public long EnqueuedSequenceNumber { get; init; } - public string? SessionId { get; init; } - public string? ReplyToSessionId { get; init; } - public string? DeadLetterReason { get; init; } - public string? DeadLetterSource { get; init; } - public string? DeadLetterErrorDescription { get; init; } - public IReadOnlyDictionary ApplicationProperties { get; init; } = new Dictionary(); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Models/SubscriptionDefinition.cs b/src/modules/Elsa.AzureServiceBus/Models/SubscriptionDefinition.cs deleted file mode 100644 index 5ddeb426d..000000000 --- a/src/modules/Elsa.AzureServiceBus/Models/SubscriptionDefinition.cs +++ /dev/null @@ -1,18 +0,0 @@ -namespace Elsa.AzureServiceBus.Models; - -/// -/// Represents a topic subscription that is available to the system. -/// -public class SubscriptionDefinition -{ - /// - /// The subscription name. - /// - public string Name { get; set; } = default!; - - /// - /// The topic. - /// - [Obsolete("Use TopicDefinition.Subscriptions instead.")] - public string? Topic { get; set; } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Models/TopicDefinition.cs b/src/modules/Elsa.AzureServiceBus/Models/TopicDefinition.cs deleted file mode 100644 index 65c91b580..000000000 --- a/src/modules/Elsa.AzureServiceBus/Models/TopicDefinition.cs +++ /dev/null @@ -1,17 +0,0 @@ -namespace Elsa.AzureServiceBus.Models; - -/// -/// Represents a topic that is available to the system. -/// -public class TopicDefinition -{ - /// - /// The topic name. - /// - public string Name { get; set; } = default!; - - /// - /// The subscriptions. - /// - public ICollection Subscriptions { get; set; } = new List(); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Options/AzureServiceBusOptions.cs b/src/modules/Elsa.AzureServiceBus/Options/AzureServiceBusOptions.cs deleted file mode 100644 index ec0377acb..000000000 --- a/src/modules/Elsa.AzureServiceBus/Options/AzureServiceBusOptions.cs +++ /dev/null @@ -1,30 +0,0 @@ -using Elsa.AzureServiceBus.Models; - -namespace Elsa.AzureServiceBus.Options; - -/// -/// A collection of settings to configure integration with Azure Service Bus. -/// -public class AzureServiceBusOptions -{ - /// - /// Th connection string or connection string name to connect with the service bus. - /// - public string ConnectionStringOrName { get; set; } = default!; - - /// - /// A list of s to create. - /// - public ICollection Queues { get; set; } = new List(); - - /// - /// A list of s to create. - /// - public ICollection Topics { get; set; } = new List(); - - /// - /// A list of s to create. - /// - [Obsolete("Use TopicDefinition.Subscriptions instead.")] - public ICollection Subscriptions { get; set; } = new List(); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Providers/ConfigurationQueueTopicAndSubscriptionProvider.cs b/src/modules/Elsa.AzureServiceBus/Providers/ConfigurationQueueTopicAndSubscriptionProvider.cs deleted file mode 100644 index e15370af5..000000000 --- a/src/modules/Elsa.AzureServiceBus/Providers/ConfigurationQueueTopicAndSubscriptionProvider.cs +++ /dev/null @@ -1,28 +0,0 @@ -using Elsa.AzureServiceBus.Contracts; -using Elsa.AzureServiceBus.Models; -using Elsa.AzureServiceBus.Options; -using Microsoft.Extensions.Options; - -namespace Elsa.AzureServiceBus.Providers; - -/// -/// Represents a queue provider that reads queue definitions from configuration. -/// -public class ConfigurationQueueTopicAndSubscriptionProvider : IQueueProvider, ITopicProvider, ISubscriptionProvider -{ - private readonly AzureServiceBusOptions _options; - - /// - /// Constructor. - /// - public ConfigurationQueueTopicAndSubscriptionProvider(IOptions options) => _options = options.Value; - - /// - public ValueTask> GetQueuesAsync(CancellationToken cancellationToken) => new(_options.Queues); - - /// - public ValueTask> GetTopicsAsync(CancellationToken cancellationToken) => new(_options.Topics); - - /// - public ValueTask> GetSubscriptionsAsync(CancellationToken cancellationToken) => new(_options.Subscriptions); -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Services/ServiceBusInitializer.cs b/src/modules/Elsa.AzureServiceBus/Services/ServiceBusInitializer.cs deleted file mode 100644 index cd54f0ff3..000000000 --- a/src/modules/Elsa.AzureServiceBus/Services/ServiceBusInitializer.cs +++ /dev/null @@ -1,90 +0,0 @@ -using Azure.Messaging.ServiceBus.Administration; -using Elsa.AzureServiceBus.Contracts; - -namespace Elsa.AzureServiceBus.Services; - -/// -public class ServiceBusInitializer( - ServiceBusAdministrationClient serviceBusAdministrationClient, - IEnumerable queueProviders, - IEnumerable topicProviders, - IEnumerable subscriptionProviders) - : IServiceBusInitializer -{ - private readonly IReadOnlyCollection _queueProviders = queueProviders.ToList(); - private readonly IReadOnlyCollection _topicProviders = topicProviders.ToList(); - private readonly IReadOnlyCollection _subscriptionProviders = subscriptionProviders.ToList(); - - /// - public async Task InitializeAsync(CancellationToken cancellationToken = default) - { - var tasks = new[] - { - CreateQueuesAsync(cancellationToken), - CreateTopicsAsync(cancellationToken), - CreateSubscriptionsAsync(cancellationToken) - }; - await Task.WhenAll(tasks); - } - - private async Task CreateQueuesAsync(CancellationToken cancellationToken) - { - var definitions = (await Task.WhenAll(_queueProviders.Select(async x => await x.GetQueuesAsync(cancellationToken)))).SelectMany(x => x); - var parallelOptions = new ParallelOptions - { - CancellationToken = cancellationToken, - MaxDegreeOfParallelism = 5 - }; - await Parallel.ForEachAsync(definitions, parallelOptions, async (definition, ct) => - { - if (!await serviceBusAdministrationClient.QueueExistsAsync(definition.Name, ct)) - await serviceBusAdministrationClient.CreateQueueAsync(definition.Name, ct); - }); - } - - private async Task CreateTopicsAsync(CancellationToken cancellationToken) - { - var definitionTasks = _topicProviders.Select(async x => await x.GetTopicsAsync(cancellationToken)); - var definitions = (await Task.WhenAll(definitionTasks)).SelectMany(x => x).ToList(); - - if (!definitions.Any()) - return; - - var parallelOptions = new ParallelOptions - { - CancellationToken = cancellationToken, - MaxDegreeOfParallelism = 5 - }; - await Parallel.ForEachAsync(definitions, parallelOptions, async (topic, ct) => - { - if (!await serviceBusAdministrationClient.TopicExistsAsync(topic.Name, ct)) - await serviceBusAdministrationClient.CreateTopicAsync(topic.Name, ct); - - foreach (var subscription in topic.Subscriptions) - { - if (!await serviceBusAdministrationClient.SubscriptionExistsAsync(topic.Name, subscription.Name, ct)) - await serviceBusAdministrationClient.CreateSubscriptionAsync(topic.Name, subscription.Name, ct); - } - }); - } - - private async Task CreateSubscriptionsAsync(CancellationToken cancellationToken) - { - var definitionTasks = _subscriptionProviders.Select(async x => await x.GetSubscriptionsAsync(cancellationToken)); - var definitions = (await Task.WhenAll(definitionTasks)).SelectMany(x => x).ToList(); - - if (!definitions.Any()) - return; - - var parallelOptions = new ParallelOptions - { - CancellationToken = cancellationToken, - MaxDegreeOfParallelism = 5 - }; - await Parallel.ForEachAsync(definitions, parallelOptions, async (definition, ct) => - { - if (!await serviceBusAdministrationClient.SubscriptionExistsAsync(definition.Topic, definition.Name, ct)) - await serviceBusAdministrationClient.CreateSubscriptionAsync(definition.Topic, definition.Name, ct); - }); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Services/Worker.cs b/src/modules/Elsa.AzureServiceBus/Services/Worker.cs deleted file mode 100644 index 6d7478c0d..000000000 --- a/src/modules/Elsa.AzureServiceBus/Services/Worker.cs +++ /dev/null @@ -1,122 +0,0 @@ -using Azure.Messaging.ServiceBus; -using Elsa.AzureServiceBus.Activities; -using Elsa.AzureServiceBus.Models; -using Elsa.Workflows.Runtime; -using Microsoft.Extensions.DependencyInjection; -using Microsoft.Extensions.Logging; - -namespace Elsa.AzureServiceBus.Services; - -/// -/// Processes messages received via a queue specified through the . -/// When a message is received, the appropriate workflows are executed. -/// -public class Worker : IAsyncDisposable -{ - private readonly ServiceBusProcessor _processor; - private readonly IServiceScopeFactory _serviceScopeFactory; - private readonly ILogger _logger; - - /// - /// Initializes a new instance of the class. - /// - public Worker(string queueOrTopic, string? subscription, ServiceBusClient client, IServiceScopeFactory serviceScopeFactory, ILogger logger) - { - QueueOrTopic = queueOrTopic; - Subscription = subscription == "" ? default : subscription; - _serviceScopeFactory = serviceScopeFactory; - _logger = logger; - - var options = new ServiceBusProcessorOptions(); - var processor = string.IsNullOrEmpty(subscription) ? client.CreateProcessor(queueOrTopic, options) : client.CreateProcessor(queueOrTopic, subscription, options); - - processor.ProcessMessageAsync += OnMessageReceivedAsync; - processor.ProcessErrorAsync += OnErrorAsync; - _processor = processor; - } - - /// - /// The name of the queue or topic that this worker is processing. - /// - public string QueueOrTopic { get; } - - /// - /// The name of the subscription that this worker is processing. Only valid if the worker is processing a topic. - /// - public string? Subscription { get; } - - /// - /// Starts the worker. - /// - /// The cancellation token. - public async Task StartAsync(CancellationToken cancellationToken = default) => await _processor.StartProcessingAsync(cancellationToken); - - /// - /// Disposes the worker. - /// - public async ValueTask DisposeAsync() - { - _processor.ProcessMessageAsync -= OnMessageReceivedAsync; - _processor.ProcessErrorAsync -= OnErrorAsync; - await _processor.DisposeAsync(); - } - - private async Task OnMessageReceivedAsync(ProcessMessageEventArgs args) => await InvokeWorkflowsAsync(args.Message, args.CancellationToken); - - private Task OnErrorAsync(ProcessErrorEventArgs args) - { - _logger.LogError(args.Exception, "An error occurred while processing {EntityPath}", args.EntityPath); - return Task.CompletedTask; - } - - private async Task InvokeWorkflowsAsync(ServiceBusReceivedMessage message, CancellationToken cancellationToken) - { - var input = new Dictionary - { - [MessageReceived.InputKey] = CreateMessageModel(message) - }; - - var metadata = new StimulusMetadata - { - CorrelationId = message.CorrelationId, - Input = input, - }; - var stimulus = new MessageReceivedStimulus(QueueOrTopic, Subscription); - await using var scope = _serviceScopeFactory.CreateAsyncScope(); - var stimulusSender = scope.ServiceProvider.GetRequiredService(); - var result = await stimulusSender.SendAsync(stimulus, metadata, cancellationToken); - - _logger.LogDebug("{Count} workflow triggered by the service bus message", result.WorkflowInstanceResponses.Count); - } - - private static ReceivedServiceBusMessageModel CreateMessageModel(ServiceBusReceivedMessage message) - { - return new ReceivedServiceBusMessageModel - { - Body = message.Body.ToArray(), - Subject = message.Subject, - ContentType = message.ContentType, - To = message.To, - CorrelationId = message.CorrelationId, - DeliveryCount = message.DeliveryCount, - EnqueuedTime = message.EnqueuedTime, - ScheduledEnqueuedTime = message.ScheduledEnqueueTime, - ExpiresAt = message.ExpiresAt, - LockedUntil = message.LockedUntil, - TimeToLive = message.TimeToLive, - LockToken = message.LockToken, - MessageId = message.MessageId, - PartitionKey = message.PartitionKey, - TransactionPartitionKey = message.TransactionPartitionKey, - ReplyTo = message.ReplyTo, - SequenceNumber = message.SequenceNumber, - EnqueuedSequenceNumber = message.EnqueuedSequenceNumber, - SessionId = message.SessionId, - ReplyToSessionId = message.ReplyToSessionId, - DeadLetterReason = message.DeadLetterReason, - DeadLetterSource = message.DeadLetterSource, - DeadLetterErrorDescription = message.DeadLetterErrorDescription, - ApplicationProperties = message.ApplicationProperties - }; - } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Services/WorkerManager.cs b/src/modules/Elsa.AzureServiceBus/Services/WorkerManager.cs deleted file mode 100644 index 06a6a96e3..000000000 --- a/src/modules/Elsa.AzureServiceBus/Services/WorkerManager.cs +++ /dev/null @@ -1,43 +0,0 @@ -using Elsa.AzureServiceBus.Contracts; -using Microsoft.Extensions.DependencyInjection; - -namespace Elsa.AzureServiceBus.Services; - -/// -/// Manages message workers. -/// -public class WorkerManager(IServiceProvider serviceProvider) : IWorkerManager, IAsyncDisposable -{ - private readonly ICollection _workers = new List(); - - /// - /// A list of workers under management. - /// - public IEnumerable Workers => _workers.ToList(); - - /// - public async Task StartWorkerAsync(string queueOrTopic, string? subscription, CancellationToken cancellationToken = default) - { - var worker = FindWorkerFor(queueOrTopic, subscription); - if (worker != null) return; - await CreateWorkerAsync(queueOrTopic, subscription, cancellationToken); - } - - /// - public Worker? FindWorkerFor(string queueOrTopic, string? subscription) => _workers.FirstOrDefault(x => x.QueueOrTopic == queueOrTopic && x.Subscription == subscription); - - private async Task CreateWorkerAsync(string queueOrTopic, string? subscription, CancellationToken cancellationToken = default) - { - subscription ??= ""; - var worker = ActivatorUtilities.CreateInstance(serviceProvider, queueOrTopic, subscription!); - - _workers.Add(worker); - await worker.StartAsync(cancellationToken); - } - - /// - public async ValueTask DisposeAsync() - { - foreach (var worker in Workers) await worker.DisposeAsync(); - } -} \ No newline at end of file diff --git a/src/modules/Elsa.AzureServiceBus/Tasks/StartWorkers.cs b/src/modules/Elsa.AzureServiceBus/Tasks/StartWorkers.cs deleted file mode 100644 index 0e6f5d367..000000000 --- a/src/modules/Elsa.AzureServiceBus/Tasks/StartWorkers.cs +++ /dev/null @@ -1,48 +0,0 @@ -using Elsa.AzureServiceBus.Activities; -using Elsa.AzureServiceBus.Contracts; -using Elsa.AzureServiceBus.Models; -using Elsa.Common; -using Elsa.Extensions; -using Elsa.Workflows.Helpers; -using Elsa.Workflows.Runtime; -using Elsa.Workflows.Runtime.Filters; -using JetBrains.Annotations; - -namespace Elsa.AzureServiceBus.Tasks; - -/// -/// Creates workers for each trigger & bookmark in response to updated workflow trigger indexes and bookmarks. -/// -[UsedImplicitly] -public class StartWorkers(ITriggerStore triggerStore, IBookmarkStore bookmarkStore, IWorkerManager workerManager) : BackgroundTask -{ - /// - public override async Task StartAsync(CancellationToken cancellationToken) - { - var activityType = ActivityTypeNameHelper.GenerateTypeName(); - var triggerFilter = new TriggerFilter - { - Name = activityType - }; - var triggerStimuli = (await triggerStore.FindManyAsync(triggerFilter, cancellationToken)).Select(x => x.GetPayload()).ToList(); - var bookmarkFilter = new BookmarkFilter - { - ActivityTypeName = activityType - }; - var bookmarkStimuli = (await bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken)).Select(x => x.GetPayload()).ToList(); - var stimuli = triggerStimuli.Concat(bookmarkStimuli).ToList(); - - await EnsureWorkersAsync(stimuli, cancellationToken); - } - - public override Task StopAsync(CancellationToken cancellationToken) - { - return Task.CompletedTask; - } - - private async Task EnsureWorkersAsync(IEnumerable stimuli, CancellationToken cancellationToken) - { - foreach (var stimulus in stimuli) - await workerManager.StartWorkerAsync(stimulus.QueueOrTopic, stimulus.Subscription, cancellationToken); - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Abstractions/AppComponentTest.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Abstractions/AppComponentTest.cs deleted file mode 100644 index f0b77831d..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Abstractions/AppComponentTest.cs +++ /dev/null @@ -1,24 +0,0 @@ -using Microsoft.Extensions.DependencyInjection; - -namespace Elsa.AzureServiceBus.ComponentTests; - -[Collection(nameof(ServiceBusAppCollection))] -public abstract class AppComponentTest(App app) : IDisposable //IAsyncLifetime -{ - protected WorkflowServer WorkflowServer { get; } = app.WorkflowServer; - protected Infrastructure Infrastructure { get; } = app.Infrastructure; - protected IServiceScope Scope { get; private set; } = app.WorkflowServer.Services.CreateScope(); - - public void Dispose() - { - // Disposing the Scope here and in other places where it is created somehow seems to cause the test runner to hang when running other test projects. - // Let's comment it out for the time being. - //Scope.Dispose(); - - OnDispose(); - } - - protected virtual void OnDispose() - { - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/AzureServiceBusTests.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/AzureServiceBusTests.cs deleted file mode 100644 index 486604c10..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/AzureServiceBusTests.cs +++ /dev/null @@ -1,83 +0,0 @@ -using Azure.Messaging.ServiceBus; -using Elsa.AzureServiceBus.ComponentTests.Workflows; -using Elsa.Testing.Shared; -using Elsa.Testing.Shared.Services; -using Elsa.Workflows; -using Elsa.Workflows.Management; -using Elsa.Workflows.Management.Filters; -using Microsoft.Extensions.DependencyInjection; - -namespace Elsa.AzureServiceBus.ComponentTests; - -public class AzureServiceBusTests : AppComponentTest -{ - private static readonly object WorkflowCompletedSignal = new(); - private readonly SignalManager _signalManager; - private readonly WorkflowEvents _workflowEvents; - - public AzureServiceBusTests(App app) : base(app) - { - _signalManager = Scope.ServiceProvider.GetRequiredService(); - _workflowEvents = Scope.ServiceProvider.GetRequiredService(); - _workflowEvents.WorkflowInstanceSaved += OnWorkflowInstanceSaved; - } - - private void OnWorkflowInstanceSaved(object? sender, WorkflowInstanceSavedEventArgs e) - { - if (e.WorkflowInstance.Status != WorkflowStatus.Finished) - return; - - if (e.WorkflowInstance.DefinitionId == MessageReceivedTriggerWorkflow.DefinitionId) - _signalManager.Trigger(WorkflowCompletedSignal, e); - } - - [Fact] - public async Task WorkflowReceivesMessage_WhenSendingMessageToTopic() - { - await using var client = Scope.ServiceProvider.GetRequiredService(); - - var topic = MessageReceivedTriggerWorkflow.Topic; - - // Generate a correlation ID so that we can find the workflow instance later. - var correlationId = Guid.NewGuid().ToString(); - - await using var sender = client.CreateSender(topic); - - // Send a message to the topic. This should trigger the workflow. - await sender.SendMessageAsync(new ServiceBusMessage("Message 1") - { - CorrelationId = correlationId - }); - - // Wait for the workflow to trigger the first signal. - await _signalManager.WaitAsync(MessageReceivedTriggerWorkflow.Signal1); - - // Send another message to the topic. This should resume the workflow. - await sender.SendMessageAsync(new ServiceBusMessage("Message 2")); - - // Wait for the workflow to trigger the second signal. - await _signalManager.WaitAsync(MessageReceivedTriggerWorkflow.Signal2); - - // Wait for the workflow to complete. - await _signalManager.WaitAsync(WorkflowCompletedSignal); - - // Find the workflow instance by correlation ID. - var workflowInstanceStore = Scope.ServiceProvider.GetRequiredService(); - var workflowInstanceFilter = new WorkflowInstanceFilter - { - CorrelationId = correlationId - }; - var workflowInstance = await workflowInstanceStore.FindAsync(workflowInstanceFilter); - - Assert.NotNull(workflowInstance); - - // Assert that the workflow is finished. - Assert.Equal(WorkflowStatus.Finished, workflowInstance.Status); - Assert.Equal(WorkflowSubStatus.Finished, workflowInstance.SubStatus); - } - - protected override void OnDispose() - { - _workflowEvents.WorkflowInstanceSaved -= OnWorkflowInstanceSaved; - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj b/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj deleted file mode 100644 index 6c8b199ea..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj +++ /dev/null @@ -1,29 +0,0 @@ - - - - net9.0 - enable - enable - - false - true - - - - - - - - - - - - - - - - - - - - diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj.DotSettings b/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj.DotSettings deleted file mode 100644 index ccbdc8a6e..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Elsa.AzureServiceBus.ComponentTests.csproj.DotSettings +++ /dev/null @@ -1,4 +0,0 @@ - - True - True - True \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Extensions/AzureServiceBusServiceCollectionExtensions.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Extensions/AzureServiceBusServiceCollectionExtensions.cs deleted file mode 100644 index 088594526..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Extensions/AzureServiceBusServiceCollectionExtensions.cs +++ /dev/null @@ -1,82 +0,0 @@ -using System.Text.Json; -using Azure.Messaging.ServiceBus; -using Azure.Messaging.ServiceBus.Administration; -using Elsa.Extensions; -using Microsoft.Extensions.DependencyInjection; -using NSubstitute; - -namespace Elsa.AzureServiceBus.ComponentTests.Extensions; - -public static class AzureServiceBusServiceCollectionExtensions -{ - public static IServiceCollection AddAzureServiceBusTestServices(this IServiceCollection services) - { - var serviceBusClient = Substitute.For(); - var senders = new Dictionary(); - var processors = new Dictionary(); - - serviceBusClient.CreateSender(Arg.Any()).Returns(createSenderCall => - { - var queueOrTopicName = createSenderCall.Arg(); - - return senders.GetOrAdd(queueOrTopicName, () => - { - var serviceBusSender = Substitute.For(); - serviceBusSender.SendMessageAsync(Arg.Any(), Arg.Any()).Returns(sendMessageCall => - { - var message = sendMessageCall.Arg(); - var args = CreateMessageArgs(message); - var serviceBusProcessor = processors.Where(x => x.Key.StartsWith(queueOrTopicName, StringComparison.OrdinalIgnoreCase)).ToList(); - foreach (var (_, processor) in serviceBusProcessor) - processor.RaiseProcessMessageAsync(args); - return Task.CompletedTask; - }); - return serviceBusSender; - }); - }); - serviceBusClient.CreateProcessor(Arg.Any(), Arg.Any()).Returns(createProcessorCall => - { - var queueOrTopicName = createProcessorCall.Arg(); - var key = queueOrTopicName; - return processors.GetOrAdd(key, () => - { - return new MockServiceBusProcessor(() => - { - processors.Remove(key); - }); - }); - }); - serviceBusClient.CreateProcessor(Arg.Any(), Arg.Any(), Arg.Any()).Returns(createProcessorCall => - { - var queueOrTopicName = createProcessorCall.ArgAt(0); - var subscription = createProcessorCall.ArgAt(1); - var key = $"{queueOrTopicName}:{subscription}"; - return processors.GetOrAdd(key, () => - { - return new MockServiceBusProcessor(() => - { - processors.Remove(key); - }); - }); - }); - - services.AddSingleton(serviceBusClient); - services.AddSingleton(Substitute.For()); - return services; - } - - private static ProcessMessageEventArgs CreateMessageArgs(ServiceBusMessage transportMessage, int deliveryCount = 1) - { - var payloadJson = JsonSerializer.Serialize(transportMessage); - var props = new Dictionary(); - - var message = ServiceBusModelFactory.ServiceBusReceivedMessage( - body: BinaryData.FromString(payloadJson), - deliveryCount: deliveryCount, - correlationId: transportMessage.CorrelationId, - properties: props - ); - - return new ProcessMessageEventArgs(message, null, new CancellationToken()); - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/App.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/App.cs deleted file mode 100644 index d28195921..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/App.cs +++ /dev/null @@ -1,24 +0,0 @@ -namespace Elsa.AzureServiceBus.ComponentTests; - -public class App : IAsyncLifetime -{ - public App() - { - Infrastructure = new(); - WorkflowServer = new(Infrastructure, "http://localhost:5004"); - } - - public Infrastructure Infrastructure { get; set; } - public WorkflowServer WorkflowServer { get; set; } - - public async Task InitializeAsync() - { - await Infrastructure.InitializeAsync(); - } - - public async Task DisposeAsync() - { - await Infrastructure.DisposeAsync(); - await WorkflowServer.DisposeAsync(); - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/Infrastructure.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/Infrastructure.cs deleted file mode 100644 index 1766a08e4..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/Infrastructure.cs +++ /dev/null @@ -1,32 +0,0 @@ -using Testcontainers.PostgreSql; -using Testcontainers.RabbitMq; - -namespace Elsa.AzureServiceBus.ComponentTests; - -public class Infrastructure : IAsyncLifetime -{ - public readonly PostgreSqlContainer DbContainer = new PostgreSqlBuilder() - .WithImage("postgres:latest") - .WithDatabase("elsa") - .WithUsername("postgres") - .WithPassword("postgres") - .Build(); - - public readonly RabbitMqContainer RabbitMqContainer = new RabbitMqBuilder() - .WithImage("rabbitmq:4-management") - .Build(); - - public Task InitializeAsync() - { - return Task.WhenAll( - DbContainer.StartAsync(), - RabbitMqContainer.StartAsync()); - } - - public Task DisposeAsync() - { - return Task.WhenAll( - DbContainer.StopAsync(), - RabbitMqContainer.StopAsync()); - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/ServiceBusAppCollection.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/ServiceBusAppCollection.cs deleted file mode 100644 index 8b87b9603..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/ServiceBusAppCollection.cs +++ /dev/null @@ -1,9 +0,0 @@ -namespace Elsa.AzureServiceBus.ComponentTests; - -[CollectionDefinition(nameof(ServiceBusAppCollection))] -public class ServiceBusAppCollection : ICollectionFixture -{ - // This class has no code, and is never created. Its purpose is simply - // to be the place to apply [CollectionDefinition] and all the - // ICollectionFixture<> interfaces. -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs deleted file mode 100644 index 93c90530b..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Fixtures/WorkflowServer.cs +++ /dev/null @@ -1,98 +0,0 @@ -using System.Net.Http.Headers; -using Elsa.Agents; -using Elsa.Alterations.Extensions; -using Elsa.AzureServiceBus.ComponentTests.Extensions; -using Elsa.EntityFrameworkCore.Extensions; -using Elsa.EntityFrameworkCore.Modules.Alterations; -using Elsa.EntityFrameworkCore.Modules.Identity; -using Elsa.EntityFrameworkCore.Modules.Management; -using Elsa.EntityFrameworkCore.Modules.Runtime; -using Elsa.Extensions; -using Elsa.Identity.Providers; -using Elsa.MassTransit.Extensions; -using Elsa.Testing.Shared.Handlers; -using Elsa.Testing.Shared.Services; -using FluentStorage; -using JetBrains.Annotations; -using Microsoft.AspNetCore.Hosting; -using Microsoft.AspNetCore.Mvc.Testing; -using Microsoft.AspNetCore.TestHost; -using Microsoft.Extensions.DependencyInjection; - -namespace Elsa.AzureServiceBus.ComponentTests; - -[UsedImplicitly] -public class WorkflowServer(Infrastructure infrastructure, string url) : WebApplicationFactory -{ - protected override void ConfigureWebHost(IWebHostBuilder builder) - { - var dbConnectionString = infrastructure.DbContainer.GetConnectionString(); - var rabbitMqConnectionString = infrastructure.RabbitMqContainer.GetConnectionString(); - - builder.UseUrls(url); - - if (Program.ConfigureForTest == null) - { - Program.ConfigureForTest = elsa => - { - elsa.AddWorkflowsFrom(); - elsa.AddActivitiesFrom(); - elsa.UseDefaultAuthentication(defaultAuthentication => defaultAuthentication.UseAdminApiKey()); - elsa.UseFluentStorageProvider(sp => - { - var assemblyLocation = System.Reflection.Assembly.GetExecutingAssembly().Location; - var assemblyDirectory = Path.GetDirectoryName(assemblyLocation)!; - var workflowsDirectorySegments = new[] - { - assemblyDirectory, "Scenarios" - }; - var workflowsDirectory = Path.Join(workflowsDirectorySegments); - return StorageFactory.Blobs.DirectoryFiles(workflowsDirectory); - }); - elsa.UseMassTransit(massTransit => - { - massTransit.UseRabbitMq(rabbitMqConnectionString); - }); - elsa.UseIdentity(identity => identity.UseEntityFrameworkCore(ef => ef.UsePostgreSql(dbConnectionString))); - elsa.UseWorkflowManagement(management => - { - management.UseEntityFrameworkCore(ef => ef.UsePostgreSql(dbConnectionString)); - management.UseMassTransitDispatcher(); - management.UseCache(); - }); - elsa.UseWorkflowRuntime(runtime => - { - runtime.UseEntityFrameworkCore(ef => ef.UsePostgreSql(dbConnectionString)); - runtime.UseCache(); - runtime.UseMassTransitDispatcher(); - runtime.UseProtoActor(); - }); - elsa.UseAlterations(alterations => - { - alterations.UseEntityFrameworkCore(e => e.UsePostgreSql(dbConnectionString)); - }); - elsa.UseHttp(http => - { - http.UseCache(); - }); - elsa.UseAzureServiceBus(); - elsa.UseAgents(); - elsa.UseAgentPersistence(feature => feature.UseEntityFrameworkCore(ef => ef.UsePostgreSql(typeof(AgentsPostgreSqlProvidersExtensions).Assembly, dbConnectionString))); - }; - } - - builder.ConfigureTestServices(services => - { - services.AddSingleton(); - services.AddSingleton(); - services.AddNotificationHandlersFrom(); - services.AddNotificationHandlersFrom(); - services.AddAzureServiceBusTestServices(); - }); - } - - protected override void ConfigureClient(HttpClient client) - { - client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("ApiKey", AdminApiKeyProvider.DefaultApiKey); - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Mocks/MockServiceBusProcessor.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Mocks/MockServiceBusProcessor.cs deleted file mode 100644 index 0946283ab..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Mocks/MockServiceBusProcessor.cs +++ /dev/null @@ -1,23 +0,0 @@ -using Azure.Messaging.ServiceBus; - -namespace Elsa.AzureServiceBus.ComponentTests; - -public class MockServiceBusProcessor(Action onClose) : ServiceBusProcessor -{ - public Task RaiseProcessMessageAsync(ProcessMessageEventArgs args) - { - return base.OnProcessMessageAsync(args); - } - - public override Task StartProcessingAsync(CancellationToken cancellationToken = default) - { - // No-op. - return Task.CompletedTask; - } - - public override async Task CloseAsync(CancellationToken cancellationToken = default) - { - await base.CloseAsync(cancellationToken); - onClose(); - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/MessageReceivedTriggerWorkflow.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/MessageReceivedTriggerWorkflow.cs deleted file mode 100644 index d3f62b19d..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/MessageReceivedTriggerWorkflow.cs +++ /dev/null @@ -1,36 +0,0 @@ -using Elsa.AzureServiceBus.Activities; -using Elsa.AzureServiceBus.Models; -using Elsa.Testing.Shared.Activities; -using Elsa.Workflows; -using Elsa.Workflows.Activities; - -namespace Elsa.AzureServiceBus.ComponentTests.Workflows; - -public class MessageReceivedTriggerWorkflow : WorkflowBase -{ - public static readonly string DefinitionId = Guid.NewGuid().ToString(); - public static readonly string Topic = nameof(MessageReceivedTriggerWorkflow); - public static readonly object Signal1 = new(); - public static readonly object Signal2 = new(); - - protected override void Build(IWorkflowBuilder builder) - { - builder.WithDefinitionId(DefinitionId); - var message = builder.WithVariable(); - builder.Root = new Sequence - { - Activities = - { - new MessageReceived(Topic, "subscription1") - { - CanStartWorkflow = true, - TransportMessage = new(message) - }, - new Correlate(context => message.Get(context)!.CorrelationId), - new TriggerSignal(Signal1), - new MessageReceived(Topic, "subscription2"), - new TriggerSignal(Signal2) - } - }; - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/SendOneMessageWithCorrelationIdWorkflow.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/SendOneMessageWithCorrelationIdWorkflow.cs deleted file mode 100644 index 8a525b1b1..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/SendOneMessageWithCorrelationIdWorkflow.cs +++ /dev/null @@ -1,34 +0,0 @@ -using Elsa.AzureServiceBus.Activities; -using Elsa.Testing.Shared.Activities; -using Elsa.Workflows; -using Elsa.Workflows.Activities; - -namespace Elsa.AzureServiceBus.ComponentTests.Workflows; - -public class SendOneMessageWithCorrelationIdWorkflow : WorkflowBase -{ - public static readonly string Topic = nameof(SendOneMessageWithCorrelationIdWorkflow); - public static readonly string CorrelationId = Guid.NewGuid().ToString(); - public static readonly object Signal1 = new(); - - protected override void Build(IWorkflowBuilder builder) - { - builder.Root = new Sequence - { - Activities = - { - new SendMessage - { - QueueOrTopic = new(Topic), - MessageBody = new("Hello World"), - }, - new Correlate - { - CorrelationId = new(CorrelationId) - }, - new MessageReceived(Topic, "subscription2"), - new TriggerSignal(Signal1) - } - }; - } -} \ No newline at end of file diff --git a/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/SendOneMessageWorkflow.cs b/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/SendOneMessageWorkflow.cs deleted file mode 100644 index 77b5a5c79..000000000 --- a/test/component/Elsa.AzureServiceBus.ComponentTests/Workflows/SendOneMessageWorkflow.cs +++ /dev/null @@ -1,35 +0,0 @@ -using Elsa.AzureServiceBus.Activities; -using Elsa.Testing.Shared.Activities; -using Elsa.Workflows; -using Elsa.Workflows.Activities; - -namespace Elsa.AzureServiceBus.ComponentTests.Workflows; - -public class SendOneMessageWorkflow : WorkflowBase -{ - public static readonly string Topic = nameof(SendOneMessageWorkflow); - public static readonly object Signal1 = new(); - - protected override void Build(IWorkflowBuilder builder) - { - builder.Root = new Sequence - { - Activities = - { - new Fork - { - Branches = - { - new MessageReceived(Topic, "subscription1"), - new SendMessage - { - QueueOrTopic = new(Topic), - MessageBody = new("Hello World"), - } - } - }, - new TriggerSignal(Signal1) - } - }; - } -} \ No newline at end of file