From 5abd71f6c7691f63f66bca3589e32aeab7198547 Mon Sep 17 00:00:00 2001 From: Gideon de Swardt Date: Sat, 3 Apr 2021 22:32:37 +0100 Subject: [PATCH 1/8] Add Azure Service Bus module Added an ABP module for Azure Service Bus that implements common configuration, services, and factories that can be reused in any module. Added extension methods to the ServiceBusAdministrationClient to check if a topic and subscription exist and if not create it. Improved the performance of publishing and processing messages by creating a pool for ServiceBusAdministrationClient, ServiceBusClient, ServiceBusSender and ServiceBusProcessor. --- framework/Volo.Abp.sln | 7 ++ .../Volo.Abp.AzureServiceBus/FodyWeavers.xml | 3 + .../Volo.Abp.AzureServiceBus/FodyWeavers.xsd | 30 +++++ .../Volo.Abp.AzureServiceBus.csproj | 17 +++ .../AbpAzureServiceBusModule.cs | 20 ++++ .../AbpAzureServiceBusOptions.cs | 12 ++ .../AzureServiceBusConnections.cs | 31 ++++++ .../AzureServiceBusMessageConsumer.cs | 104 ++++++++++++++++++ .../AzureServiceBusMessageConsumerFactory.cs | 29 +++++ .../Volo/Abp/AzureServiceBus/ClientConfig.cs | 16 +++ .../Abp/AzureServiceBus/ConnectionPool.cs | 88 +++++++++++++++ .../IAzureServiceBusMessageConsumer.cs | 11 ++ .../IAzureServiceBusMessageConsumerFactory.cs | 21 ++++ .../IAzureServiceBusSerializer.cs | 11 ++ .../Abp/AzureServiceBus/IConnectionPool.cs | 13 +++ .../Abp/AzureServiceBus/IProcessorPool.cs | 11 ++ .../Abp/AzureServiceBus/IPublisherPool.cs | 11 ++ .../Volo/Abp/AzureServiceBus/ProcessorPool.cs | 82 ++++++++++++++ .../Volo/Abp/AzureServiceBus/PublisherPool.cs | 66 +++++++++++ ...erviceBusAdministrationClientExtensions.cs | 25 +++++ .../Utf8JsonAzureServiceBusSerializer.cs | 27 +++++ 21 files changed, 635 insertions(+) create mode 100644 framework/src/Volo.Abp.AzureServiceBus/FodyWeavers.xml create mode 100644 framework/src/Volo.Abp.AzureServiceBus/FodyWeavers.xsd create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo.Abp.AzureServiceBus.csproj create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AbpAzureServiceBusModule.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AbpAzureServiceBusOptions.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusConnections.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusMessageConsumer.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusMessageConsumerFactory.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ClientConfig.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ConnectionPool.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusMessageConsumer.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusMessageConsumerFactory.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusSerializer.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IConnectionPool.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IProcessorPool.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IPublisherPool.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ProcessorPool.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/PublisherPool.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ServiceBusAdministrationClientExtensions.cs create mode 100644 framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/Utf8JsonAzureServiceBusSerializer.cs diff --git a/framework/Volo.Abp.sln b/framework/Volo.Abp.sln index 67d2fa2573..56fd40cee3 100644 --- a/framework/Volo.Abp.sln +++ b/framework/Volo.Abp.sln @@ -383,6 +383,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Volo.Abp.AspNetCore.Compone EndProject Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Volo.Abp.AspNetCore.Mvc.UI.Bundling.Abstractions", "src\Volo.Abp.AspNetCore.Mvc.UI.Bundling.Abstractions\Volo.Abp.AspNetCore.Mvc.UI.Bundling.Abstractions.csproj", "{E9CE58DB-0789-4D18-8B63-474F7D7B14B4}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Volo.Abp.AzureServiceBus", "src\Volo.Abp.AzureServiceBus\Volo.Abp.AzureServiceBus.csproj", "{808EC18E-C8CC-4F5C-82B6-984EADBBF85D}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -1141,6 +1143,10 @@ Global {E9CE58DB-0789-4D18-8B63-474F7D7B14B4}.Debug|Any CPU.Build.0 = Debug|Any CPU {E9CE58DB-0789-4D18-8B63-474F7D7B14B4}.Release|Any CPU.ActiveCfg = Release|Any CPU {E9CE58DB-0789-4D18-8B63-474F7D7B14B4}.Release|Any CPU.Build.0 = Release|Any CPU + {808EC18E-C8CC-4F5C-82B6-984EADBBF85D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {808EC18E-C8CC-4F5C-82B6-984EADBBF85D}.Debug|Any CPU.Build.0 = Debug|Any CPU + {808EC18E-C8CC-4F5C-82B6-984EADBBF85D}.Release|Any CPU.ActiveCfg = Release|Any CPU + {808EC18E-C8CC-4F5C-82B6-984EADBBF85D}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -1334,6 +1340,7 @@ Global {863C18F9-2407-49F9-9ADC-F6229AF3B385} = {5DF0E140-0513-4D0D-BE2E-3D4D85CD70E6} {B4B6B7DE-9798-4007-B1DF-7EE7929E392A} = {5DF0E140-0513-4D0D-BE2E-3D4D85CD70E6} {E9CE58DB-0789-4D18-8B63-474F7D7B14B4} = {5DF0E140-0513-4D0D-BE2E-3D4D85CD70E6} + {808EC18E-C8CC-4F5C-82B6-984EADBBF85D} = {5DF0E140-0513-4D0D-BE2E-3D4D85CD70E6} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {BB97ECF4-9A84-433F-A80B-2A3285BDD1D5} diff --git a/framework/src/Volo.Abp.AzureServiceBus/FodyWeavers.xml b/framework/src/Volo.Abp.AzureServiceBus/FodyWeavers.xml new file mode 100644 index 0000000000..be0de3a908 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/FodyWeavers.xml @@ -0,0 +1,3 @@ + + + \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/FodyWeavers.xsd b/framework/src/Volo.Abp.AzureServiceBus/FodyWeavers.xsd new file mode 100644 index 0000000000..3f3946e282 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/FodyWeavers.xsd @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + 'true' to run assembly verification (PEVerify) on the target assembly after all weavers have been executed. + + + + + A comma-separated list of error codes that can be safely ignored in assembly verification. + + + + + 'false' to turn off automatic generation of the XML Schema file. + + + + + \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo.Abp.AzureServiceBus.csproj b/framework/src/Volo.Abp.AzureServiceBus/Volo.Abp.AzureServiceBus.csproj new file mode 100644 index 0000000000..77ccf8f772 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo.Abp.AzureServiceBus.csproj @@ -0,0 +1,17 @@ + + + + + + + netstandard2.0 + latest + + + + + + + + + diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AbpAzureServiceBusModule.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AbpAzureServiceBusModule.cs new file mode 100644 index 0000000000..e9d1c20cd3 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AbpAzureServiceBusModule.cs @@ -0,0 +1,20 @@ +using Microsoft.Extensions.DependencyInjection; +using Volo.Abp.Json; +using Volo.Abp.Modularity; +using Volo.Abp.Threading; + +namespace Volo.Abp.AzureServiceBus +{ + [DependsOn( + typeof(AbpJsonModule), + typeof(AbpThreadingModule) + )] + public class AbpAzureServiceBusModule : AbpModule + { + public override void ConfigureServices(ServiceConfigurationContext context) + { + var configuration = context.Services.GetConfiguration(); + Configure(configuration.GetSection("Azure:ServiceBus")); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AbpAzureServiceBusOptions.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AbpAzureServiceBusOptions.cs new file mode 100644 index 0000000000..7672502d28 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AbpAzureServiceBusOptions.cs @@ -0,0 +1,12 @@ +namespace Volo.Abp.AzureServiceBus +{ + public class AbpAzureServiceBusOptions + { + public AzureServiceBusConnections Connections { get; } + + public AbpAzureServiceBusOptions() + { + Connections = new AzureServiceBusConnections(); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusConnections.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusConnections.cs new file mode 100644 index 0000000000..f8b0683d96 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusConnections.cs @@ -0,0 +1,31 @@ +using System; +using System.Collections.Generic; +using JetBrains.Annotations; + +namespace Volo.Abp.AzureServiceBus +{ + [Serializable] + public class AzureServiceBusConnections : Dictionary + { + public const string DefaultConnectionName = "Default"; + + [NotNull] + public ClientConfig Default + { + get => this[DefaultConnectionName]; + set => this[DefaultConnectionName] = Check.NotNull(value, nameof(value)); + } + + public AzureServiceBusConnections() + { + Default = new ClientConfig(); + } + + public ClientConfig GetOrDefault(string connectionName) + { + return TryGetValue(connectionName, out var connectionFactory) + ? connectionFactory + : Default; + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusMessageConsumer.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusMessageConsumer.cs new file mode 100644 index 0000000000..6eeb8f7693 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusMessageConsumer.cs @@ -0,0 +1,104 @@ +using System; +using System.Collections.Concurrent; +using System.Threading; +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus; +using JetBrains.Annotations; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Volo.Abp.DependencyInjection; +using Volo.Abp.ExceptionHandling; +using Volo.Abp.Threading; + +namespace Volo.Abp.AzureServiceBus +{ + public class AzureServiceBusMessageConsumer : IAzureServiceBusMessageConsumer, ITransientDependency + { + public ILogger Logger { get; set; } + + private readonly IExceptionNotifier _exceptionNotifier; + private readonly IProcessorPool _processorPool; + private readonly ConcurrentBag> _callbacks; + private string _connectionName; + private string _subscriptionName; + private string _topicName; + + public AzureServiceBusMessageConsumer( + IExceptionNotifier exceptionNotifier, + IProcessorPool processorPool) + { + _exceptionNotifier = exceptionNotifier; + _processorPool = processorPool; + Logger = NullLogger.Instance; + _callbacks = new ConcurrentBag>(); + } + + public virtual void Initialize( + [NotNull] string topicName, + [NotNull] string subscriptionName, + string connectionName) + { + Check.NotNull(topicName, nameof(topicName)); + Check.NotNull(subscriptionName, nameof(subscriptionName)); + + _topicName = topicName; + _connectionName = connectionName ?? AzureServiceBusConnections.DefaultConnectionName; + _subscriptionName = subscriptionName; + StartProcessing(); + } + + public void OnMessageReceived(Func callback) + { + _callbacks.Add(callback); + } + + protected virtual void StartProcessing() + { + Task.Factory.StartNew(function: async () => + { + var serviceBusProcessor = await _processorPool.GetAsync(_subscriptionName, _topicName, _connectionName); + serviceBusProcessor.ProcessErrorAsync += HandleIncomingError; + serviceBusProcessor.ProcessMessageAsync += HandleIncomingMessage; + + if (!serviceBusProcessor.IsProcessing) + { + await serviceBusProcessor.StartProcessingAsync(); + } + + while (true) + { + Thread.Sleep(1000); + } + }, TaskCreationOptions.LongRunning); + } + + protected virtual async Task HandleIncomingMessage(ProcessMessageEventArgs args) + { + try + { + foreach (var callback in _callbacks) + { + await callback(args.Message); + } + + await args.CompleteMessageAsync(args.Message); + } + catch (Exception exception) + { + await HandleError(exception); + } + } + + protected virtual async Task HandleIncomingError(ProcessErrorEventArgs args) + { + await HandleError(args.Exception); + } + + protected virtual async Task HandleError(Exception exception) + { + Logger.LogException(exception); + await _exceptionNotifier.NotifyAsync(exception); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusMessageConsumerFactory.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusMessageConsumerFactory.cs new file mode 100644 index 0000000000..d373de43cf --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/AzureServiceBusMessageConsumerFactory.cs @@ -0,0 +1,29 @@ +using System; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.AzureServiceBus +{ + public class AzureServiceBusMessageConsumerFactory : IAzureServiceBusMessageConsumerFactory, ISingletonDependency, IDisposable + { + protected IServiceScope ServiceScope { get; } + + public AzureServiceBusMessageConsumerFactory(IServiceScopeFactory serviceScopeFactory) + { + ServiceScope = serviceScopeFactory.CreateScope(); + } + + public IAzureServiceBusMessageConsumer CreateMessageConsumer(string topicName, string subscriptionName, string connectionName) + { + var processor = ServiceScope.ServiceProvider.GetRequiredService(); + processor.Initialize(topicName, subscriptionName, connectionName); + return processor; + } + + public void Dispose() + { + ServiceScope?.Dispose(); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ClientConfig.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ClientConfig.cs new file mode 100644 index 0000000000..9a4c8c2856 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ClientConfig.cs @@ -0,0 +1,16 @@ +using Azure.Messaging.ServiceBus; +using Azure.Messaging.ServiceBus.Administration; + +namespace Volo.Abp.AzureServiceBus +{ + public class ClientConfig + { + public string ConnectionString { get; set; } + + public ServiceBusAdministrationClientOptions Admin { get; set; } = new(); + + public ServiceBusClientOptions Client { get; set; } = new(); + + public ServiceBusProcessorOptions Processor { get; set; } = new(); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ConnectionPool.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ConnectionPool.cs new file mode 100644 index 0000000000..4f38250dec --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ConnectionPool.cs @@ -0,0 +1,88 @@ +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus; +using Azure.Messaging.ServiceBus.Administration; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.AzureServiceBus +{ + public class ConnectionPool : IConnectionPool, ISingletonDependency + { + public ILogger Logger { get; set; } + + private bool _isDisposed; + private readonly AbpAzureServiceBusOptions _options; + private readonly ConcurrentDictionary> _clients; + private readonly ConcurrentDictionary> _adminClients; + + public ConnectionPool(IOptions options) + { + _options = options.Value; + _clients = new ConcurrentDictionary>(); + _adminClients = new ConcurrentDictionary>(); + Logger = new NullLogger(); + } + + public ServiceBusClient GetClient(string connectionName) + { + connectionName ??= AzureServiceBusConnections.DefaultConnectionName; + return _clients.GetOrAdd( + connectionName, new Lazy(() => + { + var config = _options.Connections.GetOrDefault(connectionName); + return new ServiceBusClient(config.ConnectionString, config.Client); + }) + ).Value; + } + + public ServiceBusAdministrationClient GetAdministrationClient(string connectionName) + { + connectionName ??= AzureServiceBusConnections.DefaultConnectionName; + return _adminClients.GetOrAdd( + connectionName, new Lazy(() => + { + var config = _options.Connections.GetOrDefault(connectionName); + return new ServiceBusAdministrationClient(config.ConnectionString); + }) + ).Value; + } + + public async ValueTask DisposeAsync() + { + if (_isDisposed) + { + return; + } + + _isDisposed = true; + if (!_clients.Any()) + { + Logger.LogDebug($"Disposed connection pool with no connection in the pool."); + return; + } + + Logger.LogInformation($"Disposing connection pool ({_clients.Count} connections)."); + + foreach (var connection in _clients.Values) + { + await connection.Value.DisposeAsync(); + } + + _clients.Clear(); + + if (!_adminClients.Any()) + { + Logger.LogDebug($"Disposed admin connection pool with no admin connection in the pool."); + return; + } + + Logger.LogInformation($"Disposing admin connection pool ({_adminClients.Count} admin connections)."); + _adminClients.Clear(); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusMessageConsumer.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusMessageConsumer.cs new file mode 100644 index 0000000000..7a47e42e40 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusMessageConsumer.cs @@ -0,0 +1,11 @@ +using System; +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus; + +namespace Volo.Abp.AzureServiceBus +{ + public interface IAzureServiceBusMessageConsumer + { + void OnMessageReceived(Func callback); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusMessageConsumerFactory.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusMessageConsumerFactory.cs new file mode 100644 index 0000000000..0f9f20f0aa --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusMessageConsumerFactory.cs @@ -0,0 +1,21 @@ +using System.Threading.Tasks; + +namespace Volo.Abp.AzureServiceBus +{ + public interface IAzureServiceBusMessageConsumerFactory + { + /// + /// Creates a new . + /// Avoid to create too many consumers since they are + /// not disposed until end of the application. + /// + /// + /// + /// + /// + IAzureServiceBusMessageConsumer CreateMessageConsumer( + string topicName, + string subscriptionName, + string connectionName); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusSerializer.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusSerializer.cs new file mode 100644 index 0000000000..04f56d4470 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusSerializer.cs @@ -0,0 +1,11 @@ +using System; + +namespace Volo.Abp.AzureServiceBus +{ + public interface IAzureServiceBusSerializer + { + byte[] Serialize(object obj); + + object Deserialize(BinaryData value, Type type); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IConnectionPool.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IConnectionPool.cs new file mode 100644 index 0000000000..43ed188022 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IConnectionPool.cs @@ -0,0 +1,13 @@ +using System; +using Azure.Messaging.ServiceBus; +using Azure.Messaging.ServiceBus.Administration; + +namespace Volo.Abp.AzureServiceBus +{ + public interface IConnectionPool : IAsyncDisposable + { + ServiceBusClient GetClient(string connectionName); + + ServiceBusAdministrationClient GetAdministrationClient(string connectionName); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IProcessorPool.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IProcessorPool.cs new file mode 100644 index 0000000000..338bee393c --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IProcessorPool.cs @@ -0,0 +1,11 @@ +using System; +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus; + +namespace Volo.Abp.AzureServiceBus +{ + public interface IProcessorPool : IAsyncDisposable + { + Task GetAsync(string subscriptionName, string topicName, string connectionName); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IPublisherPool.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IPublisherPool.cs new file mode 100644 index 0000000000..9ae141a252 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IPublisherPool.cs @@ -0,0 +1,11 @@ +using System; +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus; + +namespace Volo.Abp.AzureServiceBus +{ + public interface IPublisherPool : IAsyncDisposable + { + Task GetAsync(string topicName, string connectionName); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ProcessorPool.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ProcessorPool.cs new file mode 100644 index 0000000000..daa804e930 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ProcessorPool.cs @@ -0,0 +1,82 @@ +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.AzureServiceBus +{ + public class ProcessorPool : IProcessorPool, ISingletonDependency + { + public ILogger Logger { get; set; } + + private bool _isDisposed; + private readonly AbpAzureServiceBusOptions _options; + private readonly IConnectionPool _connectionPool; + private readonly ConcurrentDictionary> _processors; + + public ProcessorPool( + IOptions options, + IConnectionPool connectionPool) + { + _options = options.Value; + _connectionPool = connectionPool; + _processors = new ConcurrentDictionary>(); + Logger = new NullLogger(); + } + + public async Task GetAsync(string subscriptionName, string topicName, string connectionName) + { + var admin = _connectionPool.GetAdministrationClient(connectionName); + await admin.SetupSubscriptionAsync(topicName, subscriptionName); + + return _processors.GetOrAdd( + $"{topicName}-{subscriptionName}", new Lazy(() => + { + var config = _options.Connections.GetOrDefault(connectionName); + var client = _connectionPool.GetClient(connectionName); + return client.CreateProcessor(topicName, subscriptionName, config.Processor); + }) + ).Value; + } + + public async ValueTask DisposeAsync() + { + if (_isDisposed) + { + return; + } + + _isDisposed = true; + if (!_processors.Any()) + { + Logger.LogDebug($"Disposed processor pool with no processors in the pool."); + return; + } + + Logger.LogInformation($"Disposing processor pool ({_processors.Count} processors)."); + + foreach (var item in _processors.Values) + { + var processor = item.Value; + if (processor.IsProcessing) + { + await processor.StopProcessingAsync(); + } + + if (!processor.IsClosed) + { + await processor.CloseAsync(); + } + + await processor.DisposeAsync(); + } + + _processors.Clear(); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/PublisherPool.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/PublisherPool.cs new file mode 100644 index 0000000000..57008e07cb --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/PublisherPool.cs @@ -0,0 +1,66 @@ +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Volo.Abp.DependencyInjection; + +namespace Volo.Abp.AzureServiceBus +{ + public class PublisherPool : IPublisherPool, ISingletonDependency + { + public ILogger Logger { get; set; } + + private bool _isDisposed; + private readonly IConnectionPool _connectionPool; + private readonly ConcurrentDictionary> _publishers; + + public PublisherPool(IConnectionPool connectionPool) + { + _connectionPool = connectionPool; + _publishers = new ConcurrentDictionary>(); + Logger = new NullLogger(); + } + + public async Task GetAsync(string topicName, string connectionName) + { + var admin = _connectionPool.GetAdministrationClient(connectionName); + await admin.SetupTopicAsync(topicName); + + return _publishers.GetOrAdd( + topicName, new Lazy(() => + { + var client = _connectionPool.GetClient(connectionName); + return client.CreateSender(topicName); + }) + ).Value; + } + + public async ValueTask DisposeAsync() + { + if (_isDisposed) + { + return; + } + + _isDisposed = true; + if (!_publishers.Any()) + { + Logger.LogDebug($"Disposed publisher pool with no publisher in the pool."); + return; + } + + Logger.LogInformation($"Disposing publisher pool ({_publishers.Count} publishers)."); + + foreach (var publisher in _publishers.Values) + { + await publisher.Value.CloseAsync(); + await publisher.Value.DisposeAsync(); + } + + _publishers.Clear(); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ServiceBusAdministrationClientExtensions.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ServiceBusAdministrationClientExtensions.cs new file mode 100644 index 0000000000..c6a09a155b --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/ServiceBusAdministrationClientExtensions.cs @@ -0,0 +1,25 @@ +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus.Administration; + +namespace Volo.Abp.AzureServiceBus +{ + public static class ServiceBusAdministrationClientExtensions + { + public static async Task SetupTopicAsync(this ServiceBusAdministrationClient client, string topicName) + { + if (!await client.TopicExistsAsync(topicName)) + { + await client.CreateTopicAsync(topicName); + } + } + + public static async Task SetupSubscriptionAsync(this ServiceBusAdministrationClient client, string topicName, string subscriptionName) + { + await client.SetupTopicAsync(topicName); + if (!await client.SubscriptionExistsAsync(topicName, subscriptionName)) + { + await client.CreateSubscriptionAsync(topicName, subscriptionName); + } + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/Utf8JsonAzureServiceBusSerializer.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/Utf8JsonAzureServiceBusSerializer.cs new file mode 100644 index 0000000000..0a87e22934 --- /dev/null +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/Utf8JsonAzureServiceBusSerializer.cs @@ -0,0 +1,27 @@ +using System; +using System.Text; +using Volo.Abp.DependencyInjection; +using Volo.Abp.Json; + +namespace Volo.Abp.AzureServiceBus +{ + public class Utf8JsonAzureServiceBusSerializer : IAzureServiceBusSerializer, ITransientDependency + { + private readonly IJsonSerializer _jsonSerializer; + + public Utf8JsonAzureServiceBusSerializer(IJsonSerializer jsonSerializer) + { + _jsonSerializer = jsonSerializer; + } + + public byte[] Serialize(object obj) + { + return Encoding.UTF8.GetBytes(_jsonSerializer.Serialize(obj)); + } + + public object Deserialize(BinaryData value, Type type) + { + return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value.ToArray())); + } + } +} \ No newline at end of file From 08e7aedbd9213b55d213363bbcef28a0a7ee8c28 Mon Sep 17 00:00:00 2001 From: Gideon de Swardt Date: Sat, 3 Apr 2021 22:31:37 +0100 Subject: [PATCH 2/8] Implement Distribute Events Bus with Azure Implement a new instance of the IDistributedEventBus that uses the AzureServiceBus module. Added a service that will replace the existing instance in the service registry for the distributed event bus with the Azure instance. Added configuration options for the Azure Event Bus. --- framework/Volo.Abp.sln | 7 + .../Volo.Abp.EventBus.Azure/FodyWeavers.xml | 3 + .../Volo.Abp.EventBus.Azure/FodyWeavers.xsd | 30 +++ .../Volo.Abp.EventBus.Azure.csproj | 13 ++ .../EventBus/Azure/AbpAzureEventBusOptions.cs | 11 + .../EventBus/Azure/AbpEventBusAzureModule.cs | 28 +++ .../Azure/AzureDistributedEventBus.cs | 189 ++++++++++++++++++ 7 files changed, 281 insertions(+) create mode 100644 framework/src/Volo.Abp.EventBus.Azure/FodyWeavers.xml create mode 100644 framework/src/Volo.Abp.EventBus.Azure/FodyWeavers.xsd create mode 100644 framework/src/Volo.Abp.EventBus.Azure/Volo.Abp.EventBus.Azure.csproj create mode 100644 framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AbpAzureEventBusOptions.cs create mode 100644 framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AbpEventBusAzureModule.cs create mode 100644 framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs diff --git a/framework/Volo.Abp.sln b/framework/Volo.Abp.sln index 56fd40cee3..53c15f8ea9 100644 --- a/framework/Volo.Abp.sln +++ b/framework/Volo.Abp.sln @@ -385,6 +385,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Volo.Abp.AspNetCore.Mvc.UI. EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Volo.Abp.AzureServiceBus", "src\Volo.Abp.AzureServiceBus\Volo.Abp.AzureServiceBus.csproj", "{808EC18E-C8CC-4F5C-82B6-984EADBBF85D}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Volo.Abp.EventBus.Azure", "src\Volo.Abp.EventBus.Azure\Volo.Abp.EventBus.Azure.csproj", "{FB27F78E-F10E-4810-9B8E-BCD67DCFC8A2}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -1147,6 +1149,10 @@ Global {808EC18E-C8CC-4F5C-82B6-984EADBBF85D}.Debug|Any CPU.Build.0 = Debug|Any CPU {808EC18E-C8CC-4F5C-82B6-984EADBBF85D}.Release|Any CPU.ActiveCfg = Release|Any CPU {808EC18E-C8CC-4F5C-82B6-984EADBBF85D}.Release|Any CPU.Build.0 = Release|Any CPU + {FB27F78E-F10E-4810-9B8E-BCD67DCFC8A2}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {FB27F78E-F10E-4810-9B8E-BCD67DCFC8A2}.Debug|Any CPU.Build.0 = Debug|Any CPU + {FB27F78E-F10E-4810-9B8E-BCD67DCFC8A2}.Release|Any CPU.ActiveCfg = Release|Any CPU + {FB27F78E-F10E-4810-9B8E-BCD67DCFC8A2}.Release|Any CPU.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -1341,6 +1347,7 @@ Global {B4B6B7DE-9798-4007-B1DF-7EE7929E392A} = {5DF0E140-0513-4D0D-BE2E-3D4D85CD70E6} {E9CE58DB-0789-4D18-8B63-474F7D7B14B4} = {5DF0E140-0513-4D0D-BE2E-3D4D85CD70E6} {808EC18E-C8CC-4F5C-82B6-984EADBBF85D} = {5DF0E140-0513-4D0D-BE2E-3D4D85CD70E6} + {FB27F78E-F10E-4810-9B8E-BCD67DCFC8A2} = {5DF0E140-0513-4D0D-BE2E-3D4D85CD70E6} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {BB97ECF4-9A84-433F-A80B-2A3285BDD1D5} diff --git a/framework/src/Volo.Abp.EventBus.Azure/FodyWeavers.xml b/framework/src/Volo.Abp.EventBus.Azure/FodyWeavers.xml new file mode 100644 index 0000000000..be0de3a908 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Azure/FodyWeavers.xml @@ -0,0 +1,3 @@ + + + \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus.Azure/FodyWeavers.xsd b/framework/src/Volo.Abp.EventBus.Azure/FodyWeavers.xsd new file mode 100644 index 0000000000..3f3946e282 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Azure/FodyWeavers.xsd @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + 'true' to run assembly verification (PEVerify) on the target assembly after all weavers have been executed. + + + + + A comma-separated list of error codes that can be safely ignored in assembly verification. + + + + + 'false' to turn off automatic generation of the XML Schema file. + + + + + \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo.Abp.EventBus.Azure.csproj b/framework/src/Volo.Abp.EventBus.Azure/Volo.Abp.EventBus.Azure.csproj new file mode 100644 index 0000000000..c913b79bfc --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo.Abp.EventBus.Azure.csproj @@ -0,0 +1,13 @@ + + + + netstandard2.0 + latest + + + + + + + + diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AbpAzureEventBusOptions.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AbpAzureEventBusOptions.cs new file mode 100644 index 0000000000..cb56de424a --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AbpAzureEventBusOptions.cs @@ -0,0 +1,11 @@ +namespace Volo.Abp.EventBus.Azure +{ + public class AbpAzureEventBusOptions + { + public string ConnectionName { get; set; } + + public string SubscriberName { get; set; } + + public string TopicName { get; set; } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AbpEventBusAzureModule.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AbpEventBusAzureModule.cs new file mode 100644 index 0000000000..08feabd366 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AbpEventBusAzureModule.cs @@ -0,0 +1,28 @@ +using Microsoft.Extensions.DependencyInjection; +using Volo.Abp.AzureServiceBus; +using Volo.Abp.Modularity; + +namespace Volo.Abp.EventBus.Azure +{ + [DependsOn( + typeof(AbpEventBusModule), + typeof(AbpAzureServiceBusModule) + )] + public class AbpEventBusAzureModule : AbpModule + { + public override void ConfigureServices(ServiceConfigurationContext context) + { + var configuration = context.Services.GetConfiguration(); + + Configure(configuration.GetSection("Azure:EventBus")); + } + + public override void OnApplicationInitialization(ApplicationInitializationContext context) + { + context + .ServiceProvider + .GetRequiredService() + .Initialize(); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs new file mode 100644 index 0000000000..9b0527ce69 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -0,0 +1,189 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Azure.Messaging.ServiceBus; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using Volo.Abp.DependencyInjection; +using Volo.Abp.EventBus.Distributed; +using Volo.Abp.AzureServiceBus; +using Volo.Abp.MultiTenancy; +using Volo.Abp.Threading; + +namespace Volo.Abp.EventBus.Azure +{ + [Dependency(ReplaceServices = true)] + [ExposeServices(typeof(IDistributedEventBus), typeof(AzureDistributedEventBus))] + public class AzureDistributedEventBus : EventBusBase, IDistributedEventBus, ISingletonDependency + { + private readonly AbpAzureEventBusOptions _options; + private readonly AbpDistributedEventBusOptions _distributedEventBusOptions; + private readonly IAzureServiceBusMessageConsumerFactory _messageConsumerFactory; + private readonly IPublisherPool _publisherPool; + private readonly IAzureServiceBusSerializer _serializer; + private readonly ConcurrentDictionary> _handlerFactories; + private readonly ConcurrentDictionary _eventTypes; + private IAzureServiceBusMessageConsumer _consumer; + + public AzureDistributedEventBus( + IServiceScopeFactory serviceScopeFactory, + ICurrentTenant currentTenant, + IOptions abpAzureEventBusOptions, + IOptions abpDistributedEventBusOptions, + IAzureServiceBusSerializer serializer, + IAzureServiceBusMessageConsumerFactory messageConsumerFactory, + IPublisherPool publisherPool) + : base(serviceScopeFactory, currentTenant) + { + _options = abpAzureEventBusOptions.Value; + _distributedEventBusOptions = abpDistributedEventBusOptions.Value; + _serializer = serializer; + _messageConsumerFactory = messageConsumerFactory; + _publisherPool = publisherPool; + _handlerFactories = new ConcurrentDictionary>(); + _eventTypes = new ConcurrentDictionary(); + } + + public void Initialize() + { + _consumer = _messageConsumerFactory.CreateMessageConsumer( + _options.TopicName, + _options.SubscriberName, + _options.ConnectionName); + + _consumer.OnMessageReceived(ProcessEventAsync); + SubscribeHandlers(_distributedEventBusOptions.Handlers); + } + + private async Task ProcessEventAsync(ServiceBusReceivedMessage message) + { + var eventName = message.Subject; + var eventType = _eventTypes.GetOrDefault(eventName); + if (eventType == null) + { + return; + } + + var eventData = _serializer.Deserialize(message.Body, eventType); + + await TriggerHandlersAsync(eventType, eventData); + } + + public IDisposable Subscribe(IDistributedEventHandler handler) where TEvent : class + { + return Subscribe(typeof(TEvent), handler); + } + + public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) + { + var handlerFactories = GetOrCreateHandlerFactories(eventType); + + if (factory.IsInFactories(handlerFactories)) + { + return NullDisposable.Instance; + } + + handlerFactories.Add(factory); + + return new EventHandlerFactoryUnregistrar(this, eventType, factory); + } + + public override void Unsubscribe(Func action) + { + Check.NotNull(action, nameof(action)); + + GetOrCreateHandlerFactories(typeof(TEvent)) + .Locking(factories => + { + factories.RemoveAll( + factory => + { + var singleInstanceFactory = factory as SingleInstanceHandlerFactory; + if (singleInstanceFactory == null) + { + return false; + } + + var actionHandler = singleInstanceFactory.HandlerInstance as ActionEventHandler; + if (actionHandler == null) + { + return false; + } + + return actionHandler.Action == action; + }); + }); + } + + public override void Unsubscribe(Type eventType, IEventHandler handler) + { + GetOrCreateHandlerFactories(eventType) + .Locking(factories => + { + factories.RemoveAll( + factory => + factory is SingleInstanceHandlerFactory handlerFactory && + handlerFactory.HandlerInstance == handler + ); + }); + } + + public override void Unsubscribe(Type eventType, IEventHandlerFactory factory) + { + GetOrCreateHandlerFactories(eventType) + .Locking(factories => factories.Remove(factory)); + } + + public override void UnsubscribeAll(Type eventType) + { + GetOrCreateHandlerFactories(eventType) + .Locking(factories => factories.Clear()); + } + + public override async Task PublishAsync(Type eventType, object eventData) + { + var eventName = EventNameAttribute.GetNameOrDefault(eventType); + var body = _serializer.Serialize(eventData); + + var message = new ServiceBusMessage(body) + { + Subject = eventName + }; + + var publisher = await _publisherPool.GetAsync( + _options.TopicName, + _options.ConnectionName); + + await publisher.SendMessageAsync(message); + } + + protected override IEnumerable GetHandlerFactories(Type eventType) + { + return _handlerFactories + .Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key)) + .Select(handlerFactory => + new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)) + .ToArray(); + } + + private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType) + { + return handlerEventType == targetEventType || handlerEventType.IsAssignableFrom(targetEventType); + } + + private List GetOrCreateHandlerFactories(Type eventType) + { + return _handlerFactories.GetOrAdd( + eventType, + type => + { + var eventName = EventNameAttribute.GetNameOrDefault(type); + _eventTypes[eventName] = type; + return new List(); + } + ); + } + } +} \ No newline at end of file From e6a80c0380c0b54f8a7cac11c6196a6d096ed2e6 Mon Sep 17 00:00:00 2001 From: Gideon de Swardt Date: Sat, 3 Apr 2021 23:19:24 +0100 Subject: [PATCH 3/8] Add documentation Added documentation on how to configure and setup the Azure Service Bus as the Distributed Event Bus. --- ...Distributed-Event-Bus-Azure-Integration.md | 133 ++++++++++++++++++ docs/en/Distributed-Event-Bus.md | 1 + docs/en/docs-nav.json | 4 + 3 files changed, 138 insertions(+) create mode 100644 docs/en/Distributed-Event-Bus-Azure-Integration.md diff --git a/docs/en/Distributed-Event-Bus-Azure-Integration.md b/docs/en/Distributed-Event-Bus-Azure-Integration.md new file mode 100644 index 0000000000..ca66cbca8a --- /dev/null +++ b/docs/en/Distributed-Event-Bus-Azure-Integration.md @@ -0,0 +1,133 @@ +# Distributed Event Bus Azure Integration + +> This document explains **how to configure the [Azure Service Bus](https://azure.microsoft.com/en-us/services/service-bus/)** as the distributed event bus provider. See the [distributed event bus document](Distributed-Event-Bus.md) to learn how to use the distributed event bus system + +## Installation + +Use the ABP CLI to add [Volo.Abp.EventBus.Azure](https://www.nuget.org/packages/Volo.Abp.EventBus.Azure) NuGet package to your project: + +* Install the [ABP CLI](https://docs.abp.io/en/abp/latest/CLI) if you haven't installed before. +* Open a command line (terminal) in the directory of the `.csproj` file you want to add the `Volo.Abp.EventBus.Azure` package. +* Run `abp add-package Volo.Abp.EventBus.Azure` command. + +If you want to do it manually, install the [Volo.Abp.EventBus.Azure](https://www.nuget.org/packages/Volo.Abp.EventBus.Azure) NuGet package to your project and add `[DependsOn(typeof(AbpEventBusAzureModule))]` to the [ABP module](Module-Development-Basics.md) class inside your project. + +## Configuration + +You can configure using the standard [configuration system](Configuration.md), like using the `appsettings.json` file, or using the [options](Options.md) classes. + +### `appsettings.json` file configuration + +This is the simplest way to configure the Azure Service Bus settings. It is also very strong since you can use any other configuration source (like environment variables) that is [supported by the AspNet Core](https://docs.microsoft.com/en-us/aspnet/core/fundamentals/configuration/). + +**Example: The minimal configuration to connect to Azure Service Bus Namespace with default configurations** + +````json +{ + "Azure": { + "ServiceBus": { + "Connections": { + "Default": { + "ConnectionString": "Endpoint=sb://sb-my-app.servicebus.windows.net/;SharedAccessKeyName={{Policy Name}};SharedAccessKey={};EntityPath=marketing-consent" + } + } + }, + "EventBus": { + "ConnectionName": "Default", + "SubscriberName": "MySubscriberName", + "TopicName": "MyTopicName" + } + } +} +```` + +* `MySubscriberName` is the name of this subscription, which is used as the **Subscriber** on the Azure Service Bus. +* `MyTopicName` is the **topic name**. + +See [the Azure Service Bus document](https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-queues-topics-subscriptions) to understand these options better. + +#### Connections + +If you need to connect to another Azure Service Bus Namespace the Default, you need to configure the connection properties. + +**Example: Declare two connections and use one of them for the event bus** + +````json +{ + "Azure": { + "ServiceBus": { + "Connections": { + "Default": { + "ConnectionString": "Endpoint=sb://sb-my-app.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey={{SharedAccessKey}}" + }, + "SecondConnection": { + "ConnectionString": "Endpoint=sb://sb-my-app.servicebus.windows.net/;SharedAccessKeyName={{Policy Name}};SharedAccessKey={{SharedAccessKey}}" + } + } + }, + "EventBus": { + "ConnectionName": "SecondConnection", + "SubscriberName": "MySubscriberName", + "TopicName": "MyTopicName" + } + } +} +```` + +This allows you to use multiple Azure Service Bus namespaces in your application, but select one of them for the event bus. + +You can use any of the [ServiceBusAdministrationClientOptions](https://docs.microsoft.com/en-us/dotnet/api/azure.messaging.servicebus.administration.servicebusadministrationclientoptions?view=azure-dotnet), [ServiceBusClientOptions](https://docs.microsoft.com/en-us/dotnet/api/azure.messaging.servicebus.servicebusclientoptions?view=azure-dotnet), [ServiceBusProcessorOptions](https://docs.microsoft.com/en-us/dotnet/api/azure.messaging.servicebus.servicebusprocessoroptions?view=azure-dotnet) properties for the connection. + +**Example: Specify the Admin, Client and Processor options** + +````json +{ + "Azure": { + "ServiceBus": { + "Connections": { + "Default": { + "ConnectionString": "Endpoint=sb://sb-my-app.servicebus.windows.net/;SharedAccessKeyName={{Policy Name}};SharedAccessKey={};EntityPath=marketing-consent", + "Admin": { + "Retry": { + "MaxRetries": 3 + } + }, + "Client": { + "RetryOptions": { + "MaxRetries": 1 + } + }, + "Processor": { + "AutoCompleteMessages": true, + "ReceiveMode": "ReceiveAndDelete" + } + } + } + }, + "EventBus": { + "ConnectionName": "Default", + "SubscriberName": "MySubscriberName", + "TopicName": "MyTopicName" + } + } +} +```` + +### The Options Classes + +`AbpAzureServiceBusOptions` and `AbpAzureEventBusOptions` classes can be used to configure the connection strings and event bus options for Azure Service Bus. + +You can configure this options inside the `ConfigureServices` of your [module](Module-Development-Basics.md). + +**Example: Configure the connection** + +````csharp +Configure(options => +{ + options.Connections.Default.ConnectionString = "Endpoint=sb://sb-my-app.servicebus.windows.net/;SharedAccessKeyName={{Policy Name}};SharedAccessKey={}"; + options.Connections.Default.Admin.Retry.MaxRetries = 3; + options.Connections.Default.Client.RetryOptions.MaxRetries = 1; +}); +```` + +Using these options classes can be combined with the `appsettings.json` way. Configuring an option property in the code overrides the value in the configuration file. diff --git a/docs/en/Distributed-Event-Bus.md b/docs/en/Distributed-Event-Bus.md index 02216c7ac8..70d5fbe391 100644 --- a/docs/en/Distributed-Event-Bus.md +++ b/docs/en/Distributed-Event-Bus.md @@ -7,6 +7,7 @@ Distributed Event bus system allows to **publish** and **subscribe** to events t Distributed event bus system provides an **abstraction** that can be implemented by any vendor/provider. There are two providers implemented out of the box: * `LocalDistributedEventBus` is the default implementation that implements the distributed event bus to work as in-process. Yes! The **default implementation works just like the [local event bus](Local-Event-Bus.md)**, if you don't configure a real distributed provider. +* `AzureDistributedEventBus` implements the distributed event bus with the [Azure Service Bus](https://azure.microsoft.com/en-us/services/service-bus/). See the [Azure Service Bus integration document](Distributed-Event-Bus-Azure-Integration.md) to learn how to configure it. * `RabbitMqDistributedEventBus` implements the distributed event bus with the [RabbitMQ](https://www.rabbitmq.com/). See the [RabbitMQ integration document](Distributed-Event-Bus-RabbitMQ-Integration.md) to learn how to configure it. * `KafkaDistributedEventBus` implements the distributed event bus with the [Kafka](https://kafka.apache.org/). See the [Kafka integration document](Distributed-Event-Bus-Kafka-Integration.md) to learn how to configure it. * `RebusDistributedEventBus` implements the distributed event bus with the [Rebus](http://mookid.dk/category/rebus/). See the [Rebus integration document](Distributed-Event-Bus-Rebus-Integration.md) to learn how to configure it. diff --git a/docs/en/docs-nav.json b/docs/en/docs-nav.json index efca73a484..77c1476735 100644 --- a/docs/en/docs-nav.json +++ b/docs/en/docs-nav.json @@ -296,6 +296,10 @@ "text": "Distributed Event Bus", "path": "Distributed-Event-Bus.md", "items": [ + { + "text": "Azure Service Bus Integration", + "path": "Distributed-Event-Bus-Azure-Integration.md" + }, { "text": "RabbitMQ Integration", "path": "Distributed-Event-Bus-RabbitMQ-Integration.md" From 159bc23f2d45fd6526bbdd55d913ccf325555260 Mon Sep 17 00:00:00 2001 From: maliming Date: Tue, 13 Jul 2021 13:29:59 +0800 Subject: [PATCH 4/8] Update AzureDistributedEventBus.cs --- .../Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index 9b0527ce69..456284d1e8 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -26,16 +26,16 @@ namespace Volo.Abp.EventBus.Azure private readonly ConcurrentDictionary> _handlerFactories; private readonly ConcurrentDictionary _eventTypes; private IAzureServiceBusMessageConsumer _consumer; - public AzureDistributedEventBus( IServiceScopeFactory serviceScopeFactory, ICurrentTenant currentTenant, + IEventErrorHandler errorHandler, IOptions abpAzureEventBusOptions, IOptions abpDistributedEventBusOptions, IAzureServiceBusSerializer serializer, IAzureServiceBusMessageConsumerFactory messageConsumerFactory, IPublisherPool publisherPool) - : base(serviceScopeFactory, currentTenant) + : base(serviceScopeFactory, currentTenant, errorHandler) { _options = abpAzureEventBusOptions.Value; _distributedEventBusOptions = abpDistributedEventBusOptions.Value; @@ -186,4 +186,4 @@ namespace Volo.Abp.EventBus.Azure ); } } -} \ No newline at end of file +} From ae7c867679e0ca6f1edf8e42dab7fa3a02577f88 Mon Sep 17 00:00:00 2001 From: maliming Date: Tue, 13 Jul 2021 14:19:25 +0800 Subject: [PATCH 5/8] Add packages to common.ps1 --- .../Volo.Abp.AzureServiceBus.csproj | 10 ++++++++-- .../Volo.Abp.EventBus.Azure.csproj | 11 ++++++++++- nupkg/common.ps1 | 2 ++ 3 files changed, 20 insertions(+), 3 deletions(-) diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo.Abp.AzureServiceBus.csproj b/framework/src/Volo.Abp.AzureServiceBus/Volo.Abp.AzureServiceBus.csproj index 77ccf8f772..51cab2b9bf 100644 --- a/framework/src/Volo.Abp.AzureServiceBus/Volo.Abp.AzureServiceBus.csproj +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo.Abp.AzureServiceBus.csproj @@ -2,10 +2,16 @@ - + netstandard2.0 - latest + Volo.Abp.AzureServiceBus + Volo.Abp.AzureServiceBus + $(AssetTargetFallback);portable-net45+win8+wp8+wpa81; + false + false + false + diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo.Abp.EventBus.Azure.csproj b/framework/src/Volo.Abp.EventBus.Azure/Volo.Abp.EventBus.Azure.csproj index c913b79bfc..2ffd853cdc 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo.Abp.EventBus.Azure.csproj +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo.Abp.EventBus.Azure.csproj @@ -1,8 +1,17 @@ + + + netstandard2.0 - latest + Volo.Abp.EventBus.Azure + Volo.Abp.EventBus.Azure + $(AssetTargetFallback);portable-net45+win8+wp8+wpa81; + false + false + false + diff --git a/nupkg/common.ps1 b/nupkg/common.ps1 index b71ebccfd0..1b07380f42 100644 --- a/nupkg/common.ps1 +++ b/nupkg/common.ps1 @@ -64,6 +64,7 @@ $projects = ( "framework/src/Volo.Abp.Autofac", "framework/src/Volo.Abp.Autofac.WebAssembly", "framework/src/Volo.Abp.AutoMapper", + "framework/src/Volo.Abp.AzureServiceBus", "framework/src/Volo.Abp.BackgroundJobs.Abstractions", "framework/src/Volo.Abp.BackgroundJobs", "framework/src/Volo.Abp.BackgroundJobs.HangFire", @@ -100,6 +101,7 @@ $projects = ( "framework/src/Volo.Abp.EntityFrameworkCore.SqlServer", "framework/src/Volo.Abp.EventBus.Abstractions", "framework/src/Volo.Abp.EventBus", + "framework/src/Volo.Abp.EventBus.Azure", "framework/src/Volo.Abp.EventBus.RabbitMQ", "framework/src/Volo.Abp.EventBus.Kafka", "framework/src/Volo.Abp.EventBus.Rebus", From 0b62e07236f167b51c588294dc1b91e13ffa412b Mon Sep 17 00:00:00 2001 From: maliming Date: Mon, 4 Oct 2021 11:01:47 +0800 Subject: [PATCH 6/8] Inherit `DistributedEventBusBase`. --- .../IAzureServiceBusSerializer.cs | 6 +- .../Utf8JsonAzureServiceBusSerializer.cs | 11 ++- .../Azure/AzureDistributedEventBus.cs | 67 +++++++++++++++---- 3 files changed, 67 insertions(+), 17 deletions(-) diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusSerializer.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusSerializer.cs index 04f56d4470..61fc9e5ca0 100644 --- a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusSerializer.cs +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/IAzureServiceBusSerializer.cs @@ -6,6 +6,8 @@ namespace Volo.Abp.AzureServiceBus { byte[] Serialize(object obj); - object Deserialize(BinaryData value, Type type); + object Deserialize(byte[] value, Type type); + + T Deserialize(byte[] value); } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/Utf8JsonAzureServiceBusSerializer.cs b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/Utf8JsonAzureServiceBusSerializer.cs index 0a87e22934..d8d6fb82a2 100644 --- a/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/Utf8JsonAzureServiceBusSerializer.cs +++ b/framework/src/Volo.Abp.AzureServiceBus/Volo/Abp/AzureServiceBus/Utf8JsonAzureServiceBusSerializer.cs @@ -19,9 +19,14 @@ namespace Volo.Abp.AzureServiceBus return Encoding.UTF8.GetBytes(_jsonSerializer.Serialize(obj)); } - public object Deserialize(BinaryData value, Type type) + public object Deserialize(byte[] value, Type type) { - return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value.ToArray())); + return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value)); + } + + public T Deserialize(byte[] value) + { + return _jsonSerializer.Deserialize(Encoding.UTF8.GetString(value)); } } -} \ No newline at end of file +} diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index 456284d1e8..f4fbca13b4 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -9,36 +9,47 @@ using Microsoft.Extensions.Options; using Volo.Abp.DependencyInjection; using Volo.Abp.EventBus.Distributed; using Volo.Abp.AzureServiceBus; +using Volo.Abp.Guids; using Volo.Abp.MultiTenancy; using Volo.Abp.Threading; +using Volo.Abp.Timing; +using Volo.Abp.Uow; namespace Volo.Abp.EventBus.Azure { [Dependency(ReplaceServices = true)] [ExposeServices(typeof(IDistributedEventBus), typeof(AzureDistributedEventBus))] - public class AzureDistributedEventBus : EventBusBase, IDistributedEventBus, ISingletonDependency + public class AzureDistributedEventBus : DistributedEventBusBase, IDistributedEventBus, ISingletonDependency { private readonly AbpAzureEventBusOptions _options; - private readonly AbpDistributedEventBusOptions _distributedEventBusOptions; private readonly IAzureServiceBusMessageConsumerFactory _messageConsumerFactory; private readonly IPublisherPool _publisherPool; private readonly IAzureServiceBusSerializer _serializer; private readonly ConcurrentDictionary> _handlerFactories; private readonly ConcurrentDictionary _eventTypes; private IAzureServiceBusMessageConsumer _consumer; + public AzureDistributedEventBus( IServiceScopeFactory serviceScopeFactory, ICurrentTenant currentTenant, + IUnitOfWorkManager unitOfWorkManager, IEventErrorHandler errorHandler, - IOptions abpAzureEventBusOptions, IOptions abpDistributedEventBusOptions, + IGuidGenerator guidGenerator, + IClock clock, + IOptions abpAzureEventBusOptions, IAzureServiceBusSerializer serializer, IAzureServiceBusMessageConsumerFactory messageConsumerFactory, IPublisherPool publisherPool) - : base(serviceScopeFactory, currentTenant, errorHandler) + : base(serviceScopeFactory, + currentTenant, + unitOfWorkManager, + errorHandler, + abpDistributedEventBusOptions, + guidGenerator, + clock) { _options = abpAzureEventBusOptions.Value; - _distributedEventBusOptions = abpDistributedEventBusOptions.Value; _serializer = serializer; _messageConsumerFactory = messageConsumerFactory; _publisherPool = publisherPool; @@ -54,7 +65,7 @@ namespace Volo.Abp.EventBus.Azure _options.ConnectionName); _consumer.OnMessageReceived(ProcessEventAsync); - SubscribeHandlers(_distributedEventBusOptions.Handlers); + SubscribeHandlers(AbpDistributedEventBusOptions.Handlers); } private async Task ProcessEventAsync(ServiceBusReceivedMessage message) @@ -66,14 +77,36 @@ namespace Volo.Abp.EventBus.Azure return; } - var eventData = _serializer.Deserialize(message.Body, eventType); + var eventData = _serializer.Deserialize(message.Body.ToArray(), eventType); await TriggerHandlersAsync(eventType, eventData); } - public IDisposable Subscribe(IDistributedEventHandler handler) where TEvent : class + public override async Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) + { + await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData); + } + + public override async Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) + { + var eventType = _eventTypes.GetOrDefault(incomingEvent.EventName); + if (eventType == null) + { + return; + } + + var eventData = _serializer.Deserialize(incomingEvent.EventData, eventType); + var exceptions = new List(); + await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig); + if (exceptions.Any()) + { + ThrowOriginalExceptions(eventType, exceptions); + } + } + + protected override byte[] Serialize(object eventData) { - return Subscribe(typeof(TEvent), handler); + return _serializer.Serialize(eventData); } public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) @@ -142,16 +175,25 @@ namespace Volo.Abp.EventBus.Azure .Locking(factories => factories.Clear()); } - public override async Task PublishAsync(Type eventType, object eventData) + protected override async Task PublishToEventBusAsync(Type eventType, object eventData) + { + await PublishAsync(eventType, eventData); + } + + protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) + { + unitOfWork.AddOrReplaceDistributedEvent(eventRecord); + } + + protected virtual async Task PublishAsync(string eventName, object eventData) { - var eventName = EventNameAttribute.GetNameOrDefault(eventType); var body = _serializer.Serialize(eventData); var message = new ServiceBusMessage(body) { Subject = eventName }; - + var publisher = await _publisherPool.GetAsync( _options.TopicName, _options.ConnectionName); @@ -185,5 +227,6 @@ namespace Volo.Abp.EventBus.Azure } ); } + } } From c39ee0b6a29e814ff3c1125319f4d5d9ce50108e Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Tue, 12 Oct 2021 13:25:17 +0800 Subject: [PATCH 7/8] Update AzureDistributedEventBus.cs --- .../Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index f4fbca13b4..91f20072a2 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -77,6 +77,11 @@ namespace Volo.Abp.EventBus.Azure return; } + if (await AddToInboxAsync(message.MessageId, eventName, eventType, message.Body.ToArray())) + { + return; + } + var eventData = _serializer.Deserialize(message.Body.ToArray(), eventType); await TriggerHandlersAsync(eventType, eventData); From 8415b969faf2a6334e8f5866f05b742b2400b1d3 Mon Sep 17 00:00:00 2001 From: liangshiwei Date: Tue, 12 Oct 2021 13:34:45 +0800 Subject: [PATCH 8/8] Update AzureDistributedEventBus.cs --- .../Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index 91f20072a2..a719e0a8d5 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -19,7 +19,7 @@ namespace Volo.Abp.EventBus.Azure { [Dependency(ReplaceServices = true)] [ExposeServices(typeof(IDistributedEventBus), typeof(AzureDistributedEventBus))] - public class AzureDistributedEventBus : DistributedEventBusBase, IDistributedEventBus, ISingletonDependency + public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDependency { private readonly AbpAzureEventBusOptions _options; private readonly IAzureServiceBusMessageConsumerFactory _messageConsumerFactory; @@ -33,7 +33,6 @@ namespace Volo.Abp.EventBus.Azure IServiceScopeFactory serviceScopeFactory, ICurrentTenant currentTenant, IUnitOfWorkManager unitOfWorkManager, - IEventErrorHandler errorHandler, IOptions abpDistributedEventBusOptions, IGuidGenerator guidGenerator, IClock clock, @@ -44,7 +43,6 @@ namespace Volo.Abp.EventBus.Azure : base(serviceScopeFactory, currentTenant, unitOfWorkManager, - errorHandler, abpDistributedEventBusOptions, guidGenerator, clock) @@ -232,6 +230,5 @@ namespace Volo.Abp.EventBus.Azure } ); } - } }