mirror of https://github.com/abpframework/abp.git
Halil ibrahim Kalkan
8 years ago
4 changed files with 242 additions and 247 deletions
@ -0,0 +1,212 @@ |
|||
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); |
|||
|
|||
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() |
|||
{ |
|||
} |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue