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