mirror of https://github.com/abpframework/abp.git
Halil ibrahim Kalkan
8 years ago
39 changed files with 1201 additions and 791 deletions
@ -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 |
|||
{ |
|||
/// <summary>
|
|||
/// Subscribes to an event.
|
|||
/// Given action is called for all event occurrences.
|
|||
/// </summary>
|
|||
/// <param name="action">Action to handle events</param>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
IDisposable Subscribe<TEvent>(Func<TEvent, Task> action) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Subscribes to an event.
|
|||
/// Same (given) instance of the handler is used for all event occurrences.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="handler">Object to handle the event</param>
|
|||
IDisposable Subscribe<TEvent>(IEventHandler<TEvent> handler) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Subscribes to an event.
|
|||
/// A new instance of <see cref="THandler"/> object is created for every event occurrence.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <typeparam name="THandler">Type of the event handler</typeparam>
|
|||
IDisposable Subscribe<TEvent, THandler>() |
|||
where TEvent : class |
|||
where THandler : IEventHandler, new(); |
|||
|
|||
/// <summary>
|
|||
/// Subscribes to an event.
|
|||
/// Same (given) instance of the handler is used for all event occurrences.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="handler">Object to handle the event</param>
|
|||
IDisposable Subscribe(Type eventType, IEventHandler handler); |
|||
|
|||
/// <summary>
|
|||
/// Subscribes to an event.
|
|||
/// Given factory is used to create/release handlers
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="factory">A factory to create/release handlers</param>
|
|||
IDisposable Subscribe<TEvent>(IEventHandlerFactory factory) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Subscribes to an event.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="factory">A factory to create/release handlers</param>
|
|||
IDisposable Subscribe(Type eventType, IEventHandlerFactory factory); |
|||
|
|||
Task PublishAsync<TEvent>(TEvent eventData) |
|||
where TEvent : class; |
|||
|
|||
Task PublishAsync(Type eventType, object eventData); |
|||
|
|||
} |
|||
} |
|||
|
|||
@ -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<TEvent>(Func<TEvent, Task> action) where TEvent : class |
|||
{ |
|||
return _eventBus.Register(action); |
|||
return _localEventBus.Subscribe(action); |
|||
} |
|||
|
|||
public IDisposable Subscribe<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
return _eventBus.Register(handler); |
|||
return _localEventBus.Subscribe(handler); |
|||
} |
|||
|
|||
public IDisposable Subscribe<TEvent, THandler>() where TEvent : class where THandler : IEventHandler, new() |
|||
{ |
|||
return _eventBus.Register<TEvent, THandler>(); |
|||
return _localEventBus.Subscribe<TEvent, THandler>(); |
|||
} |
|||
|
|||
public IDisposable Subscribe(Type eventType, IEventHandler handler) |
|||
{ |
|||
return _eventBus.Register(eventType, handler); |
|||
return _localEventBus.Subscribe(eventType, handler); |
|||
} |
|||
|
|||
public IDisposable Subscribe<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
return _eventBus.Register<TEvent>(factory); |
|||
return _localEventBus.Subscribe<TEvent>(factory); |
|||
} |
|||
|
|||
public IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
return _eventBus.Register(eventType, factory); |
|||
return _localEventBus.Subscribe(eventType, factory); |
|||
} |
|||
|
|||
public void Unsubscribe<TEvent>(Func<TEvent, Task> action) where TEvent : class |
|||
{ |
|||
_localEventBus.Unsubscribe(action); |
|||
} |
|||
|
|||
public void Unsubscribe<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
_localEventBus.Unsubscribe(handler); |
|||
} |
|||
|
|||
public void Unsubscribe(Type eventType, IEventHandler handler) |
|||
{ |
|||
_localEventBus.Unsubscribe(eventType, handler); |
|||
} |
|||
|
|||
public void Unsubscribe<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
_localEventBus.Unsubscribe<TEvent>(factory); |
|||
} |
|||
|
|||
public void Unsubscribe(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
_localEventBus.Unsubscribe(eventType, factory); |
|||
} |
|||
|
|||
public void UnsubscribeAll<TEvent>() where TEvent : class |
|||
{ |
|||
_localEventBus.UnsubscribeAll<TEvent>(); |
|||
} |
|||
|
|||
public void UnsubscribeAll(Type eventType) |
|||
{ |
|||
_localEventBus.UnsubscribeAll(eventType); |
|||
} |
|||
|
|||
public Task PublishAsync<TEvent>(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); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,14 @@ |
|||
using Volo.Abp.Collections; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public class DistributedEventBusOptions |
|||
{ |
|||
public ITypeList<IEventHandler> Handlers { get; } |
|||
|
|||
public DistributedEventBusOptions() |
|||
{ |
|||
Handlers = new TypeList<IEventHandler>(); |
|||
} |
|||
} |
|||
} |
|||
@ -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 |
|||
{ |
|||
/// <summary>
|
|||
/// Implements EventBus as Singleton pattern.
|
|||
/// </summary>
|
|||
public class EventBus : IEventBus, ISingletonDependency |
|||
{ |
|||
/// <summary>
|
|||
/// Reference to the Logger.
|
|||
/// </summary>
|
|||
public ILogger<EventBus> Logger { get; set; } |
|||
|
|||
/// <summary>
|
|||
/// All registered handler factories.
|
|||
/// Key: Type of the event
|
|||
/// Value: List of handler factories
|
|||
/// </summary>
|
|||
protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; } |
|||
|
|||
protected EventBusOptions Options { get; } |
|||
|
|||
public EventBus( |
|||
IOptions<EventBusOptions> options, |
|||
IServiceProvider serviceProvider) |
|||
{ |
|||
Options = options.Value; |
|||
Logger = NullLogger<EventBus>.Instance; |
|||
HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>(); |
|||
|
|||
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)); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public IDisposable Register<TEvent>(Func<TEvent, Task> action) where TEvent : class |
|||
{ |
|||
return Register(typeof(TEvent), new ActionEventHandler<TEvent>(action)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public IDisposable Register<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
return Register(typeof(TEvent), handler); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public IDisposable Register<TEvent, THandler>() |
|||
where TEvent : class |
|||
where THandler : IEventHandler, new() |
|||
{ |
|||
return Register(typeof(TEvent), new TransientEventHandlerFactory<THandler>()); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public IDisposable Register(Type eventType, IEventHandler handler) |
|||
{ |
|||
return Register(eventType, new SingleInstanceHandlerFactory(handler)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public IDisposable Register<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
return Register(typeof(TEvent), factory); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public IDisposable Register(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType) |
|||
.Locking(factories => factories.Add(factory)); |
|||
|
|||
return new EventHandlerFactoryUnregistrar(this, eventType, factory); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public void Unregister<TEvent>(Func<TEvent, Task> 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<TEvent>; |
|||
if (actionHandler == null) |
|||
{ |
|||
return false; |
|||
} |
|||
|
|||
return actionHandler.Action == action; |
|||
}); |
|||
}); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public void Unregister<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
Unregister(typeof(TEvent), handler); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public void Unregister(Type eventType, IEventHandler handler) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType) |
|||
.Locking(factories => |
|||
{ |
|||
factories.RemoveAll( |
|||
factory => |
|||
factory is SingleInstanceHandlerFactory && |
|||
(factory as SingleInstanceHandlerFactory).HandlerInstance == handler |
|||
); |
|||
}); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public void Unregister<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
Unregister(typeof(TEvent), factory); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public void Unregister(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public void UnregisterAll<TEvent>() where TEvent : class |
|||
{ |
|||
UnregisterAll(typeof(TEvent)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public void UnregisterAll(Type eventType) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear()); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public Task TriggerAsync<TEvent>(TEvent eventData) where TEvent : class |
|||
{ |
|||
return TriggerAsync(typeof(TEvent), eventData); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public async Task TriggerAsync(Type eventType, object eventData) |
|||
{ |
|||
var exceptions = new List<Exception>(); |
|||
|
|||
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<Exception> 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<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType) |
|||
{ |
|||
var handlerFactoryList = new List<EventTypeWithEventHandlerFactories>(); |
|||
|
|||
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<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType) |
|||
{ |
|||
return HandlerFactories.GetOrAdd(eventType, (type) => new List<IEventHandlerFactory>()); |
|||
} |
|||
|
|||
private class EventTypeWithEventHandlerFactories |
|||
{ |
|||
public Type EventType { get; } |
|||
|
|||
public List<IEventHandlerFactory> EventHandlerFactories { get; } |
|||
|
|||
public EventTypeWithEventHandlerFactories(Type eventType, List<IEventHandlerFactory> 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() |
|||
{ |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -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 |
|||
{ |
|||
/// <inheritdoc/>
|
|||
public virtual IDisposable Subscribe<TEvent>(Func<TEvent, Task> action) where TEvent : class |
|||
{ |
|||
return Subscribe(typeof(TEvent), new ActionEventHandler<TEvent>(action)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public virtual IDisposable Subscribe<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
return Subscribe(typeof(TEvent), handler); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public virtual IDisposable Subscribe<TEvent, THandler>() |
|||
where TEvent : class |
|||
where THandler : IEventHandler, new() |
|||
{ |
|||
return Subscribe(typeof(TEvent), new TransientEventHandlerFactory<THandler>()); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public virtual IDisposable Subscribe(Type eventType, IEventHandler handler) |
|||
{ |
|||
return Subscribe(eventType, new SingleInstanceHandlerFactory(handler)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public virtual IDisposable Subscribe<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
return Subscribe(typeof(TEvent), factory); |
|||
} |
|||
|
|||
public abstract IDisposable Subscribe(Type eventType, IEventHandlerFactory factory); |
|||
|
|||
public abstract void Unsubscribe<TEvent>(Func<TEvent, Task> action) where TEvent : class; |
|||
|
|||
/// <inheritdoc/>
|
|||
public virtual void Unsubscribe<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
Unsubscribe(typeof(TEvent), handler); |
|||
} |
|||
|
|||
public abstract void Unsubscribe(Type eventType, IEventHandler handler); |
|||
|
|||
/// <inheritdoc/>
|
|||
public virtual void Unsubscribe<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
Unsubscribe(typeof(TEvent), factory); |
|||
} |
|||
|
|||
public abstract void Unsubscribe(Type eventType, IEventHandlerFactory factory); |
|||
|
|||
/// <inheritdoc/>
|
|||
public virtual void UnsubscribeAll<TEvent>() where TEvent : class |
|||
{ |
|||
UnsubscribeAll(typeof(TEvent)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public abstract void UnsubscribeAll(Type eventType); |
|||
|
|||
/// <inheritdoc/>
|
|||
public virtual Task PublishAsync<TEvent>(TEvent eventData) where TEvent : class |
|||
{ |
|||
return PublishAsync(typeof(TEvent), eventData); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public abstract Task PublishAsync(Type eventType, object eventData); |
|||
|
|||
public virtual async Task TriggerHandlersAsync(Type eventType, object eventData) |
|||
{ |
|||
var exceptions = new List<Exception>(); |
|||
|
|||
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<Exception> 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<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType); |
|||
|
|||
protected virtual async Task TriggerHandlerAsync(IEventHandlerFactory asyncHandlerFactory, Type eventType, object eventData, List<Exception> 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<object>.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<object>.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<IEventHandlerFactory> EventHandlerFactories { get; } |
|||
|
|||
public EventTypeWithEventHandlerFactories(Type eventType, List<IEventHandlerFactory> 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() |
|||
{ |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -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<TEvent>() |
|||
{ |
|||
return GetName(typeof(TEvent)); |
|||
} |
|||
|
|||
public static string GetName([NotNull] Type eventType) |
|||
{ |
|||
Check.NotNull(eventType, nameof(eventType)); |
|||
|
|||
return eventType |
|||
.GetCustomAttributes(true) |
|||
.OfType<IEventNameProvider>() |
|||
.FirstOrDefault() |
|||
?.Name |
|||
?? eventType.FullName; |
|||
} |
|||
} |
|||
|
|||
public interface IEventNameProvider |
|||
{ |
|||
string Name { get; } |
|||
} |
|||
} |
|||
@ -1,131 +1,7 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus |
|||
{ |
|||
/// <summary>
|
|||
/// Defines interface of the event bus.
|
|||
/// </summary>
|
|||
public interface IEventBus |
|||
public interface IEventBus : IEventSubscriber, IEventPublisher |
|||
{ |
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// Given action is called for all event occurrences.
|
|||
/// </summary>
|
|||
/// <param name="action">Action to handle events</param>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
IDisposable Register<TEvent>(Func<TEvent, Task> action) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// Same (given) instance of the handler is used for all event occurrences.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="handler">Object to handle the event</param>
|
|||
IDisposable Register<TEvent>(IEventHandler<TEvent> handler) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// A new instance of <see cref="THandler"/> object is created for every event occurrence.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <typeparam name="THandler">Type of the event handler</typeparam>
|
|||
IDisposable Register<TEvent, THandler>() |
|||
where TEvent : class |
|||
where THandler : IEventHandler, new(); |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// Same (given) instance of the handler is used for all event occurrences.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="handler">Object to handle the event</param>
|
|||
IDisposable Register(Type eventType, IEventHandler handler); |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// Given factory is used to create/release handlers
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="factory">A factory to create/release handlers</param>
|
|||
IDisposable Register<TEvent>(IEventHandlerFactory factory) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="factory">A factory to create/release handlers</param>
|
|||
IDisposable Register(Type eventType, IEventHandlerFactory factory); |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="action"></param>
|
|||
void Unregister<TEvent>(Func<TEvent, Task> action) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="handler">Handler object that is registered before</param>
|
|||
void Unregister<TEvent>(IEventHandler<TEvent> handler) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="handler">Handler object that is registered before</param>
|
|||
void Unregister(Type eventType, IEventHandler handler); |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="factory">Factory object that is registered before</param>
|
|||
void Unregister<TEvent>(IEventHandlerFactory factory) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="factory">Factory object that is registered before</param>
|
|||
void Unregister(Type eventType, IEventHandlerFactory factory); |
|||
|
|||
/// <summary>
|
|||
/// Unregisters all event handlers of given event type.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
void UnregisterAll<TEvent>() |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Unregisters all event handlers of given event type.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
void UnregisterAll(Type eventType); |
|||
|
|||
/// <summary>
|
|||
/// Triggers an event asynchronously.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="eventData">Related data for the event</param>
|
|||
/// <returns>The task to handle async operation</returns>
|
|||
Task TriggerAsync<TEvent>(TEvent eventData) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Triggers an event asynchronously.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="eventData">Related data for the event</param>
|
|||
/// <returns>The task to handle async operation</returns>
|
|||
Task TriggerAsync(Type eventType, object eventData); |
|||
} |
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus |
|||
{ |
|||
public interface IEventPublisher |
|||
{ |
|||
/// <summary>
|
|||
/// Triggers an event asynchronously.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="eventData">Related data for the event</param>
|
|||
/// <returns>The task to handle async operation</returns>
|
|||
Task PublishAsync<TEvent>(TEvent eventData) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Triggers an event asynchronously.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="eventData">Related data for the event</param>
|
|||
/// <returns>The task to handle async operation</returns>
|
|||
Task PublishAsync(Type eventType, object eventData); |
|||
} |
|||
} |
|||
@ -0,0 +1,111 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus |
|||
{ |
|||
public interface IEventSubscriber |
|||
{ |
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// Given action is called for all event occurrences.
|
|||
/// </summary>
|
|||
/// <param name="action">Action to handle events</param>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
IDisposable Subscribe<TEvent>(Func<TEvent, Task> action) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// Same (given) instance of the handler is used for all event occurrences.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="handler">Object to handle the event</param>
|
|||
IDisposable Subscribe<TEvent>(IEventHandler<TEvent> handler) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// A new instance of <see cref="THandler"/> object is created for every event occurrence.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <typeparam name="THandler">Type of the event handler</typeparam>
|
|||
IDisposable Subscribe<TEvent, THandler>() |
|||
where TEvent : class |
|||
where THandler : IEventHandler, new(); |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// Same (given) instance of the handler is used for all event occurrences.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="handler">Object to handle the event</param>
|
|||
IDisposable Subscribe(Type eventType, IEventHandler handler); |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// Given factory is used to create/release handlers
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="factory">A factory to create/release handlers</param>
|
|||
IDisposable Subscribe<TEvent>(IEventHandlerFactory factory) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Registers to an event.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="factory">A factory to create/release handlers</param>
|
|||
IDisposable Subscribe(Type eventType, IEventHandlerFactory factory); |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="action"></param>
|
|||
void Unsubscribe<TEvent>(Func<TEvent, Task> action) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="handler">Handler object that is registered before</param>
|
|||
void Unsubscribe<TEvent>(IEventHandler<TEvent> handler) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="handler">Handler object that is registered before</param>
|
|||
void Unsubscribe(Type eventType, IEventHandler handler); |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
/// <param name="factory">Factory object that is registered before</param>
|
|||
void Unsubscribe<TEvent>(IEventHandlerFactory factory) |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Unregisters from an event.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
/// <param name="factory">Factory object that is registered before</param>
|
|||
void Unsubscribe(Type eventType, IEventHandlerFactory factory); |
|||
|
|||
/// <summary>
|
|||
/// Unregisters all event handlers of given event type.
|
|||
/// </summary>
|
|||
/// <typeparam name="TEvent">Event type</typeparam>
|
|||
void UnsubscribeAll<TEvent>() |
|||
where TEvent : class; |
|||
|
|||
/// <summary>
|
|||
/// Unregisters all event handlers of given event type.
|
|||
/// </summary>
|
|||
/// <param name="eventType">Event type</param>
|
|||
void UnsubscribeAll(Type eventType); |
|||
} |
|||
} |
|||
@ -0,0 +1,10 @@ |
|||
namespace Volo.Abp.EventBus.Local |
|||
{ |
|||
/// <summary>
|
|||
/// Defines interface of the event bus.
|
|||
/// </summary>
|
|||
public interface ILocalEventBus : IEventBus |
|||
{ |
|||
|
|||
} |
|||
} |
|||
@ -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 |
|||
{ |
|||
/// <summary>
|
|||
/// Implements EventBus as Singleton pattern.
|
|||
/// </summary>
|
|||
[ExposeServices(typeof(ILocalEventBus), typeof(LocalEventBus))] |
|||
public class LocalEventBus : EventBusBase, ILocalEventBus, ISingletonDependency |
|||
{ |
|||
/// <summary>
|
|||
/// Reference to the Logger.
|
|||
/// </summary>
|
|||
public ILogger<LocalEventBus> Logger { get; set; } |
|||
|
|||
protected LocalEventBusOptions Options { get; } |
|||
|
|||
protected ConcurrentDictionary<Type, List<IEventHandlerFactory>> HandlerFactories { get; } |
|||
|
|||
protected IServiceProvider ServiceProvider { get; } |
|||
|
|||
public LocalEventBus( |
|||
IOptions<LocalEventBusOptions> options, |
|||
IServiceProvider serviceProvider) |
|||
{ |
|||
ServiceProvider = serviceProvider; |
|||
Options = options.Value; |
|||
Logger = NullLogger<LocalEventBus>.Instance; |
|||
|
|||
HandlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>(); |
|||
Subscribe(Options.Handlers); |
|||
} |
|||
|
|||
public virtual void Subscribe(ITypeList<IEventHandler> 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)); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType) |
|||
.Locking(factories => factories.Add(factory)); |
|||
|
|||
return new EventHandlerFactoryUnregistrar(this, eventType, factory); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public override void Unsubscribe<TEvent>(Func<TEvent, Task> 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<TEvent>; |
|||
if (actionHandler == null) |
|||
{ |
|||
return false; |
|||
} |
|||
|
|||
return actionHandler.Action == action; |
|||
}); |
|||
}); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public override void Unsubscribe(Type eventType, IEventHandler handler) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType) |
|||
.Locking(factories => |
|||
{ |
|||
factories.RemoveAll( |
|||
factory => |
|||
factory is SingleInstanceHandlerFactory && |
|||
(factory as SingleInstanceHandlerFactory).HandlerInstance == handler |
|||
); |
|||
}); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
public override void Unsubscribe(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory)); |
|||
} |
|||
|
|||
/// <inheritdoc/>
|
|||
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<Exception>(); |
|||
|
|||
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<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType) |
|||
{ |
|||
var handlerFactoryList = new List<EventTypeWithEventHandlerFactories>(); |
|||
|
|||
foreach (var handlerFactory in HandlerFactories.Where(hf => ShouldTriggerEventForHandler(eventType, hf.Key))) |
|||
{ |
|||
handlerFactoryList.Add(new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value)); |
|||
} |
|||
|
|||
return handlerFactoryList.ToArray(); |
|||
} |
|||
|
|||
private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType) |
|||
{ |
|||
return HandlerFactories.GetOrAdd(eventType, (type) => new List<IEventHandlerFactory>()); |
|||
} |
|||
|
|||
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; |
|||
} |
|||
} |
|||
} |
|||
@ -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<IEventHandler> Handlers { get; } |
|||
|
|||
public EventBusOptions() |
|||
public LocalEventBusOptions() |
|||
{ |
|||
Handlers = new TypeList<IEventHandler>(); |
|||
} |
|||
@ -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<TEvent>(Func<TEvent, Task> action) where TEvent : class |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Register<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Register<TEvent, THandler>() where TEvent : class where THandler : IEventHandler, new() |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Register(Type eventType, IEventHandler handler) |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Register<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Register(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public void Unregister<TEvent>(Func<TEvent, Task> action) where TEvent : class |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void Unregister<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void Unregister(Type eventType, IEventHandler handler) |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void Unregister<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void Unregister(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void UnregisterAll<TEvent>() where TEvent : class |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void UnregisterAll(Type eventType) |
|||
{ |
|||
|
|||
} |
|||
|
|||
public Task TriggerAsync<TEvent>(TEvent eventData) where TEvent : class |
|||
{ |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public Task TriggerAsync(Type eventType, object eventData) |
|||
{ |
|||
return Task.CompletedTask; |
|||
} |
|||
} |
|||
} |
|||
@ -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<TEvent>(Func<TEvent, Task> action) where TEvent : class |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Subscribe<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Subscribe<TEvent, THandler>() where TEvent : class where THandler : IEventHandler, new() |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Subscribe(Type eventType, IEventHandler handler) |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Subscribe<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
return NullDisposable.Instance; |
|||
} |
|||
|
|||
public void Unsubscribe<TEvent>(Func<TEvent, Task> action) where TEvent : class |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void Unsubscribe<TEvent>(IEventHandler<TEvent> handler) where TEvent : class |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void Unsubscribe(Type eventType, IEventHandler handler) |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void Unsubscribe<TEvent>(IEventHandlerFactory factory) where TEvent : class |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void Unsubscribe(Type eventType, IEventHandlerFactory factory) |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void UnsubscribeAll<TEvent>() where TEvent : class |
|||
{ |
|||
|
|||
} |
|||
|
|||
public void UnsubscribeAll(Type eventType) |
|||
{ |
|||
|
|||
} |
|||
|
|||
public Task PublishAsync<TEvent>(TEvent eventData) where TEvent : class |
|||
{ |
|||
return Task.CompletedTask; |
|||
} |
|||
|
|||
public Task PublishAsync(Type eventType, object eventData) |
|||
{ |
|||
return Task.CompletedTask; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,17 @@ |
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public abstract class LocalDistributedEventBusTestBase : AbpIntegratedTest<EventBusTestModule> |
|||
{ |
|||
protected LocalDistributedEventBus LocalEventBus; |
|||
|
|||
protected LocalDistributedEventBusTestBase() |
|||
{ |
|||
LocalEventBus = GetRequiredService<LocalDistributedEventBus>(); |
|||
} |
|||
|
|||
protected override void SetAbpApplicationCreationOptions(AbpApplicationCreationOptions options) |
|||
{ |
|||
options.UseAutofac(); |
|||
} |
|||
} |
|||
} |
|||
@ -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<MySimpleEventData, MySimpleDistributedTransientEventHandler>(); |
|||
|
|||
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); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,23 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus.Distributed |
|||
{ |
|||
public class MySimpleDistributedTransientEventHandler : IDistributedEventHandler<MySimpleEventData>, 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; |
|||
} |
|||
} |
|||
} |
|||
@ -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<MySimpleEventData>( |
|||
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<MyDerivedEventData>( |
|||
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); |
|||
} |
|||
} |
|||
} |
|||
@ -1,12 +1,12 @@ |
|||
namespace Volo.Abp.EventBus |
|||
namespace Volo.Abp.EventBus.Local |
|||
{ |
|||
public abstract class EventBusTestBase : AbpIntegratedTest<EventBusTestModule> |
|||
{ |
|||
protected IEventBus EventBus; |
|||
protected ILocalEventBus LocalEventBus; |
|||
|
|||
protected EventBusTestBase() |
|||
{ |
|||
EventBus = GetRequiredService<IEventBus>(); |
|||
LocalEventBus = GetRequiredService<ILocalEventBus>(); |
|||
} |
|||
|
|||
protected override void SetAbpApplicationCreationOptions(AbpApplicationCreationOptions options) |
|||
@ -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<MySimpleEventData>( |
|||
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<MyDerivedEventData>( |
|||
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); |
|||
} |
|||
} |
|||
} |
|||
@ -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<MySimpleEventData>, ISingletonDependency |
|||
{ |
|||
@ -1,7 +1,7 @@ |
|||
using System; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Volo.Abp.EventBus |
|||
namespace Volo.Abp.EventBus.Local |
|||
{ |
|||
public class MySimpleTransientEventHandler : IEventHandler<MySimpleEventData>, IDisposable |
|||
{ |
|||
@ -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<MySimpleEventData, MySimpleTransientEventHandler>(); |
|||
LocalEventBus.Subscribe<MySimpleEventData, MySimpleTransientEventHandler>(); |
|||
|
|||
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); |
|||
Loading…
Reference in new issue