diff --git a/framework/src/Volo.Abp.Core/System/Collections/Generic/AbpCollectionExtensions.cs b/framework/src/Volo.Abp.Core/System/Collections/Generic/AbpCollectionExtensions.cs index 4240d895ba..69b4d795e0 100644 --- a/framework/src/Volo.Abp.Core/System/Collections/Generic/AbpCollectionExtensions.cs +++ b/framework/src/Volo.Abp.Core/System/Collections/Generic/AbpCollectionExtensions.cs @@ -37,6 +37,33 @@ namespace System.Collections.Generic return true; } + /// + /// Adds items to the collection which are not already in the collection. + /// + /// The collection + /// Item to check and add + /// Type of the items in the collection + /// Returns the added items. + public static IEnumerable AddIfNotContains([NotNull] this ICollection source, IEnumerable items) + { + Check.NotNull(source, nameof(source)); + + var addedItems = new List(); + + foreach (var item in items) + { + if (source.Contains(item)) + { + continue; + } + + source.Add(item); + addedItems.Add(item); + } + + return addedItems; + } + /// /// Adds an item to the collection if it's not already in the collection based on the given . /// diff --git a/framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/EntityChangeEventHelper.cs b/framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/EntityChangeEventHelper.cs index 832f511eaa..c557013a03 100644 --- a/framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/EntityChangeEventHelper.cs +++ b/framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/EntityChangeEventHelper.cs @@ -5,6 +5,7 @@ using Volo.Abp.Auditing; using Volo.Abp.DependencyInjection; using Volo.Abp.DynamicProxy; using Volo.Abp.EventBus; +using Volo.Abp.EventBus.Local; using Volo.Abp.Uow; namespace Volo.Abp.Domain.Entities.Events @@ -14,14 +15,14 @@ namespace Volo.Abp.Domain.Entities.Events /// public class EntityChangeEventHelper : IEntityChangeEventHelper, ITransientDependency { - public IEventBus EventBus { get; set; } + public ILocalEventBus LocalEventBus { get; set; } private readonly IUnitOfWorkManager _unitOfWorkManager; public EntityChangeEventHelper(IUnitOfWorkManager unitOfWorkManager) { _unitOfWorkManager = unitOfWorkManager; - EventBus = NullEventBus.Instance; + LocalEventBus = NullLocalEventBus.Instance; } public async Task TriggerEventsAsync(EntityChangeReport changeReport) @@ -115,7 +116,7 @@ namespace Volo.Abp.Domain.Entities.Events { foreach (var domainEvent in domainEvents) { - await EventBus.TriggerAsync(domainEvent.EventData.GetType(), domainEvent.EventData); + await LocalEventBus.PublishAsync(domainEvent.EventData.GetType(), domainEvent.EventData); } } @@ -126,11 +127,11 @@ namespace Volo.Abp.Domain.Entities.Events if (triggerInCurrentUnitOfWork || _unitOfWorkManager.Current == null) { - await EventBus.TriggerAsync(eventType, Activator.CreateInstance(eventType, entity)); + await LocalEventBus.PublishAsync(eventType, Activator.CreateInstance(eventType, entity)); return; } - _unitOfWorkManager.Current.OnCompleted(() => EventBus.TriggerAsync(eventType, Activator.CreateInstance(eventType, entity))); + _unitOfWorkManager.Current.OnCompleted(() => LocalEventBus.PublishAsync(eventType, Activator.CreateInstance(eventType, entity))); } } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBus.cs index 0976be74f4..053b41dc5b 100644 --- a/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBus.cs @@ -1,83 +1,221 @@ using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; using System.Threading.Tasks; using Microsoft.Extensions.Options; using Volo.Abp.DependencyInjection; using Volo.Abp.RabbitMQ; using RabbitMQ.Client; +using RabbitMQ.Client.Events; +using Volo.Abp.Collections; +using Volo.Abp.EventBus.Local; +using Volo.Abp.Threading; namespace Volo.Abp.EventBus.Distributed.RabbitMq { - /* Inspired from the implementation of "eShopOnContainers" + /* TODO: How to handle unsubscribe to unbind on RabbitMq (may not be possible for) * TODO: Implement Retry system * TODO: Should be improved */ - public class RabbitMqDistributedEventBus : IDistributedEventBus, ITransientDependency + [Dependency(ReplaceServices = true)] + [ExposeServices(typeof(IDistributedEventBus), typeof(RabbitMqDistributedEventBus))] + public class RabbitMqDistributedEventBus : EventBusBase, IDistributedEventBus, ISingletonDependency { - protected RabbitMqDistributedEventBusOptions Options { get; } - protected IChannelPool ChannelPool { get; } + protected RabbitMqDistributedEventBusOptions RabbitMqDistributedEventBusOptions { get; } + protected DistributedEventBusOptions DistributedEventBusOptions { get; } + protected IConnectionPool ConnectionPool { get; } protected IRabbitMqSerializer Serializer { get; } + protected ConcurrentDictionary> HandlerFactories { get; } //TODO: Accessing to the List may not be thread-safe! + protected ConcurrentDictionary EventTypes { get; } + protected IModel ConsumerChannel; + protected IServiceProvider ServiceProvider { get; } public RabbitMqDistributedEventBus( IOptions options, - IChannelPool channelPool, - IRabbitMqSerializer serializer) + IConnectionPool connectionPool, + IRabbitMqSerializer serializer, + IServiceProvider serviceProvider, + DistributedEventBusOptions distributedEventBusOptions) { - ChannelPool = channelPool; + ConnectionPool = connectionPool; Serializer = serializer; - Options = options.Value; + ServiceProvider = serviceProvider; + DistributedEventBusOptions = distributedEventBusOptions; + RabbitMqDistributedEventBusOptions = options.Value; + + HandlerFactories = new ConcurrentDictionary>(); + EventTypes = new ConcurrentDictionary(); + + ConsumerChannel = CreateConsumerChannel(); + Subscribe(DistributedEventBusOptions.Handlers); } - public IDisposable Subscribe(Func action) where TEvent : class + public virtual void Subscribe(ITypeList handlers) { - throw new NotImplementedException(); + foreach (var handler in handlers) + { + var interfaces = handler.GetInterfaces(); + foreach (var @interface in interfaces) + { + if (!typeof(IEventHandler).GetTypeInfo().IsAssignableFrom(@interface)) + { + continue; + } + + var genericArgs = @interface.GetGenericArguments(); + if (genericArgs.Length == 1) + { + Subscribe(genericArgs[0], new IocEventHandlerFactory(ServiceProvider, handler)); + } + } + } } - public IDisposable Subscribe(IEventHandler handler) where TEvent : class + private IModel CreateConsumerChannel() { - throw new NotImplementedException(); + //TODO: Support multiple connection (and consumer)? + var channel = ConnectionPool.Get().CreateModel(); + channel.ExchangeDeclare( + exchange: RabbitMqDistributedEventBusOptions.ExchangeName, + type: "direct" + ); + + channel.QueueDeclare( + queue: RabbitMqDistributedEventBusOptions.ClientName, + durable: true, + exclusive: false, + autoDelete: false, + arguments: null + ); + + var consumer = new EventingBasicConsumer(channel); + consumer.Received += async (model, ea) => { await ProcessEventAsync(channel, ea); }; + + channel.BasicConsume( + queue: RabbitMqDistributedEventBusOptions.ClientName, + autoAck: false, + consumer: consumer + ); + + channel.CallbackException += (sender, ea) => + { + ConsumerChannel.Dispose(); + ConsumerChannel = CreateConsumerChannel(); + }; + + return channel; } - public IDisposable Subscribe() where TEvent : class where THandler : IEventHandler, new() + private async Task ProcessEventAsync(IModel channel, BasicDeliverEventArgs ea) { - throw new NotImplementedException(); + var eventName = ea.RoutingKey; + var eventType = EventTypes.GetOrDefault(eventName); + if (eventType == null) + { + return; + } + + var eventData = Serializer.Deserialize(ea.Body, eventType); + + await TriggerHandlersAsync(eventType, eventData); + + channel.BasicAck(ea.DeliveryTag, multiple: false); } - public IDisposable Subscribe(Type eventType, IEventHandler handler) + public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) { - throw new NotImplementedException(); + var handlerFactories = GetOrCreateHandlerFactories(eventType); + + handlerFactories.Add(factory); + + if (handlerFactories.Count == 1) //TODO: Multi-threading! + { + var eventName = EventNameAttribute.GetName(eventType); + + using (var channel = ConnectionPool.Get().CreateModel()) //TODO: Connection name per event! + { + channel.QueueBind( + queue: RabbitMqDistributedEventBusOptions.ClientName, + exchange: RabbitMqDistributedEventBusOptions.ExchangeName, + routingKey: eventName + ); + } + } + + return new EventHandlerFactoryUnregistrar(this, eventType, factory); } - public IDisposable Subscribe(IEventHandlerFactory factory) where TEvent : class + /// + public override void Unsubscribe(Func action) { - throw new NotImplementedException(); + 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 IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) + /// + public override void Unsubscribe(Type eventType, IEventHandler handler) { - throw new NotImplementedException(); + GetOrCreateHandlerFactories(eventType) + .Locking(factories => + { + factories.RemoveAll( + factory => + factory is SingleInstanceHandlerFactory && + (factory as SingleInstanceHandlerFactory).HandlerInstance == handler + ); + }); } - public Task PublishAsync(TEvent eventData) - where TEvent : class + /// + public override void Unsubscribe(Type eventType, IEventHandlerFactory factory) { - return PublishAsync(typeof(TEvent), eventData); + GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory)); } - public Task PublishAsync(Type eventType, object eventData) + /// + public override void UnsubscribeAll(Type eventType) { - var eventName = eventType.FullName; //TODO: Get eventname from an attribute if available + GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear()); + } + + public override Task PublishAsync(Type eventType, object eventData) + { + var eventName = EventNameAttribute.GetName(eventType); var body = Serializer.Serialize(eventData); - using (var channelAccessor = ChannelPool.Acquire(Guid.NewGuid().ToString())) + using (var channel = ConnectionPool.Get().CreateModel()) //TODO: Connection name per event! { //TODO: Other properties like durable? - channelAccessor.Channel.ExchangeDeclare(Options.ExchangeName, ""); + channel.ExchangeDeclare(RabbitMqDistributedEventBusOptions.ExchangeName, ""); - var properties = channelAccessor.Channel.CreateBasicProperties(); + var properties = channel.CreateBasicProperties(); properties.DeliveryMode = 2; //persistent - channelAccessor.Channel.BasicPublish( - exchange: Options.ExchangeName, + channel.BasicPublish( + exchange: RabbitMqDistributedEventBusOptions.ExchangeName, routingKey: eventName, mandatory: true, basicProperties: properties, @@ -87,5 +225,48 @@ namespace Volo.Abp.EventBus.Distributed.RabbitMq return Task.CompletedTask; } + + private List GetOrCreateHandlerFactories(Type eventType) + { + return HandlerFactories.GetOrAdd( + eventType, + type => + { + var eventName = EventNameAttribute.GetName(type); + EventTypes[eventName] = type; + return new List(); + } + ); + } + + protected override IEnumerable GetHandlerFactories(Type eventType) + { + var handlerFactoryList = new List(); + + foreach (var handlerFactory in HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))) + { + handlerFactoryList.Add(new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)); + } + + return handlerFactoryList.ToArray(); + } + + private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType) + { + //Should trigger same type + if (handlerEventType == targetEventType) + { + return true; + } + + //TODO: Support inheritance? But it does not support on subscription to RabbitMq! + //Should trigger for inherited types + if (handlerEventType.IsAssignableFrom(targetEventType)) + { + return true; + } + + return false; + } } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBusOptions.cs b/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBusOptions.cs index a998861db4..fe277f8599 100644 --- a/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBusOptions.cs +++ b/framework/src/Volo.Abp.EventBus.Distributed.RabbitMQ/Volo/Abp/EventBus/Distributed/RabbitMq/RabbitMqDistributedEventBusOptions.cs @@ -2,6 +2,8 @@ { public class RabbitMqDistributedEventBusOptions { + public string ClientName { get; set; } + public string ExchangeName { get; set; } } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusModule.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusModule.cs index 92c15a08d7..7b95c34a3d 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusModule.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/AbpEventBusModule.cs @@ -1,8 +1,11 @@ -using System; +using Microsoft.Extensions.DependencyInjection; +using System; using System.Collections.Generic; using System.Linq; -using Microsoft.Extensions.DependencyInjection; +using Volo.Abp.EventBus.Distributed; +using Volo.Abp.EventBus.Local; using Volo.Abp.Modularity; +using Volo.Abp.Reflection; namespace Volo.Abp.EventBus { @@ -15,22 +18,29 @@ namespace Volo.Abp.EventBus private static void AddEventHandlers(IServiceCollection services) { - var handlers = new List(); + var localHandlers = new List(); + var distributedHandlers = new List(); services.OnRegistred(context => { - if (context.ImplementationType.GetInterfaces().Any(i => typeof(IEventHandler).IsAssignableFrom(i))) + if (ReflectionHelper.IsAssignableToGenericType(context.ImplementationType, typeof(IEventHandler<>))) { - handlers.Add(context.ImplementationType); + localHandlers.Add(context.ImplementationType); + } + else if (ReflectionHelper.IsAssignableToGenericType(context.ImplementationType, typeof(IDistributedEventHandler<>))) + { + distributedHandlers.Add(context.ImplementationType); } }); - services.Configure(options => + services.Configure(options => { - foreach (var handler in handlers) - { - options.Handlers.AddIfNotContains(handler); - } + options.Handlers.AddIfNotContains(localHandlers); + }); + + services.Configure(options => + { + options.Handlers.AddIfNotContains(distributedHandlers); }); } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/ActionEventHandler.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/ActionEventHandler.cs index 32b41ec800..995ce39356 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/ActionEventHandler.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/ActionEventHandler.cs @@ -8,7 +8,7 @@ namespace Volo.Abp.EventBus /// This event handler is an adapter to be able to use an action as implementation. /// /// Event type - internal class ActionEventHandler : + public class ActionEventHandler : IEventHandler, ITransientDependency { diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IDistributedEventBus.cs index 26bbe9cb1c..88cbf721b3 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IDistributedEventBus.cs @@ -1,65 +1,7 @@ -using System; -using System.Threading.Tasks; - -namespace Volo.Abp.EventBus.Distributed +namespace Volo.Abp.EventBus.Distributed { - public interface IDistributedEventBus + public interface IDistributedEventBus : IEventBus { - /// - /// Subscribes to an event. - /// Given action is called for all event occurrences. - /// - /// Action to handle events - /// Event type - IDisposable Subscribe(Func action) - where TEvent : class; - - /// - /// Subscribes to an event. - /// Same (given) instance of the handler is used for all event occurrences. - /// - /// Event type - /// Object to handle the event - IDisposable Subscribe(IEventHandler handler) - where TEvent : class; - - /// - /// Subscribes to an event. - /// A new instance of object is created for every event occurrence. - /// - /// Event type - /// Type of the event handler - IDisposable Subscribe() - where TEvent : class - where THandler : IEventHandler, new(); - - /// - /// Subscribes to an event. - /// Same (given) instance of the handler is used for all event occurrences. - /// - /// Event type - /// Object to handle the event - IDisposable Subscribe(Type eventType, IEventHandler handler); - - /// - /// Subscribes to an event. - /// Given factory is used to create/release handlers - /// - /// Event type - /// A factory to create/release handlers - IDisposable Subscribe(IEventHandlerFactory factory) - where TEvent : class; - - /// - /// Subscribes to an event. - /// - /// Event type - /// A factory to create/release handlers - IDisposable Subscribe(Type eventType, IEventHandlerFactory factory); - - Task PublishAsync(TEvent eventData) - where TEvent : class; - - Task PublishAsync(Type eventType, object eventData); + } } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs index 38c3687e48..ed5b0027cc 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs @@ -1,58 +1,95 @@ using System; using System.Threading.Tasks; using Volo.Abp.DependencyInjection; +using Volo.Abp.EventBus.Local; namespace Volo.Abp.EventBus.Distributed { [Dependency(TryRegister = true)] + [ExposeServices(typeof(IDistributedEventBus), typeof(LocalDistributedEventBus))] public class LocalDistributedEventBus : IDistributedEventBus, ITransientDependency { - private readonly IEventBus _eventBus; + private readonly ILocalEventBus _localEventBus; - public LocalDistributedEventBus(IEventBus eventBus) + public LocalDistributedEventBus(ILocalEventBus localEventBus) { - _eventBus = eventBus; + _localEventBus = localEventBus; } public IDisposable Subscribe(Func action) where TEvent : class { - return _eventBus.Register(action); + return _localEventBus.Subscribe(action); } public IDisposable Subscribe(IEventHandler handler) where TEvent : class { - return _eventBus.Register(handler); + return _localEventBus.Subscribe(handler); } public IDisposable Subscribe() where TEvent : class where THandler : IEventHandler, new() { - return _eventBus.Register(); + return _localEventBus.Subscribe(); } public IDisposable Subscribe(Type eventType, IEventHandler handler) { - return _eventBus.Register(eventType, handler); + return _localEventBus.Subscribe(eventType, handler); } public IDisposable Subscribe(IEventHandlerFactory factory) where TEvent : class { - return _eventBus.Register(factory); + return _localEventBus.Subscribe(factory); } public IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) { - return _eventBus.Register(eventType, factory); + return _localEventBus.Subscribe(eventType, factory); + } + + public void Unsubscribe(Func action) where TEvent : class + { + _localEventBus.Unsubscribe(action); + } + + public void Unsubscribe(IEventHandler handler) where TEvent : class + { + _localEventBus.Unsubscribe(handler); + } + + public void Unsubscribe(Type eventType, IEventHandler handler) + { + _localEventBus.Unsubscribe(eventType, handler); + } + + public void Unsubscribe(IEventHandlerFactory factory) where TEvent : class + { + _localEventBus.Unsubscribe(factory); + } + + public void Unsubscribe(Type eventType, IEventHandlerFactory factory) + { + _localEventBus.Unsubscribe(eventType, factory); + } + + public void UnsubscribeAll() where TEvent : class + { + _localEventBus.UnsubscribeAll(); + } + + public void UnsubscribeAll(Type eventType) + { + _localEventBus.UnsubscribeAll(eventType); } public Task PublishAsync(TEvent eventData) where TEvent : class { - return _eventBus.TriggerAsync(eventData); + return _localEventBus.PublishAsync(eventData); } public Task PublishAsync(Type eventType, object eventData) { - return _eventBus.TriggerAsync(eventType, eventData); + return _localEventBus.PublishAsync(eventType, eventData); } } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalEventBusOptions.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalEventBusOptions.cs new file mode 100644 index 0000000000..9e48bfa158 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalEventBusOptions.cs @@ -0,0 +1,14 @@ +using Volo.Abp.Collections; + +namespace Volo.Abp.EventBus.Distributed +{ + public class DistributedEventBusOptions + { + public ITypeList Handlers { get; } + + public DistributedEventBusOptions() + { + Handlers = new TypeList(); + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBus.cs deleted file mode 100644 index 3c7243c70e..0000000000 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBus.cs +++ /dev/null @@ -1,335 +0,0 @@ -using System; -using System.Collections.Concurrent; -using System.Collections.Generic; -using System.Linq; -using System.Reflection; -using System.Runtime.CompilerServices; -using System.Threading; -using System.Threading.Tasks; -using Microsoft.Extensions.Logging; -using Microsoft.Extensions.Logging.Abstractions; -using Microsoft.Extensions.Options; -using Volo.Abp.DependencyInjection; -using Volo.Abp.Threading; - -namespace Volo.Abp.EventBus -{ - /// - /// Implements EventBus as Singleton pattern. - /// - public class EventBus : IEventBus, ISingletonDependency - { - /// - /// Reference to the Logger. - /// - public ILogger Logger { get; set; } - - /// - /// All registered handler factories. - /// Key: Type of the event - /// Value: List of handler factories - /// - protected ConcurrentDictionary> HandlerFactories { get; } - - protected EventBusOptions Options { get; } - - public EventBus( - IOptions options, - IServiceProvider serviceProvider) - { - Options = options.Value; - Logger = NullLogger.Instance; - HandlerFactories = new ConcurrentDictionary>(); - - RegisterHandlersInOptions(serviceProvider); - } - - protected virtual void RegisterHandlersInOptions(IServiceProvider serviceProvider) - { - foreach (var handler in Options.Handlers) - { - var interfaces = handler.GetInterfaces(); - foreach (var @interface in interfaces) - { - if (!typeof(IEventHandler).GetTypeInfo().IsAssignableFrom(@interface)) - { - continue; - } - - var genericArgs = @interface.GetGenericArguments(); - if (genericArgs.Length == 1) - { - Register(genericArgs[0], new IocEventHandlerFactory(serviceProvider, handler)); - } - } - } - } - - /// - public IDisposable Register(Func action) where TEvent : class - { - return Register(typeof(TEvent), new ActionEventHandler(action)); - } - - /// - public IDisposable Register(IEventHandler handler) where TEvent : class - { - return Register(typeof(TEvent), handler); - } - - /// - public IDisposable Register() - where TEvent : class - where THandler : IEventHandler, new() - { - return Register(typeof(TEvent), new TransientEventHandlerFactory()); - } - - /// - public IDisposable Register(Type eventType, IEventHandler handler) - { - return Register(eventType, new SingleInstanceHandlerFactory(handler)); - } - - /// - public IDisposable Register(IEventHandlerFactory factory) where TEvent : class - { - return Register(typeof(TEvent), factory); - } - - /// - public IDisposable Register(Type eventType, IEventHandlerFactory factory) - { - GetOrCreateHandlerFactories(eventType) - .Locking(factories => factories.Add(factory)); - - return new EventHandlerFactoryUnregistrar(this, eventType, factory); - } - - /// - public void Unregister(Func action) where TEvent : class - { - 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 void Unregister(IEventHandler handler) where TEvent : class - { - Unregister(typeof(TEvent), handler); - } - - /// - public void Unregister(Type eventType, IEventHandler handler) - { - GetOrCreateHandlerFactories(eventType) - .Locking(factories => - { - factories.RemoveAll( - factory => - factory is SingleInstanceHandlerFactory && - (factory as SingleInstanceHandlerFactory).HandlerInstance == handler - ); - }); - } - - /// - public void Unregister(IEventHandlerFactory factory) where TEvent : class - { - Unregister(typeof(TEvent), factory); - } - - /// - public void Unregister(Type eventType, IEventHandlerFactory factory) - { - GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory)); - } - - /// - public void UnregisterAll() where TEvent : class - { - UnregisterAll(typeof(TEvent)); - } - - /// - public void UnregisterAll(Type eventType) - { - GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear()); - } - - /// - public Task TriggerAsync(TEvent eventData) where TEvent : class - { - return TriggerAsync(typeof(TEvent), eventData); - } - - /// - public async Task TriggerAsync(Type eventType, object eventData) - { - var exceptions = new List(); - - await new SynchronizationContextRemover(); - - foreach (var handlerFactories in GetHandlerFactories(eventType)) - { - foreach (var handlerFactory in handlerFactories.EventHandlerFactories) - { - await TriggerAsyncHandlingException(handlerFactory, handlerFactories.EventType, eventData, exceptions); - } - } - - //Implements generic argument inheritance. See classWithInheritableGenericArgument - if (eventType.GetTypeInfo().IsGenericType && - eventType.GetGenericArguments().Length == 1 && - typeof(IEventDataWithInheritableGenericArgument).IsAssignableFrom(eventType)) - { - var genericArg = eventType.GetGenericArguments()[0]; - var baseArg = genericArg.GetTypeInfo().BaseType; - if (baseArg != null) - { - var baseEventType = eventType.GetGenericTypeDefinition().MakeGenericType(baseArg); - var constructorArgs = ((IEventDataWithInheritableGenericArgument)eventData).GetConstructorArgs(); - var baseEventData = Activator.CreateInstance(baseEventType, constructorArgs); - await TriggerAsync(baseEventType, baseEventData); - } - } - - if (exceptions.Any()) - { - if (exceptions.Count == 1) - { - exceptions[0].ReThrow(); - } - - throw new AggregateException("More than one error has occurred while triggering the event: " + eventType, exceptions); - } - } - - private async Task TriggerAsyncHandlingException(IEventHandlerFactory asyncHandlerFactory, Type eventType, object eventData, List exceptions) - { - using (var eventHandlerWrapper = asyncHandlerFactory.GetHandler()) - { - try - { - var asyncHandlerType = typeof(IEventHandler<>).MakeGenericType(eventType); - - var method = asyncHandlerType.GetMethod( - "HandleEventAsync", - new[] { eventType } - ); - - await (Task)method.Invoke(eventHandlerWrapper.EventHandler, new[] { eventData }); - } - catch (TargetInvocationException ex) - { - exceptions.Add(ex.InnerException); - } - catch (Exception ex) - { - exceptions.Add(ex); - } - } - } - - private IEnumerable GetHandlerFactories(Type eventType) - { - var handlerFactoryList = new List(); - - foreach (var handlerFactory in HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))) - { - handlerFactoryList.Add(new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)); - } - - return handlerFactoryList.ToArray(); - } - - private static bool ShouldTriggerEventForHandler(Type eventType, Type handlerType) - { - //Should trigger same type - if (handlerType == eventType) - { - return true; - } - - //Should trigger for inherited types - if (handlerType.IsAssignableFrom(eventType)) - { - return true; - } - - return false; - } - - private List GetOrCreateHandlerFactories(Type eventType) - { - return HandlerFactories.GetOrAdd(eventType, (type) => new List()); - } - - private class EventTypeWithEventHandlerFactories - { - public Type EventType { get; } - - public List EventHandlerFactories { get; } - - public EventTypeWithEventHandlerFactories(Type eventType, List eventHandlerFactories) - { - EventType = eventType; - EventHandlerFactories = eventHandlerFactories; - } - } - - // Reference from - // https://blogs.msdn.microsoft.com/benwilli/2017/02/09/an-alternative-to-configureawaitfalse-everywhere/ - private struct SynchronizationContextRemover : INotifyCompletion - { - public bool IsCompleted - { - get { return SynchronizationContext.Current == null; } - } - - public void OnCompleted(Action continuation) - { - var prevContext = SynchronizationContext.Current; - try - { - SynchronizationContext.SetSynchronizationContext(null); - continuation(); - } - finally - { - SynchronizationContext.SetSynchronizationContext(prevContext); - } - } - - public SynchronizationContextRemover GetAwaiter() - { - return this; - } - - public void GetResult() - { - } - } - } -} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs new file mode 100644 index 0000000000..4c82dbd8eb --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusBase.cs @@ -0,0 +1,225 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; +using System.Runtime.CompilerServices; +using System.Threading; +using System.Threading.Tasks; +using Volo.Abp.EventBus.Distributed; +using Volo.Abp.Reflection; + +namespace Volo.Abp.EventBus.Local +{ + public abstract class EventBusBase : IEventBus + { + /// + public virtual IDisposable Subscribe(Func action) where TEvent : class + { + return Subscribe(typeof(TEvent), new ActionEventHandler(action)); + } + + /// + public virtual IDisposable Subscribe(IEventHandler handler) where TEvent : class + { + return Subscribe(typeof(TEvent), handler); + } + + /// + public virtual IDisposable Subscribe() + where TEvent : class + where THandler : IEventHandler, new() + { + return Subscribe(typeof(TEvent), new TransientEventHandlerFactory()); + } + + /// + public virtual IDisposable Subscribe(Type eventType, IEventHandler handler) + { + return Subscribe(eventType, new SingleInstanceHandlerFactory(handler)); + } + + /// + public virtual IDisposable Subscribe(IEventHandlerFactory factory) where TEvent : class + { + return Subscribe(typeof(TEvent), factory); + } + + public abstract IDisposable Subscribe(Type eventType, IEventHandlerFactory factory); + + public abstract void Unsubscribe(Func action) where TEvent : class; + + /// + public virtual void Unsubscribe(IEventHandler handler) where TEvent : class + { + Unsubscribe(typeof(TEvent), handler); + } + + public abstract void Unsubscribe(Type eventType, IEventHandler handler); + + /// + public virtual void Unsubscribe(IEventHandlerFactory factory) where TEvent : class + { + Unsubscribe(typeof(TEvent), factory); + } + + public abstract void Unsubscribe(Type eventType, IEventHandlerFactory factory); + + /// + public virtual void UnsubscribeAll() where TEvent : class + { + UnsubscribeAll(typeof(TEvent)); + } + + /// + public abstract void UnsubscribeAll(Type eventType); + + /// + public virtual Task PublishAsync(TEvent eventData) where TEvent : class + { + return PublishAsync(typeof(TEvent), eventData); + } + + /// + public abstract Task PublishAsync(Type eventType, object eventData); + + public virtual async Task TriggerHandlersAsync(Type eventType, object eventData) + { + var exceptions = new List(); + + await TriggerHandlersAsync(eventType, eventData, exceptions); + + if (exceptions.Any()) + { + if (exceptions.Count == 1) + { + exceptions[0].ReThrow(); + } + + throw new AggregateException("More than one error has occurred while triggering the event: " + eventType, exceptions); + } + } + + protected virtual async Task TriggerHandlersAsync(Type eventType, object eventData, List exceptions) + { + await new SynchronizationContextRemover(); + + foreach (var handlerFactories in GetHandlerFactories(eventType)) + { + foreach (var handlerFactory in handlerFactories.EventHandlerFactories) + { + await TriggerHandlerAsync(handlerFactory, handlerFactories.EventType, eventData, exceptions); + } + } + + //Implements generic argument inheritance. See IEventDataWithInheritableGenericArgument + if (eventType.GetTypeInfo().IsGenericType && + eventType.GetGenericArguments().Length == 1 && + typeof(IEventDataWithInheritableGenericArgument).IsAssignableFrom(eventType)) + { + var genericArg = eventType.GetGenericArguments()[0]; + var baseArg = genericArg.GetTypeInfo().BaseType; + if (baseArg != null) + { + var baseEventType = eventType.GetGenericTypeDefinition().MakeGenericType(baseArg); + var constructorArgs = ((IEventDataWithInheritableGenericArgument)eventData).GetConstructorArgs(); + var baseEventData = Activator.CreateInstance(baseEventType, constructorArgs); + await PublishAsync(baseEventType, baseEventData); + } + } + } + + protected abstract IEnumerable GetHandlerFactories(Type eventType); + + protected virtual async Task TriggerHandlerAsync(IEventHandlerFactory asyncHandlerFactory, Type eventType, object eventData, List exceptions) + { + using (var eventHandlerWrapper = asyncHandlerFactory.GetHandler()) + { + try + { + var handlerType = eventHandlerWrapper.EventHandler.GetType(); + + if (ReflectionHelper.IsAssignableToGenericType(handlerType, typeof(IEventHandler<>))) + { + var method = typeof(IEventHandler<>) //TODO: to a static field + .MakeGenericType(eventType) + .GetMethod( + nameof(IEventHandler.HandleEventAsync), + new[] { eventType } + ); + + await (Task)method.Invoke(eventHandlerWrapper.EventHandler, new[] { eventData }); + } + else if (ReflectionHelper.IsAssignableToGenericType(handlerType, typeof(IDistributedEventHandler<>))) + { + var method = typeof(IDistributedEventHandler<>) //TODO: to a static field + .MakeGenericType(eventType) + .GetMethod( + nameof(IDistributedEventHandler.HandleEventAsync), + new[] { eventType } + ); + + await (Task)method.Invoke(eventHandlerWrapper.EventHandler, new[] { eventData }); + } + else + { + throw new AbpException("The object instance is not an event handler. Object type: " + handlerType.AssemblyQualifiedName); + } + } + catch (TargetInvocationException ex) + { + exceptions.Add(ex.InnerException); + } + catch (Exception ex) + { + exceptions.Add(ex); + } + } + } + + protected class EventTypeWithEventHandlerFactories + { + public Type EventType { get; } + + public List EventHandlerFactories { get; } + + public EventTypeWithEventHandlerFactories(Type eventType, List eventHandlerFactories) + { + EventType = eventType; + EventHandlerFactories = eventHandlerFactories; + } + } + + // Reference from + // https://blogs.msdn.microsoft.com/benwilli/2017/02/09/an-alternative-to-configureawaitfalse-everywhere/ + protected struct SynchronizationContextRemover : INotifyCompletion + { + public bool IsCompleted + { + get { return SynchronizationContext.Current == null; } + } + + public void OnCompleted(Action continuation) + { + var prevContext = SynchronizationContext.Current; + try + { + SynchronizationContext.SetSynchronizationContext(null); + continuation(); + } + finally + { + SynchronizationContext.SetSynchronizationContext(prevContext); + } + } + + public SynchronizationContextRemover GetAwaiter() + { + return this; + } + + public void GetResult() + { + } + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventHandlerFactoryUnregistrar.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventHandlerFactoryUnregistrar.cs index 671d8e4705..dad858773c 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventHandlerFactoryUnregistrar.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventHandlerFactoryUnregistrar.cs @@ -5,7 +5,7 @@ namespace Volo.Abp.EventBus /// /// Used to unregister a on method. /// - internal class EventHandlerFactoryUnregistrar : IDisposable + public class EventHandlerFactoryUnregistrar : IDisposable { private readonly IEventBus _eventBus; private readonly Type _eventType; @@ -20,7 +20,7 @@ namespace Volo.Abp.EventBus public void Dispose() { - _eventBus.Unregister(_eventType, _factory); + _eventBus.Unsubscribe(_eventType, _factory); } } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventName.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventName.cs new file mode 100644 index 0000000000..3d634c76f5 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventName.cs @@ -0,0 +1,39 @@ +using System; +using System.Linq; +using JetBrains.Annotations; + +namespace Volo.Abp.EventBus +{ + [AttributeUsage(AttributeTargets.Class)] + public class EventNameAttribute : Attribute, IEventNameProvider + { + public string Name { get; } + + public EventNameAttribute([NotNull] string name) + { + Name = Check.NotNullOrWhiteSpace(name, nameof(name)); + } + + public static string GetName() + { + return GetName(typeof(TEvent)); + } + + public static string GetName([NotNull] Type eventType) + { + Check.NotNull(eventType, nameof(eventType)); + + return eventType + .GetCustomAttributes(true) + .OfType() + .FirstOrDefault() + ?.Name + ?? eventType.FullName; + } + } + + public interface IEventNameProvider + { + string Name { get; } + } +} diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventBus.cs index 1ca7c4ba50..a5d1c27a15 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventBus.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventBus.cs @@ -1,131 +1,7 @@ -using System; -using System.Threading.Tasks; - namespace Volo.Abp.EventBus { - /// - /// Defines interface of the event bus. - /// - public interface IEventBus + public interface IEventBus : IEventSubscriber, IEventPublisher { - /// - /// Registers to an event. - /// Given action is called for all event occurrences. - /// - /// Action to handle events - /// Event type - IDisposable Register(Func action) - where TEvent : class; - - /// - /// Registers to an event. - /// Same (given) instance of the handler is used for all event occurrences. - /// - /// Event type - /// Object to handle the event - IDisposable Register(IEventHandler handler) - where TEvent : class; - - /// - /// Registers to an event. - /// A new instance of object is created for every event occurrence. - /// - /// Event type - /// Type of the event handler - IDisposable Register() - where TEvent : class - where THandler : IEventHandler, new(); - - /// - /// Registers to an event. - /// Same (given) instance of the handler is used for all event occurrences. - /// - /// Event type - /// Object to handle the event - IDisposable Register(Type eventType, IEventHandler handler); - - /// - /// Registers to an event. - /// Given factory is used to create/release handlers - /// - /// Event type - /// A factory to create/release handlers - IDisposable Register(IEventHandlerFactory factory) - where TEvent : class; - - /// - /// Registers to an event. - /// - /// Event type - /// A factory to create/release handlers - IDisposable Register(Type eventType, IEventHandlerFactory factory); - - /// - /// Unregisters from an event. - /// - /// Event type - /// - void Unregister(Func action) - where TEvent : class; - - /// - /// Unregisters from an event. - /// - /// Event type - /// Handler object that is registered before - void Unregister(IEventHandler handler) - where TEvent : class; - - /// - /// Unregisters from an event. - /// - /// Event type - /// Handler object that is registered before - void Unregister(Type eventType, IEventHandler handler); - - /// - /// Unregisters from an event. - /// - /// Event type - /// Factory object that is registered before - void Unregister(IEventHandlerFactory factory) - where TEvent : class; - - /// - /// Unregisters from an event. - /// - /// Event type - /// Factory object that is registered before - void Unregister(Type eventType, IEventHandlerFactory factory); - - /// - /// Unregisters all event handlers of given event type. - /// - /// Event type - void UnregisterAll() - where TEvent : class; - - /// - /// Unregisters all event handlers of given event type. - /// - /// Event type - void UnregisterAll(Type eventType); - - /// - /// Triggers an event asynchronously. - /// - /// Event type - /// Related data for the event - /// The task to handle async operation - Task TriggerAsync(TEvent eventData) - where TEvent : class; - /// - /// Triggers an event asynchronously. - /// - /// Event type - /// Related data for the event - /// The task to handle async operation - Task TriggerAsync(Type eventType, object eventData); } } \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventPublisher.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventPublisher.cs new file mode 100644 index 0000000000..0192522304 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventPublisher.cs @@ -0,0 +1,25 @@ +using System; +using System.Threading.Tasks; + +namespace Volo.Abp.EventBus +{ + public interface IEventPublisher + { + /// + /// Triggers an event asynchronously. + /// + /// Event type + /// Related data for the event + /// The task to handle async operation + Task PublishAsync(TEvent eventData) + where TEvent : class; + + /// + /// Triggers an event asynchronously. + /// + /// Event type + /// Related data for the event + /// The task to handle async operation + Task PublishAsync(Type eventType, object eventData); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventSubscriber.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventSubscriber.cs new file mode 100644 index 0000000000..86330433ea --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/IEventSubscriber.cs @@ -0,0 +1,111 @@ +using System; +using System.Threading.Tasks; + +namespace Volo.Abp.EventBus +{ + public interface IEventSubscriber + { + /// + /// Registers to an event. + /// Given action is called for all event occurrences. + /// + /// Action to handle events + /// Event type + IDisposable Subscribe(Func action) + where TEvent : class; + + /// + /// Registers to an event. + /// Same (given) instance of the handler is used for all event occurrences. + /// + /// Event type + /// Object to handle the event + IDisposable Subscribe(IEventHandler handler) + where TEvent : class; + + /// + /// Registers to an event. + /// A new instance of object is created for every event occurrence. + /// + /// Event type + /// Type of the event handler + IDisposable Subscribe() + where TEvent : class + where THandler : IEventHandler, new(); + + /// + /// Registers to an event. + /// Same (given) instance of the handler is used for all event occurrences. + /// + /// Event type + /// Object to handle the event + IDisposable Subscribe(Type eventType, IEventHandler handler); + + /// + /// Registers to an event. + /// Given factory is used to create/release handlers + /// + /// Event type + /// A factory to create/release handlers + IDisposable Subscribe(IEventHandlerFactory factory) + where TEvent : class; + + /// + /// Registers to an event. + /// + /// Event type + /// A factory to create/release handlers + IDisposable Subscribe(Type eventType, IEventHandlerFactory factory); + + /// + /// Unregisters from an event. + /// + /// Event type + /// + void Unsubscribe(Func action) + where TEvent : class; + + /// + /// Unregisters from an event. + /// + /// Event type + /// Handler object that is registered before + void Unsubscribe(IEventHandler handler) + where TEvent : class; + + /// + /// Unregisters from an event. + /// + /// Event type + /// Handler object that is registered before + void Unsubscribe(Type eventType, IEventHandler handler); + + /// + /// Unregisters from an event. + /// + /// Event type + /// Factory object that is registered before + void Unsubscribe(IEventHandlerFactory factory) + where TEvent : class; + + /// + /// Unregisters from an event. + /// + /// Event type + /// Factory object that is registered before + void Unsubscribe(Type eventType, IEventHandlerFactory factory); + + /// + /// Unregisters all event handlers of given event type. + /// + /// Event type + void UnsubscribeAll() + where TEvent : class; + + /// + /// Unregisters all event handlers of given event type. + /// + /// Event type + void UnsubscribeAll(Type eventType); + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/ILocalEventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/ILocalEventBus.cs new file mode 100644 index 0000000000..7e81a363af --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/ILocalEventBus.cs @@ -0,0 +1,10 @@ +namespace Volo.Abp.EventBus.Local +{ + /// + /// Defines interface of the event bus. + /// + public interface ILocalEventBus : IEventBus + { + + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBus.cs new file mode 100644 index 0000000000..c07e31a642 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBus.cs @@ -0,0 +1,180 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Reflection; +using System.Threading.Tasks; +using Volo.Abp.Collections; +using Volo.Abp.DependencyInjection; +using Volo.Abp.Threading; + +namespace Volo.Abp.EventBus.Local +{ + /// + /// Implements EventBus as Singleton pattern. + /// + [ExposeServices(typeof(ILocalEventBus), typeof(LocalEventBus))] + public class LocalEventBus : EventBusBase, ILocalEventBus, ISingletonDependency + { + /// + /// Reference to the Logger. + /// + public ILogger Logger { get; set; } + + protected LocalEventBusOptions Options { get; } + + protected ConcurrentDictionary> HandlerFactories { get; } + + protected IServiceProvider ServiceProvider { get; } + + public LocalEventBus( + IOptions options, + IServiceProvider serviceProvider) + { + ServiceProvider = serviceProvider; + Options = options.Value; + Logger = NullLogger.Instance; + + HandlerFactories = new ConcurrentDictionary>(); + Subscribe(Options.Handlers); + } + + public virtual void Subscribe(ITypeList handlers) + { + foreach (var handler in handlers) + { + var interfaces = handler.GetInterfaces(); + foreach (var @interface in interfaces) + { + if (!typeof(IEventHandler).GetTypeInfo().IsAssignableFrom(@interface)) + { + continue; + } + + var genericArgs = @interface.GetGenericArguments(); + if (genericArgs.Length == 1) + { + Subscribe(genericArgs[0], new IocEventHandlerFactory(ServiceProvider, handler)); + } + } + } + } + + /// + public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) + { + GetOrCreateHandlerFactories(eventType) + .Locking(factories => factories.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 && + (factory as SingleInstanceHandlerFactory).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 exceptions = new List(); + + await TriggerHandlersAsync(eventType, eventData, exceptions); + + if (exceptions.Any()) + { + if (exceptions.Count == 1) + { + exceptions[0].ReThrow(); + } + + throw new AggregateException("More than one error has occurred while triggering the event: " + eventType, exceptions); + } + } + + protected override IEnumerable GetHandlerFactories(Type eventType) + { + var handlerFactoryList = new List(); + + foreach (var handlerFactory in HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))) + { + handlerFactoryList.Add(new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)); + } + + return handlerFactoryList.ToArray(); + } + + private List GetOrCreateHandlerFactories(Type eventType) + { + return HandlerFactories.GetOrAdd(eventType, (type) => new List()); + } + + private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType) + { + //Should trigger same type + if (handlerEventType == targetEventType) + { + return true; + } + + //Should trigger for inherited types + if (handlerEventType.IsAssignableFrom(targetEventType)) + { + return true; + } + + return false; + } + } +} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusOptions.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBusOptions.cs similarity index 61% rename from framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusOptions.cs rename to framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBusOptions.cs index 5486813597..24606a119a 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/EventBusOptions.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Local/LocalEventBusOptions.cs @@ -1,12 +1,12 @@ using Volo.Abp.Collections; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { - public class EventBusOptions + public class LocalEventBusOptions { public ITypeList Handlers { get; } - public EventBusOptions() + public LocalEventBusOptions() { Handlers = new TypeList(); } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/NullEventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/NullEventBus.cs deleted file mode 100644 index ed55f4217f..0000000000 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/NullEventBus.cs +++ /dev/null @@ -1,90 +0,0 @@ -using System; -using System.Threading.Tasks; - -namespace Volo.Abp.EventBus -{ - public sealed class NullEventBus : IEventBus - { - public static NullEventBus Instance { get; } = new NullEventBus(); - - private NullEventBus() - { - - } - - public IDisposable Register(Func action) where TEvent : class - { - return NullDisposable.Instance; - } - - public IDisposable Register(IEventHandler handler) where TEvent : class - { - return NullDisposable.Instance; - } - - public IDisposable Register() where TEvent : class where THandler : IEventHandler, new() - { - return NullDisposable.Instance; - } - - public IDisposable Register(Type eventType, IEventHandler handler) - { - return NullDisposable.Instance; - } - - public IDisposable Register(IEventHandlerFactory factory) where TEvent : class - { - return NullDisposable.Instance; - } - - public IDisposable Register(Type eventType, IEventHandlerFactory factory) - { - return NullDisposable.Instance; - } - - public void Unregister(Func action) where TEvent : class - { - - } - - public void Unregister(IEventHandler handler) where TEvent : class - { - - } - - public void Unregister(Type eventType, IEventHandler handler) - { - - } - - public void Unregister(IEventHandlerFactory factory) where TEvent : class - { - - } - - public void Unregister(Type eventType, IEventHandlerFactory factory) - { - - } - - public void UnregisterAll() where TEvent : class - { - - } - - public void UnregisterAll(Type eventType) - { - - } - - public Task TriggerAsync(TEvent eventData) where TEvent : class - { - return Task.CompletedTask; - } - - public Task TriggerAsync(Type eventType, object eventData) - { - return Task.CompletedTask; - } - } -} diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/NullLocalEventBus.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/NullLocalEventBus.cs new file mode 100644 index 0000000000..f02365d8fa --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/NullLocalEventBus.cs @@ -0,0 +1,91 @@ +using System; +using System.Threading.Tasks; +using Volo.Abp.EventBus.Local; + +namespace Volo.Abp.EventBus +{ + public sealed class NullLocalEventBus : ILocalEventBus + { + public static NullLocalEventBus Instance { get; } = new NullLocalEventBus(); + + private NullLocalEventBus() + { + + } + + public IDisposable Subscribe(Func action) where TEvent : class + { + return NullDisposable.Instance; + } + + public IDisposable Subscribe(IEventHandler handler) where TEvent : class + { + return NullDisposable.Instance; + } + + public IDisposable Subscribe() where TEvent : class where THandler : IEventHandler, new() + { + return NullDisposable.Instance; + } + + public IDisposable Subscribe(Type eventType, IEventHandler handler) + { + return NullDisposable.Instance; + } + + public IDisposable Subscribe(IEventHandlerFactory factory) where TEvent : class + { + return NullDisposable.Instance; + } + + public IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) + { + return NullDisposable.Instance; + } + + public void Unsubscribe(Func action) where TEvent : class + { + + } + + public void Unsubscribe(IEventHandler handler) where TEvent : class + { + + } + + public void Unsubscribe(Type eventType, IEventHandler handler) + { + + } + + public void Unsubscribe(IEventHandlerFactory factory) where TEvent : class + { + + } + + public void Unsubscribe(Type eventType, IEventHandlerFactory factory) + { + + } + + public void UnsubscribeAll() where TEvent : class + { + + } + + public void UnsubscribeAll(Type eventType) + { + + } + + public Task PublishAsync(TEvent eventData) where TEvent : class + { + return Task.CompletedTask; + } + + public Task PublishAsync(Type eventType, object eventData) + { + return Task.CompletedTask; + } + } +} diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/SingleInstanceHandlerFactory.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/SingleInstanceHandlerFactory.cs index 3362084639..4fdd9cce69 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/SingleInstanceHandlerFactory.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/SingleInstanceHandlerFactory.cs @@ -7,7 +7,7 @@ namespace Volo.Abp.EventBus /// /// This class always gets the same single instance of handler. /// - internal class SingleInstanceHandlerFactory : IEventHandlerFactory + public class SingleInstanceHandlerFactory : IEventHandlerFactory { /// /// The event handler instance. diff --git a/framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs b/framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs index b7e2cf2a16..17bc43ad14 100644 --- a/framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs +++ b/framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs @@ -10,6 +10,7 @@ using Volo.Abp.Auditing; using Volo.Abp.Domain.Entities; using Volo.Abp.Domain.Entities.Events; using Volo.Abp.EventBus; +using Volo.Abp.EventBus.Local; using Volo.Abp.Guids; using Volo.Abp.MongoDB; using Volo.Abp.MultiTenancy; @@ -31,7 +32,7 @@ namespace Volo.Abp.Domain.Repositories.MongoDB protected IMongoDbContextProvider DbContextProvider { get; } - public IEventBus EventBus { get; set; } + public ILocalEventBus LocalEventBus { get; set; } public IEntityChangeEventHelper EntityChangeEventHelper { get; set; } @@ -43,7 +44,7 @@ namespace Volo.Abp.Domain.Repositories.MongoDB { DbContextProvider = dbContextProvider; - EventBus = NullEventBus.Instance; + LocalEventBus = NullLocalEventBus.Instance; EntityChangeEventHelper = NullEntityChangeEventHelper.Instance; } @@ -313,7 +314,7 @@ namespace Volo.Abp.Domain.Repositories.MongoDB foreach (var entityEvent in entityEvents) { - await EventBus.TriggerAsync(entityEvent.GetType(), entityEvent); + await LocalEventBus.PublishAsync(entityEvent.GetType(), entityEvent); } generatesDomainEventsEntity.ClearDomainEvents(); diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/LocalDistributedEventBusTestBase.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/LocalDistributedEventBusTestBase.cs new file mode 100644 index 0000000000..45671715a3 --- /dev/null +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/LocalDistributedEventBusTestBase.cs @@ -0,0 +1,17 @@ +namespace Volo.Abp.EventBus.Distributed +{ + public abstract class LocalDistributedEventBusTestBase : AbpIntegratedTest + { + protected LocalDistributedEventBus LocalEventBus; + + protected LocalDistributedEventBusTestBase() + { + LocalEventBus = GetRequiredService(); + } + + protected override void SetAbpApplicationCreationOptions(AbpApplicationCreationOptions options) + { + options.UseAutofac(); + } + } +} \ No newline at end of file diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus_Test.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus_Test.cs new file mode 100644 index 0000000000..24dfbf357f --- /dev/null +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus_Test.cs @@ -0,0 +1,21 @@ +using System.Threading.Tasks; +using Xunit; + +namespace Volo.Abp.EventBus.Distributed +{ + public class LocalDistributedEventBus_Test : LocalDistributedEventBusTestBase + { + [Fact] + public async Task Should_Call_Handler_AndDispose() + { + LocalEventBus.Subscribe(); + + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); + await LocalEventBus.PublishAsync(new MySimpleEventData(3)); + + Assert.Equal(3, MySimpleDistributedTransientEventHandler.HandleCount); + Assert.Equal(3, MySimpleDistributedTransientEventHandler.DisposeCount); + } + } +} diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/MySimpleDistributedTransientEventHandler.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/MySimpleDistributedTransientEventHandler.cs new file mode 100644 index 0000000000..178626373d --- /dev/null +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Distributed/MySimpleDistributedTransientEventHandler.cs @@ -0,0 +1,23 @@ +using System; +using System.Threading.Tasks; + +namespace Volo.Abp.EventBus.Distributed +{ + public class MySimpleDistributedTransientEventHandler : IDistributedEventHandler, IDisposable + { + public static int HandleCount { get; set; } + + public static int DisposeCount { get; set; } + + public Task HandleEventAsync(MySimpleEventData eventData) + { + ++HandleCount; + return Task.CompletedTask; + } + + public void Dispose() + { + ++DisposeCount; + } + } +} \ No newline at end of file diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/InheritanceTest.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/InheritanceTest.cs deleted file mode 100644 index 5168e11054..0000000000 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/InheritanceTest.cs +++ /dev/null @@ -1,48 +0,0 @@ -using System.Threading.Tasks; -using Xunit; - -namespace Volo.Abp.EventBus -{ - public class InheritanceTest : EventBusTestBase - { - [Fact] - public async Task Should_Handle_Events_For_Derived_Classes() - { - var totalData = 0; - - EventBus.Register( - eventData => - { - totalData += eventData.Value; - return Task.CompletedTask; - }); - - await EventBus.TriggerAsync(new MySimpleEventData(1)); //Should handle directly registered class - await EventBus.TriggerAsync(new MySimpleEventData(2)); //Should handle directly registered class - await EventBus.TriggerAsync(new MyDerivedEventData(3)); //Should handle derived class too - await EventBus.TriggerAsync(new MyDerivedEventData(4)); //Should handle derived class too - - Assert.Equal(10, totalData); - } - - [Fact] - public async Task Should_Not_Handle_Events_For_Base_Classes() - { - var totalData = 0; - - EventBus.Register( - eventData => - { - totalData += eventData.Value; - return Task.CompletedTask; - }); - - await EventBus.TriggerAsync(new MySimpleEventData(1)); //Should not handle - await EventBus.TriggerAsync(new MySimpleEventData(2)); //Should not handle - await EventBus.TriggerAsync(new MyDerivedEventData(3)); //Should handle - await EventBus.TriggerAsync(new MyDerivedEventData(4)); //Should handle - - Assert.Equal(7, totalData); - } - } -} \ No newline at end of file diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/ActionBasedEventHandlerTest.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/ActionBasedEventHandlerTest.cs similarity index 51% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/ActionBasedEventHandlerTest.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/ActionBasedEventHandlerTest.cs index c2938af979..6e6bae2fae 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/ActionBasedEventHandlerTest.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/ActionBasedEventHandlerTest.cs @@ -3,7 +3,7 @@ using System.Threading; using System.Threading.Tasks; using Xunit; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public class ActionBasedEventHandlerTest : EventBusTestBase { @@ -12,17 +12,17 @@ namespace Volo.Abp.EventBus { var totalData = 0; - EventBus.Register( + LocalEventBus.Subscribe( eventData => { totalData += eventData.Value; return Task.CompletedTask; }); - await EventBus.TriggerAsync(new MySimpleEventData(1)); - await EventBus.TriggerAsync(new MySimpleEventData(2)); - await EventBus.TriggerAsync(new MySimpleEventData(3)); - await EventBus.TriggerAsync(new MySimpleEventData(4)); + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); + await LocalEventBus.PublishAsync(new MySimpleEventData(3)); + await LocalEventBus.PublishAsync(new MySimpleEventData(4)); Assert.Equal(10, totalData); } @@ -32,17 +32,17 @@ namespace Volo.Abp.EventBus { var totalData = 0; - EventBus.Register( + LocalEventBus.Subscribe( eventData => { totalData += eventData.Value; return Task.CompletedTask; }); - await EventBus.TriggerAsync(typeof(MySimpleEventData), new MySimpleEventData(1)); - await EventBus.TriggerAsync(typeof(MySimpleEventData), new MySimpleEventData(2)); - await EventBus.TriggerAsync(typeof(MySimpleEventData), new MySimpleEventData(3)); - await EventBus.TriggerAsync(typeof(MySimpleEventData), new MySimpleEventData(4)); + await LocalEventBus.PublishAsync(typeof(MySimpleEventData), new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(typeof(MySimpleEventData), new MySimpleEventData(2)); + await LocalEventBus.PublishAsync(typeof(MySimpleEventData), new MySimpleEventData(3)); + await LocalEventBus.PublishAsync(typeof(MySimpleEventData), new MySimpleEventData(4)); Assert.Equal(10, totalData); } @@ -52,20 +52,20 @@ namespace Volo.Abp.EventBus { var totalData = 0; - var registerDisposer = EventBus.Register( + var registerDisposer = LocalEventBus.Subscribe( eventData => { totalData += eventData.Value; return Task.CompletedTask; }); - await EventBus.TriggerAsync(new MySimpleEventData(1)); - await EventBus.TriggerAsync(new MySimpleEventData(2)); - await EventBus.TriggerAsync(new MySimpleEventData(3)); + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); + await LocalEventBus.PublishAsync(new MySimpleEventData(3)); registerDisposer.Dispose(); - await EventBus.TriggerAsync(new MySimpleEventData(4)); + await LocalEventBus.PublishAsync(new MySimpleEventData(4)); Assert.Equal(6, totalData); } @@ -82,15 +82,15 @@ namespace Volo.Abp.EventBus return Task.CompletedTask; }); - EventBus.Register(action); + LocalEventBus.Subscribe(action); - await EventBus.TriggerAsync(new MySimpleEventData(1)); - await EventBus.TriggerAsync(new MySimpleEventData(2)); - await EventBus.TriggerAsync(new MySimpleEventData(3)); + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); + await LocalEventBus.PublishAsync(new MySimpleEventData(3)); - EventBus.Unregister(action); + LocalEventBus.Unsubscribe(action); - await EventBus.TriggerAsync(new MySimpleEventData(4)); + await LocalEventBus.PublishAsync(new MySimpleEventData(4)); Assert.Equal(6, totalData); } @@ -100,7 +100,7 @@ namespace Volo.Abp.EventBus { int totalData = 0; - EventBus.Register( + LocalEventBus.Subscribe( async eventData => { await Task.Delay(20); @@ -108,10 +108,10 @@ namespace Volo.Abp.EventBus await Task.Delay(20); }); - await EventBus.TriggerAsync(new MySimpleEventData(1)); - await EventBus.TriggerAsync(new MySimpleEventData(2)); - await EventBus.TriggerAsync(new MySimpleEventData(3)); - await EventBus.TriggerAsync(new MySimpleEventData(4)); + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); + await LocalEventBus.PublishAsync(new MySimpleEventData(3)); + await LocalEventBus.PublishAsync(new MySimpleEventData(4)); Assert.Equal(10, totalData); } diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBusTestBase.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBusTestBase.cs similarity index 67% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBusTestBase.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBusTestBase.cs index eca0970da4..b3a55c340b 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBusTestBase.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBusTestBase.cs @@ -1,12 +1,12 @@ -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public abstract class EventBusTestBase : AbpIntegratedTest { - protected IEventBus EventBus; + protected ILocalEventBus LocalEventBus; protected EventBusTestBase() { - EventBus = GetRequiredService(); + LocalEventBus = GetRequiredService(); } protected override void SetAbpApplicationCreationOptions(AbpApplicationCreationOptions options) diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_DI_Services_Test.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_DI_Services_Test.cs similarity index 51% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_DI_Services_Test.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_DI_Services_Test.cs index 0567522d68..e66deb0e1c 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_DI_Services_Test.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_DI_Services_Test.cs @@ -2,17 +2,17 @@ using Shouldly; using Xunit; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public class EventBus_DI_Services_Test : EventBusTestBase { [Fact] public async Task Should_Automatically_Register_EventHandlers_From_Services() { - await EventBus.TriggerAsync(new MySimpleEventData(1)); - await EventBus.TriggerAsync(new MySimpleEventData(2)); - await EventBus.TriggerAsync(new MySimpleEventData(3)); - await EventBus.TriggerAsync(new MySimpleEventData(4)); + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); + await LocalEventBus.PublishAsync(new MySimpleEventData(3)); + await LocalEventBus.PublishAsync(new MySimpleEventData(4)); GetRequiredService().TotalData.ShouldBe(10); } diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_Exception_Test.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_Exception_Test.cs similarity index 73% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_Exception_Test.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_Exception_Test.cs index 9613580b6b..446825c697 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_Exception_Test.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_Exception_Test.cs @@ -3,18 +3,18 @@ using System.Threading.Tasks; using Shouldly; using Xunit; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public class EventBus_Exception_Test : EventBusTestBase { [Fact] public async Task Should_Throw_Single_Exception_If_Only_One_Of_Handlers_Fails() { - EventBus.Register(eventData => throw new Exception("This exception is intentionally thrown!")); + LocalEventBus.Subscribe(eventData => throw new Exception("This exception is intentionally thrown!")); var appException = await Assert.ThrowsAsync(async () => { - await EventBus.TriggerAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); }); appException.Message.ShouldBe("This exception is intentionally thrown!"); @@ -23,15 +23,15 @@ namespace Volo.Abp.EventBus [Fact] public async Task Should_Throw_Aggregate_Exception_If_More_Than_One_Of_Handlers_Fail() { - EventBus.Register( + LocalEventBus.Subscribe( eventData => throw new Exception("This exception is intentionally thrown #1!")); - EventBus.Register( + LocalEventBus.Subscribe( eventData => throw new Exception("This exception is intentionally thrown #2!")); var aggrException = await Assert.ThrowsAsync(async () => { - await EventBus.TriggerAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); }); aggrException.InnerExceptions.Count.ShouldBe(2); diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_MultipleHandle_Test.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_MultipleHandle_Test.cs similarity index 82% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_MultipleHandle_Test.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_MultipleHandle_Test.cs index 6892891f90..b8dfe43e17 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/EventBus_MultipleHandle_Test.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/EventBus_MultipleHandle_Test.cs @@ -4,7 +4,7 @@ using Volo.Abp.Domain.Entities; using Volo.Abp.Domain.Entities.Events; using Xunit; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public class EventBus_EntityEvents_Test : EventBusTestBase { @@ -13,10 +13,10 @@ namespace Volo.Abp.EventBus { var handler = new MyEventHandler(); - EventBus.Register>(handler); - EventBus.Register>(handler); + LocalEventBus.Subscribe>(handler); + LocalEventBus.Subscribe>(handler); - await EventBus.TriggerAsync(new EntityCreatedEventData(new MyEntity())); + await LocalEventBus.PublishAsync(new EntityCreatedEventData(new MyEntity())); handler.EntityCreatedEventCount.ShouldBe(1); handler.EntityChangedEventCount.ShouldBe(1); diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/GenericInheritanceTest.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/GenericInheritanceTest.cs similarity index 74% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/GenericInheritanceTest.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/GenericInheritanceTest.cs index 4a3fbccce8..d770a81325 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/GenericInheritanceTest.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/GenericInheritanceTest.cs @@ -4,7 +4,7 @@ using Volo.Abp.Domain.Entities; using Volo.Abp.Domain.Entities.Events; using Xunit; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public class GenericInheritanceTest : EventBusTestBase { @@ -13,7 +13,7 @@ namespace Volo.Abp.EventBus { var triggeredEvent = false; - EventBus.Register>( + LocalEventBus.Subscribe>( eventData => { eventData.Entity.Id.ShouldBe(42); @@ -21,7 +21,7 @@ namespace Volo.Abp.EventBus return Task.CompletedTask; }); - await EventBus.TriggerAsync(new EntityUpdatedEventData(new Person { Id = 42 })); + await LocalEventBus.PublishAsync(new EntityUpdatedEventData(new Person { Id = 42 })); triggeredEvent.ShouldBe(true); } @@ -31,7 +31,7 @@ namespace Volo.Abp.EventBus { var triggeredEvent = false; - EventBus.Register>( + LocalEventBus.Subscribe>( eventData => { eventData.Entity.Id.ShouldBe(42); @@ -39,7 +39,7 @@ namespace Volo.Abp.EventBus return Task.CompletedTask; }); - await EventBus.TriggerAsync(new EntityChangedEventData(new Student { Id = 42 })); + await LocalEventBus.PublishAsync(new EntityChangedEventData(new Student { Id = 42 })); triggeredEvent.ShouldBe(true); } diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/InheritanceTest.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/InheritanceTest.cs new file mode 100644 index 0000000000..c3b1dd4283 --- /dev/null +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/InheritanceTest.cs @@ -0,0 +1,48 @@ +using System.Threading.Tasks; +using Xunit; + +namespace Volo.Abp.EventBus.Local +{ + public class InheritanceTest : EventBusTestBase + { + [Fact] + public async Task Should_Handle_Events_For_Derived_Classes() + { + var totalData = 0; + + LocalEventBus.Subscribe( + eventData => + { + totalData += eventData.Value; + return Task.CompletedTask; + }); + + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); //Should handle directly registered class + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); //Should handle directly registered class + await LocalEventBus.PublishAsync(new MyDerivedEventData(3)); //Should handle derived class too + await LocalEventBus.PublishAsync(new MyDerivedEventData(4)); //Should handle derived class too + + Assert.Equal(10, totalData); + } + + [Fact] + public async Task Should_Not_Handle_Events_For_Base_Classes() + { + var totalData = 0; + + LocalEventBus.Subscribe( + eventData => + { + totalData += eventData.Value; + return Task.CompletedTask; + }); + + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); //Should not handle + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); //Should not handle + await LocalEventBus.PublishAsync(new MyDerivedEventData(3)); //Should handle + await LocalEventBus.PublishAsync(new MyDerivedEventData(4)); //Should handle + + Assert.Equal(7, totalData); + } + } +} \ No newline at end of file diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/MySimpleEventDataHandler.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/MySimpleEventDataHandler.cs similarity index 91% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/MySimpleEventDataHandler.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/MySimpleEventDataHandler.cs index 700a68ef05..ac725f2ae3 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/MySimpleEventDataHandler.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/MySimpleEventDataHandler.cs @@ -1,7 +1,7 @@ using System.Threading.Tasks; using Volo.Abp.DependencyInjection; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public class MySimpleEventDataHandler : IEventHandler, ISingletonDependency { diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/MySimpleTransientEventHandler.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/MySimpleTransientEventHandler.cs similarity index 93% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/MySimpleTransientEventHandler.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/MySimpleTransientEventHandler.cs index be2b98026f..6adf34b51b 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/MySimpleTransientEventHandler.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/MySimpleTransientEventHandler.cs @@ -1,7 +1,7 @@ using System; using System.Threading.Tasks; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public class MySimpleTransientEventHandler : IEventHandler, IDisposable { diff --git a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/TransientDisposableEventHandlerTest.cs b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/TransientDisposableEventHandlerTest.cs similarity index 52% rename from framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/TransientDisposableEventHandlerTest.cs rename to framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/TransientDisposableEventHandlerTest.cs index 3250128ea8..5ab7ed51af 100644 --- a/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/TransientDisposableEventHandlerTest.cs +++ b/framework/test/Volo.Abp.EventBus.Tests/Volo/Abp/EventBus/Local/TransientDisposableEventHandlerTest.cs @@ -1,18 +1,18 @@ using System.Threading.Tasks; using Xunit; -namespace Volo.Abp.EventBus +namespace Volo.Abp.EventBus.Local { public class TransientDisposableEventHandlerTest : EventBusTestBase { [Fact] public async Task Should_Call_Handler_AndDispose() { - EventBus.Register(); + LocalEventBus.Subscribe(); - await EventBus.TriggerAsync(new MySimpleEventData(1)); - await EventBus.TriggerAsync(new MySimpleEventData(2)); - await EventBus.TriggerAsync(new MySimpleEventData(3)); + await LocalEventBus.PublishAsync(new MySimpleEventData(1)); + await LocalEventBus.PublishAsync(new MySimpleEventData(2)); + await LocalEventBus.PublishAsync(new MySimpleEventData(3)); Assert.Equal(3, MySimpleTransientEventHandler.HandleCount); Assert.Equal(3, MySimpleTransientEventHandler.DisposeCount); diff --git a/framework/test/Volo.Abp.TestApp/Volo/Abp/TestApp/Testing/DomainEvents_Tests.cs b/framework/test/Volo.Abp.TestApp/Volo/Abp/TestApp/Testing/DomainEvents_Tests.cs index 38e219677f..ed62e14ecd 100644 --- a/framework/test/Volo.Abp.TestApp/Volo/Abp/TestApp/Testing/DomainEvents_Tests.cs +++ b/framework/test/Volo.Abp.TestApp/Volo/Abp/TestApp/Testing/DomainEvents_Tests.cs @@ -4,6 +4,7 @@ using System.Threading.Tasks; using Shouldly; using Volo.Abp.Domain.Repositories; using Volo.Abp.EventBus; +using Volo.Abp.EventBus.Local; using Volo.Abp.Modularity; using Volo.Abp.TestApp.Domain; using Xunit; @@ -14,12 +15,12 @@ namespace Volo.Abp.TestApp.Testing where TStartupModule : IAbpModule { protected readonly IRepository PersonRepository; - protected readonly IEventBus EventBus; + protected readonly ILocalEventBus LocalEventBus; protected DomainEvents_Tests() { PersonRepository = GetRequiredService>(); - EventBus = GetRequiredService(); + LocalEventBus = GetRequiredService(); } [Fact] @@ -29,7 +30,7 @@ namespace Volo.Abp.TestApp.Testing var isTriggered = false; - EventBus.Register(data => + LocalEventBus.Subscribe(data => { data.OldName.ShouldBe("Douglas"); data.Person.Name.ShouldBe("Douglas-Changed"); diff --git a/framework/test/Volo.Abp.TestApp/Volo/Abp/TestApp/Testing/EntityChangeEvents_Tests.cs b/framework/test/Volo.Abp.TestApp/Volo/Abp/TestApp/Testing/EntityChangeEvents_Tests.cs index a945120a43..0fd6a72a04 100644 --- a/framework/test/Volo.Abp.TestApp/Volo/Abp/TestApp/Testing/EntityChangeEvents_Tests.cs +++ b/framework/test/Volo.Abp.TestApp/Volo/Abp/TestApp/Testing/EntityChangeEvents_Tests.cs @@ -4,6 +4,7 @@ using Shouldly; using Volo.Abp.Domain.Entities.Events; using Volo.Abp.Domain.Repositories; using Volo.Abp.EventBus; +using Volo.Abp.EventBus.Local; using Volo.Abp.Modularity; using Volo.Abp.TestApp.Domain; using Volo.Abp.Uow; @@ -15,12 +16,12 @@ namespace Volo.Abp.TestApp.Testing where TStartupModule : IAbpModule { protected IRepository PersonRepository { get; } - protected IEventBus EventBus { get; } + protected ILocalEventBus LocalEventBus { get; } protected EntityChangeEvents_Tests() { PersonRepository = GetRequiredService>(); - EventBus = GetRequiredService(); + LocalEventBus = GetRequiredService(); } [Fact] @@ -35,7 +36,7 @@ namespace Volo.Abp.TestApp.Testing using (var uow = GetRequiredService().Begin()) { - EventBus.Register>(data => + LocalEventBus.Subscribe>(data => { creatingEventTriggered.ShouldBeFalse(); createdEventTriggered.ShouldBeFalse(); @@ -54,7 +55,7 @@ namespace Volo.Abp.TestApp.Testing return Task.CompletedTask; }); - EventBus.Register>(data => + LocalEventBus.Subscribe>(data => { creatingEventTriggered.ShouldBeTrue(); createdEventTriggered.ShouldBeFalse(); @@ -69,7 +70,7 @@ namespace Volo.Abp.TestApp.Testing return Task.CompletedTask; }); - EventBus.Register>(data => + LocalEventBus.Subscribe>(data => { creatingEventTriggered.ShouldBeTrue(); createdEventTriggered.ShouldBeFalse(); @@ -84,7 +85,7 @@ namespace Volo.Abp.TestApp.Testing return Task.CompletedTask; }); - EventBus.Register>(data => + LocalEventBus.Subscribe>(data => { creatingEventTriggered.ShouldBeTrue(); createdEventTriggered.ShouldBeTrue();