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