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.
This commit is contained in:
Sipke Schoorstra
parent
ce0ea7e680
commit
43157fb6d6
14
Elsa.sln
14
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}
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@
|
|||
<ProjectReference Include="..\..\modules\Elsa.Agents.Api\Elsa.Agents.Api.csproj" />
|
||||
<ProjectReference Include="..\..\modules\Elsa.Agents.Persistence.EntityFrameworkCore.MySql\Elsa.Agents.Persistence.EntityFrameworkCore.MySql.csproj" Condition=" '$(TargetFramework)' != 'net9.0' " />
|
||||
<ProjectReference Include="..\..\modules\Elsa.Agents.Persistence.EntityFrameworkCore.Sqlite\Elsa.Agents.Persistence.EntityFrameworkCore.Sqlite.csproj" />
|
||||
<ProjectReference Include="..\..\modules\Elsa.AzureServiceBus\Elsa.AzureServiceBus.csproj" />
|
||||
<ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.MassTransit\Elsa.Caching.Distributed.MassTransit.csproj" />
|
||||
<ProjectReference Include="..\..\modules\Elsa.Caching.Distributed.ProtoActor\Elsa.Caching.Distributed.ProtoActor.csproj" />
|
||||
<ProjectReference Include="..\..\modules\Elsa.EntityFrameworkCore.MySql\Elsa.EntityFrameworkCore.MySql.csproj" Condition=" '$(TargetFramework)' != 'net9.0' " />
|
||||
|
|
|
|||
|
|
@ -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 =>
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Triggered when a message is received on a specified queue or topic and subscription.
|
||||
/// </summary>
|
||||
[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";
|
||||
|
||||
/// <inheritdoc />
|
||||
public MessageReceived([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
|
||||
{
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public MessageReceived(Input<string> queue)
|
||||
{
|
||||
QueueOrTopic = queue;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public MessageReceived(string queue) : this(new Input<string>(queue))
|
||||
{
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public MessageReceived(Input<string> topic, Input<string> subscription)
|
||||
{
|
||||
QueueOrTopic = topic;
|
||||
Subscription = subscription;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public MessageReceived(string topic, string subscription) : this(new Input<string>(topic), new Input<string>(subscription))
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The name of the queue or topic to read from.
|
||||
/// </summary>
|
||||
[Input(Description = "The name of the queue or topic to read from.")]
|
||||
public Input<string> QueueOrTopic { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The name of the subscription to read from.
|
||||
/// </summary>
|
||||
[Input(Description = "The name of the subscription to read from.")]
|
||||
public Input<string>? Subscription { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// The .NET type to deserialize the message into. Defaults to <see cref="string"/>.
|
||||
/// </summary>
|
||||
[Input(Description = "The .NET type to deserialize the message into.")]
|
||||
public Input<Type> MessageType { get; set; } = new(typeof(string));
|
||||
|
||||
/// <summary>
|
||||
/// The received transport message.
|
||||
/// </summary>
|
||||
[Output(Description = "The received transport message.")]
|
||||
public Output<ReceivedServiceBusMessageModel> TransportMessage { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The received transport message.
|
||||
/// </summary>
|
||||
[Output(Description = "The received message.")]
|
||||
public Output<object> Message { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The formatter to use to parse the message.
|
||||
/// </summary>
|
||||
[Input(Description = "The formatter to use to serialize the message.")]
|
||||
public Input<IFormatter?> Formatter { get; set; } = default!;
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override object GetTriggerPayload(TriggerIndexingContext context) => GetStimulus(context.ExpressionExecutionContext);
|
||||
|
||||
/// <inheritdoc />
|
||||
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<ReceivedServiceBusMessageModel>(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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Sends a message to a queue or topic in Azure Service Bus.
|
||||
/// </summary>
|
||||
[Activity("Elsa.AzureServiceBus.Send", "Azure Service Bus", "Send a message to a queue or topic")]
|
||||
[PublicAPI]
|
||||
public class SendMessage : CodeActivity
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public SendMessage([CallerFilePath] string? source = default, [CallerLineNumber] int? line = default) : base(source, line)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The contents of the message to send.
|
||||
/// </summary>
|
||||
[Input(Description = "The contents of the message to send.")]
|
||||
public Input<object> MessageBody { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The queue or topic to send the message to.
|
||||
/// </summary>
|
||||
public Input<string> QueueOrTopic { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The content type of the message.
|
||||
/// </summary>
|
||||
public Input<string>? ContentType { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// The subject of the message.
|
||||
/// </summary>
|
||||
public Input<string>? Subject { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// The correlation ID of the message.
|
||||
/// </summary>
|
||||
public Input<string>? CorrelationId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// The formatter to use when serializing the message body.
|
||||
/// </summary>
|
||||
public Input<Type?> FormatterType { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The application properties to embed with the Service Bus Message
|
||||
/// </summary>
|
||||
[Input(Category = "Advanced",
|
||||
DefaultSyntax = "Json",
|
||||
SupportedSyntaxes = ["JavaScript", "Json"],
|
||||
UIHint = InputUIHints.MultiLine)
|
||||
]
|
||||
public Input<IDictionary<string, object>?> ApplicationProperties { get; set; } = default!;
|
||||
|
||||
/// <inheritdoc />
|
||||
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<ServiceBusClient>();
|
||||
await using var sender = client.CreateSender(queueOrTopic);
|
||||
await sender.SendMessageAsync(message, cancellationToken);
|
||||
}
|
||||
|
||||
private async ValueTask<BinaryData> 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<IFormatter>().First(x => x.GetType() == formatterType);
|
||||
var data = await formatter.ToStringAsync(value, cancellationToken);
|
||||
|
||||
return BinaryData.FromString(data);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,14 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Models;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Contracts;
|
||||
|
||||
/// <summary>
|
||||
/// Provides queue definitions to the system.
|
||||
/// </summary>
|
||||
public interface IQueueProvider
|
||||
{
|
||||
/// <summary>
|
||||
/// Returns a list of <see cref="QueueDefinition"/>s.
|
||||
/// </summary>
|
||||
ValueTask<ICollection<QueueDefinition>> GetQueuesAsync(CancellationToken cancellationToken);
|
||||
}
|
||||
|
|
@ -1,12 +0,0 @@
|
|||
namespace Elsa.AzureServiceBus.Contracts;
|
||||
|
||||
/// <summary>
|
||||
/// Creates queues, topics and subscriptions provided by <see cref="IQueueProvider"/>, <see cref="ITopicProvider"/> and <see cref="ISubscriptionProvider"/> implementations.
|
||||
/// </summary>
|
||||
public interface IServiceBusInitializer
|
||||
{
|
||||
/// <summary>
|
||||
/// Creates queues, topics and subscriptions provided by <see cref="IQueueProvider"/>, <see cref="ITopicProvider"/> and <see cref="ISubscriptionProvider"/> implementations.
|
||||
/// </summary>
|
||||
Task InitializeAsync(CancellationToken cancellationToken = default);
|
||||
}
|
||||
|
|
@ -1,16 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Models;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Contracts;
|
||||
|
||||
/// <summary>
|
||||
/// Provides subscription definitions to the system.
|
||||
/// </summary>
|
||||
[Obsolete("Use AzureServiceBusOptions.Topics instead.")]
|
||||
|
||||
public interface ISubscriptionProvider
|
||||
{
|
||||
/// <summary>
|
||||
/// Return a list of <see cref="SubscriptionDefinition"/>s.
|
||||
/// </summary>
|
||||
ValueTask<ICollection<SubscriptionDefinition>> GetSubscriptionsAsync(CancellationToken cancellationToken);
|
||||
}
|
||||
|
|
@ -1,14 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Models;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Contracts;
|
||||
|
||||
/// <summary>
|
||||
/// Provides topic definitions to the system.
|
||||
/// </summary>
|
||||
public interface ITopicProvider
|
||||
{
|
||||
/// <summary>
|
||||
/// Returns a list of <see cref="TopicDefinition"/>s.
|
||||
/// </summary>
|
||||
ValueTask<ICollection<TopicDefinition>> GetTopicsAsync(CancellationToken cancellationToken);
|
||||
}
|
||||
|
|
@ -1,27 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Services;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Contracts;
|
||||
|
||||
/// <summary>
|
||||
/// Manages message workers.
|
||||
/// </summary>
|
||||
public interface IWorkerManager
|
||||
{
|
||||
/// <summary>
|
||||
/// A list of workers under management.
|
||||
/// </summary>
|
||||
IEnumerable<Worker> Workers { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Ensures that at least one worker exists for the specified queue/topic and subscription.
|
||||
/// </summary>
|
||||
Task StartWorkerAsync(string queueOrTopic, string? subscription, CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// Finds a worker for the specified queue or topic and subscription.
|
||||
/// </summary>
|
||||
/// <param name="queueOrTopic">The name of the queue or topic.</param>
|
||||
/// <param name="subscription">The name of the subscription.</param>
|
||||
/// <returns>The worker, or null if no worker was found.</returns>
|
||||
Worker? FindWorkerFor(string queueOrTopic, string? subscription);
|
||||
}
|
||||
|
|
@ -1,23 +0,0 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<Description>
|
||||
Provides Azure Service Bus integration and activities.
|
||||
</Description>
|
||||
<PackageTags>elsa module azure-service-bus service-bus</PackageTags>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.Messaging.ServiceBus" />
|
||||
<PackageReference Include="Azure.ResourceManager.ServiceBus" />
|
||||
<PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" />
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" />
|
||||
<PackageReference Include="System.Linq.Async" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj" />
|
||||
<ProjectReference Include="..\Elsa.Workflows.Runtime\Elsa.Workflows.Runtime.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
@ -1,25 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Features;
|
||||
using Elsa.Features.Services;
|
||||
|
||||
// ReSharper disable once CheckNamespace
|
||||
namespace Elsa.Extensions;
|
||||
|
||||
/// <summary>
|
||||
/// Adds extension methods to <see cref="IModule"/> to register Azure Service Bus related services.
|
||||
/// </summary>
|
||||
public static class ModuleExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Enable and configure the <see cref="AzureServiceBusFeature"/> feature.
|
||||
/// </summary>
|
||||
public static IModule UseAzureServiceBus(this IModule module, string connectionStringOrName, Action<AzureServiceBusFeature>? setup = null)
|
||||
{
|
||||
setup += feature => feature.AzureServiceBusOptions += options => options.ConnectionStringOrName = connectionStringOrName;
|
||||
return module.Use(setup);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Enable and configure the <see cref="AzureServiceBusFeature"/> feature.
|
||||
/// </summary>
|
||||
public static IModule UseAzureServiceBus(this IModule module, Action<AzureServiceBusFeature>? setup = null) => module.Use(setup);
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Enables and configures the Azure Service Bus feature.
|
||||
/// </summary>
|
||||
public class AzureServiceBusFeature : FeatureBase
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public AzureServiceBusFeature(IModule module) : base(module)
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A value controlling whether queues, topics and subscriptions should be created automatically.
|
||||
/// </summary>
|
||||
public bool CreateQueuesTopicsAndSubscriptions { get; set; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// A delegate to configure <see cref="AzureServiceBusOptions"/>.
|
||||
/// </summary>
|
||||
public Action<AzureServiceBusOptions> AzureServiceBusOptions { get; set; } = _ => { };
|
||||
|
||||
/// <summary>
|
||||
/// A delegate to create a <see cref="ServiceBusAdministrationClient"/> instance.
|
||||
/// </summary>
|
||||
public Func<IServiceProvider, ServiceBusClient> ServiceBusClientFactory { get; set; } = sp => new(GetConnectionString(sp));
|
||||
|
||||
/// <summary>
|
||||
/// A delegate to create a <see cref="ServiceBusAdministrationClient"/> instance.
|
||||
/// </summary>
|
||||
public Func<IServiceProvider, ServiceBusAdministrationClient> ServiceBusAdministrationClientFactory { get; set; } = sp => new(GetConnectionString(sp));
|
||||
|
||||
/// <inheritdoc />
|
||||
public override void ConfigureHostedServices()
|
||||
{
|
||||
if (CreateQueuesTopicsAndSubscriptions)
|
||||
Module.ConfigureHostedService<CreateQueuesTopicsAndSubscriptions>();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override void Configure()
|
||||
{
|
||||
// Activities.
|
||||
Module.AddActivitiesFrom<AzureServiceBusFeature>();
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override void Apply()
|
||||
{
|
||||
Services.Configure(AzureServiceBusOptions);
|
||||
|
||||
Services
|
||||
.AddSingleton(ServiceBusAdministrationClientFactory)
|
||||
.AddSingleton(ServiceBusClientFactory)
|
||||
.AddSingleton<ConfigurationQueueTopicAndSubscriptionProvider>()
|
||||
.AddSingleton<IWorkerManager, WorkerManager>()
|
||||
.AddScoped<IServiceBusInitializer, ServiceBusInitializer>();
|
||||
|
||||
// Tasks.
|
||||
Services.AddBackgroundTask<StartWorkers>();
|
||||
|
||||
// Definition providers.
|
||||
Services
|
||||
.AddSingleton<IQueueProvider>(sp => sp.GetRequiredService<ConfigurationQueueTopicAndSubscriptionProvider>())
|
||||
.AddSingleton<ITopicProvider>(sp => sp.GetRequiredService<ConfigurationQueueTopicAndSubscriptionProvider>())
|
||||
.AddSingleton<ISubscriptionProvider>(sp => sp.GetRequiredService<ConfigurationQueueTopicAndSubscriptionProvider>());
|
||||
|
||||
// Handlers.
|
||||
Services.AddHandlersFrom<UpdateWorkers>();
|
||||
}
|
||||
|
||||
private static string GetConnectionString(IServiceProvider serviceProvider)
|
||||
{
|
||||
var options = serviceProvider.GetRequiredService<IOptions<AzureServiceBusOptions>>().Value;
|
||||
var configuration = serviceProvider.GetRequiredService<IConfiguration>();
|
||||
return configuration.GetConnectionString(options.ConnectionStringOrName) ?? options.ConnectionStringOrName;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
<Weavers xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="FodyWeavers.xsd">
|
||||
<ConfigureAwait />
|
||||
</Weavers>
|
||||
|
|
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Creates workers for each trigger & bookmark in response to updated workflow trigger indexes and bookmarks.
|
||||
/// </summary>
|
||||
[UsedImplicitly]
|
||||
public class UpdateWorkers(IWorkerManager workerManager) : INotificationHandler<WorkflowTriggersIndexed>, INotificationHandler<WorkflowBookmarksIndexed>
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds, updates and removes workers based on added and removed triggers.
|
||||
/// </summary>
|
||||
public async Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken)
|
||||
{
|
||||
var added = notification.IndexedWorkflowTriggers.AddedTriggers.Filter<MessageReceived>().Select(x => x.GetPayload<MessageReceivedStimulus>());
|
||||
await StartWorkersAsync(added, cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adds, updates and removes workers based on added and removed bookmarks.
|
||||
/// </summary>
|
||||
public async Task HandleAsync(WorkflowBookmarksIndexed notification, CancellationToken cancellationToken)
|
||||
{
|
||||
var added = notification.IndexedWorkflowBookmarks.AddedBookmarks.Filter<MessageReceived>().Select(x => x.GetPayload<MessageReceivedStimulus>());
|
||||
await StartWorkersAsync(added, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task StartWorkersAsync(IEnumerable<MessageReceivedStimulus> payloads, CancellationToken cancellationToken)
|
||||
{
|
||||
foreach (var payload in payloads) await workerManager.StartWorkerAsync(payload.QueueOrTopic, payload.Subscription, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,24 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Contracts;
|
||||
using JetBrains.Annotations;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
|
||||
namespace Elsa.AzureServiceBus.HostedServices;
|
||||
|
||||
/// <summary>
|
||||
/// A blocking hosted service that creates queues, topics and subscriptions.
|
||||
/// </summary>
|
||||
[UsedImplicitly]
|
||||
public class CreateQueuesTopicsAndSubscriptions(IServiceScopeFactory scopeFactory) : IHostedService
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await using var scope = scopeFactory.CreateAsyncScope();
|
||||
var initializer = scope.ServiceProvider.GetRequiredService<IServiceBusInitializer>();
|
||||
await initializer.InitializeAsync(cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
}
|
||||
|
|
@ -1,47 +0,0 @@
|
|||
using System.Text.Json.Serialization;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Models;
|
||||
|
||||
/// <summary>
|
||||
/// A bookmark payload model for triggering workflows when messages come in at a given queue or topic and subscription.
|
||||
/// </summary>
|
||||
public record MessageReceivedStimulus
|
||||
{
|
||||
private readonly string _queueOrTopic = default!;
|
||||
private readonly string? _subscription;
|
||||
|
||||
/// <summary>
|
||||
/// Constructor.
|
||||
/// </summary>
|
||||
[JsonConstructor]
|
||||
public MessageReceivedStimulus()
|
||||
{
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Constructor.
|
||||
/// </summary>
|
||||
public MessageReceivedStimulus(string queueOrTopic, string? subscription)
|
||||
{
|
||||
QueueOrTopic = queueOrTopic;
|
||||
Subscription = subscription;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The queue or topic to trigger from.
|
||||
/// </summary>
|
||||
public string QueueOrTopic
|
||||
{
|
||||
get => _queueOrTopic;
|
||||
init => _queueOrTopic = value.ToLowerInvariant();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The subscription to trigger from.
|
||||
/// </summary>
|
||||
public string? Subscription
|
||||
{
|
||||
get => _subscription;
|
||||
init => _subscription = value?.ToLowerInvariant();
|
||||
}
|
||||
}
|
||||
|
|
@ -1,6 +0,0 @@
|
|||
namespace Elsa.AzureServiceBus.Models;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a queue that is available to the system.
|
||||
/// </summary>
|
||||
public record QueueDefinition(string Name);
|
||||
|
|
@ -1,36 +0,0 @@
|
|||
using JetBrains.Annotations;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Models;
|
||||
|
||||
/// <summary>
|
||||
/// A serializable version of <see cref="Azure.Messaging.ServiceBus.ServiceBusReceivedMessage"/>.
|
||||
/// </summary>
|
||||
// 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<string, object> ApplicationProperties { get; init; } = new Dictionary<string, object>();
|
||||
}
|
||||
|
|
@ -1,18 +0,0 @@
|
|||
namespace Elsa.AzureServiceBus.Models;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a topic subscription that is available to the system.
|
||||
/// </summary>
|
||||
public class SubscriptionDefinition
|
||||
{
|
||||
/// <summary>
|
||||
/// The subscription name.
|
||||
/// </summary>
|
||||
public string Name { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The topic.
|
||||
/// </summary>
|
||||
[Obsolete("Use TopicDefinition.Subscriptions instead.")]
|
||||
public string? Topic { get; set; }
|
||||
}
|
||||
|
|
@ -1,17 +0,0 @@
|
|||
namespace Elsa.AzureServiceBus.Models;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a topic that is available to the system.
|
||||
/// </summary>
|
||||
public class TopicDefinition
|
||||
{
|
||||
/// <summary>
|
||||
/// The topic name.
|
||||
/// </summary>
|
||||
public string Name { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// The subscriptions.
|
||||
/// </summary>
|
||||
public ICollection<SubscriptionDefinition> Subscriptions { get; set; } = new List<SubscriptionDefinition>();
|
||||
}
|
||||
|
|
@ -1,30 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Models;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Options;
|
||||
|
||||
/// <summary>
|
||||
/// A collection of settings to configure integration with Azure Service Bus.
|
||||
/// </summary>
|
||||
public class AzureServiceBusOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// Th connection string or connection string name to connect with the service bus.
|
||||
/// </summary>
|
||||
public string ConnectionStringOrName { get; set; } = default!;
|
||||
|
||||
/// <summary>
|
||||
/// A list of <see cref="QueueDefinition"/>s to create.
|
||||
/// </summary>
|
||||
public ICollection<QueueDefinition> Queues { get; set; } = new List<QueueDefinition>();
|
||||
|
||||
/// <summary>
|
||||
/// A list of <see cref="TopicDefinition"/>s to create.
|
||||
/// </summary>
|
||||
public ICollection<TopicDefinition> Topics { get; set; } = new List<TopicDefinition>();
|
||||
|
||||
/// <summary>
|
||||
/// A list of <see cref="SubscriptionDefinition"/>s to create.
|
||||
/// </summary>
|
||||
[Obsolete("Use TopicDefinition.Subscriptions instead.")]
|
||||
public ICollection<SubscriptionDefinition> Subscriptions { get; set; } = new List<SubscriptionDefinition>();
|
||||
}
|
||||
|
|
@ -1,28 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Contracts;
|
||||
using Elsa.AzureServiceBus.Models;
|
||||
using Elsa.AzureServiceBus.Options;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Providers;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a queue provider that reads queue definitions from configuration.
|
||||
/// </summary>
|
||||
public class ConfigurationQueueTopicAndSubscriptionProvider : IQueueProvider, ITopicProvider, ISubscriptionProvider
|
||||
{
|
||||
private readonly AzureServiceBusOptions _options;
|
||||
|
||||
/// <summary>
|
||||
/// Constructor.
|
||||
/// </summary>
|
||||
public ConfigurationQueueTopicAndSubscriptionProvider(IOptions<AzureServiceBusOptions> options) => _options = options.Value;
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<ICollection<QueueDefinition>> GetQueuesAsync(CancellationToken cancellationToken) => new(_options.Queues);
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<ICollection<TopicDefinition>> GetTopicsAsync(CancellationToken cancellationToken) => new(_options.Topics);
|
||||
|
||||
/// <inheritdoc />
|
||||
public ValueTask<ICollection<SubscriptionDefinition>> GetSubscriptionsAsync(CancellationToken cancellationToken) => new(_options.Subscriptions);
|
||||
}
|
||||
|
|
@ -1,90 +0,0 @@
|
|||
using Azure.Messaging.ServiceBus.Administration;
|
||||
using Elsa.AzureServiceBus.Contracts;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Services;
|
||||
|
||||
/// <inheritdoc />
|
||||
public class ServiceBusInitializer(
|
||||
ServiceBusAdministrationClient serviceBusAdministrationClient,
|
||||
IEnumerable<IQueueProvider> queueProviders,
|
||||
IEnumerable<ITopicProvider> topicProviders,
|
||||
IEnumerable<ISubscriptionProvider> subscriptionProviders)
|
||||
: IServiceBusInitializer
|
||||
{
|
||||
private readonly IReadOnlyCollection<IQueueProvider> _queueProviders = queueProviders.ToList();
|
||||
private readonly IReadOnlyCollection<ITopicProvider> _topicProviders = topicProviders.ToList();
|
||||
private readonly IReadOnlyCollection<ISubscriptionProvider> _subscriptionProviders = subscriptionProviders.ToList();
|
||||
|
||||
/// <inheritdoc />
|
||||
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);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Processes messages received via a queue specified through the <see cref="MessageReceivedStimulus"/>.
|
||||
/// When a message is received, the appropriate workflows are executed.
|
||||
/// </summary>
|
||||
public class Worker : IAsyncDisposable
|
||||
{
|
||||
private readonly ServiceBusProcessor _processor;
|
||||
private readonly IServiceScopeFactory _serviceScopeFactory;
|
||||
private readonly ILogger _logger;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="Worker"/> class.
|
||||
/// </summary>
|
||||
public Worker(string queueOrTopic, string? subscription, ServiceBusClient client, IServiceScopeFactory serviceScopeFactory, ILogger<Worker> 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;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The name of the queue or topic that this worker is processing.
|
||||
/// </summary>
|
||||
public string QueueOrTopic { get; }
|
||||
|
||||
/// <summary>
|
||||
/// The name of the subscription that this worker is processing. Only valid if the worker is processing a topic.
|
||||
/// </summary>
|
||||
public string? Subscription { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Starts the worker.
|
||||
/// </summary>
|
||||
/// <param name="cancellationToken">The cancellation token.</param>
|
||||
public async Task StartAsync(CancellationToken cancellationToken = default) => await _processor.StartProcessingAsync(cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Disposes the worker.
|
||||
/// </summary>
|
||||
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<string, object>
|
||||
{
|
||||
[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<IStimulusSender>();
|
||||
var result = await stimulusSender.SendAsync<MessageReceived>(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
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -1,43 +0,0 @@
|
|||
using Elsa.AzureServiceBus.Contracts;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace Elsa.AzureServiceBus.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Manages message workers.
|
||||
/// </summary>
|
||||
public class WorkerManager(IServiceProvider serviceProvider) : IWorkerManager, IAsyncDisposable
|
||||
{
|
||||
private readonly ICollection<Worker> _workers = new List<Worker>();
|
||||
|
||||
/// <summary>
|
||||
/// A list of workers under management.
|
||||
/// </summary>
|
||||
public IEnumerable<Worker> Workers => _workers.ToList();
|
||||
|
||||
/// <inheritdoc />
|
||||
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);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
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<Worker>(serviceProvider, queueOrTopic, subscription!);
|
||||
|
||||
_workers.Add(worker);
|
||||
await worker.StartAsync(cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
foreach (var worker in Workers) await worker.DisposeAsync();
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Creates workers for each trigger & bookmark in response to updated workflow trigger indexes and bookmarks.
|
||||
/// </summary>
|
||||
[UsedImplicitly]
|
||||
public class StartWorkers(ITriggerStore triggerStore, IBookmarkStore bookmarkStore, IWorkerManager workerManager) : BackgroundTask
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public override async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var activityType = ActivityTypeNameHelper.GenerateTypeName<MessageReceived>();
|
||||
var triggerFilter = new TriggerFilter
|
||||
{
|
||||
Name = activityType
|
||||
};
|
||||
var triggerStimuli = (await triggerStore.FindManyAsync(triggerFilter, cancellationToken)).Select(x => x.GetPayload<MessageReceivedStimulus>()).ToList();
|
||||
var bookmarkFilter = new BookmarkFilter
|
||||
{
|
||||
ActivityTypeName = activityType
|
||||
};
|
||||
var bookmarkStimuli = (await bookmarkStore.FindManyAsync(bookmarkFilter, cancellationToken)).Select(x => x.GetPayload<MessageReceivedStimulus>()).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<MessageReceivedStimulus> stimuli, CancellationToken cancellationToken)
|
||||
{
|
||||
foreach (var stimulus in stimuli)
|
||||
await workerManager.StartWorkerAsync(stimulus.QueueOrTopic, stimulus.Subscription, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
|
@ -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()
|
||||
{
|
||||
}
|
||||
}
|
||||
|
|
@ -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<SignalManager>();
|
||||
_workflowEvents = Scope.ServiceProvider.GetRequiredService<WorkflowEvents>();
|
||||
_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<ServiceBusClient>();
|
||||
|
||||
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<IWorkflowInstanceStore>();
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,29 +0,0 @@
|
|||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net9.0</TargetFramework>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
|
||||
<IsPackable>false</IsPackable>
|
||||
<IsTestProject>true</IsTestProject>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Azure.Messaging.ServiceBus"/>
|
||||
<PackageReference Include="JetBrains.Annotations" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<Using Include="Xunit"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\..\src\apps\Elsa.Server.Web\Elsa.Server.Web.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\common\Elsa.Testing.Shared.Component\Elsa.Testing.Shared.Component.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\modules\Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql\Elsa.Agents.Persistence.EntityFrameworkCore.PostgreSql.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\modules\Elsa.AzureServiceBus\Elsa.AzureServiceBus.csproj" />
|
||||
<ProjectReference Include="..\..\..\src\modules\Elsa.Workflows.Core\Elsa.Workflows.Core.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
|
|
@ -1,4 +0,0 @@
|
|||
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=abstractions/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=fixtures/@EntryIndexedValue">True</s:Boolean>
|
||||
<s:Boolean x:Key="/Default/CodeInspection/NamespaceProvider/NamespaceFoldersToSkip/=mocks/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
|
||||
|
|
@ -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<ServiceBusClient>();
|
||||
var senders = new Dictionary<string, ServiceBusSender>();
|
||||
var processors = new Dictionary<string, MockServiceBusProcessor>();
|
||||
|
||||
serviceBusClient.CreateSender(Arg.Any<string>()).Returns(createSenderCall =>
|
||||
{
|
||||
var queueOrTopicName = createSenderCall.Arg<string>();
|
||||
|
||||
return senders.GetOrAdd(queueOrTopicName, () =>
|
||||
{
|
||||
var serviceBusSender = Substitute.For<ServiceBusSender>();
|
||||
serviceBusSender.SendMessageAsync(Arg.Any<ServiceBusMessage>(), Arg.Any<CancellationToken>()).Returns(sendMessageCall =>
|
||||
{
|
||||
var message = sendMessageCall.Arg<ServiceBusMessage>();
|
||||
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<string>(), Arg.Any<ServiceBusProcessorOptions>()).Returns(createProcessorCall =>
|
||||
{
|
||||
var queueOrTopicName = createProcessorCall.Arg<string>();
|
||||
var key = queueOrTopicName;
|
||||
return processors.GetOrAdd(key, () =>
|
||||
{
|
||||
return new MockServiceBusProcessor(() =>
|
||||
{
|
||||
processors.Remove(key);
|
||||
});
|
||||
});
|
||||
});
|
||||
serviceBusClient.CreateProcessor(Arg.Any<string>(), Arg.Any<string>(), Arg.Any<ServiceBusProcessorOptions>()).Returns(createProcessorCall =>
|
||||
{
|
||||
var queueOrTopicName = createProcessorCall.ArgAt<string>(0);
|
||||
var subscription = createProcessorCall.ArgAt<string>(1);
|
||||
var key = $"{queueOrTopicName}:{subscription}";
|
||||
return processors.GetOrAdd(key, () =>
|
||||
{
|
||||
return new MockServiceBusProcessor(() =>
|
||||
{
|
||||
processors.Remove(key);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
services.AddSingleton(serviceBusClient);
|
||||
services.AddSingleton(Substitute.For<ServiceBusAdministrationClient>());
|
||||
return services;
|
||||
}
|
||||
|
||||
private static ProcessMessageEventArgs CreateMessageArgs(ServiceBusMessage transportMessage, int deliveryCount = 1)
|
||||
{
|
||||
var payloadJson = JsonSerializer.Serialize(transportMessage);
|
||||
var props = new Dictionary<string, object>();
|
||||
|
||||
var message = ServiceBusModelFactory.ServiceBusReceivedMessage(
|
||||
body: BinaryData.FromString(payloadJson),
|
||||
deliveryCount: deliveryCount,
|
||||
correlationId: transportMessage.CorrelationId,
|
||||
properties: props
|
||||
);
|
||||
|
||||
return new ProcessMessageEventArgs(message, null, new CancellationToken());
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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());
|
||||
}
|
||||
}
|
||||
|
|
@ -1,9 +0,0 @@
|
|||
namespace Elsa.AzureServiceBus.ComponentTests;
|
||||
|
||||
[CollectionDefinition(nameof(ServiceBusAppCollection))]
|
||||
public class ServiceBusAppCollection : ICollectionFixture<App>
|
||||
{
|
||||
// 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.
|
||||
}
|
||||
|
|
@ -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<Program>
|
||||
{
|
||||
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<WorkflowServer>();
|
||||
elsa.AddActivitiesFrom<WorkflowServer>();
|
||||
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<SignalManager>();
|
||||
services.AddSingleton<WorkflowEvents>();
|
||||
services.AddNotificationHandlersFrom<WorkflowServer>();
|
||||
services.AddNotificationHandlersFrom<WorkflowEventHandlers>();
|
||||
services.AddAzureServiceBusTestServices();
|
||||
});
|
||||
}
|
||||
|
||||
protected override void ConfigureClient(HttpClient client)
|
||||
{
|
||||
client.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("ApiKey", AdminApiKeyProvider.DefaultApiKey);
|
||||
}
|
||||
}
|
||||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
@ -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<ReceivedServiceBusMessageModel>();
|
||||
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)
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue