Browse Source

Public entity events always in the same unit of work while the uow is being completed.

pull/9909/head
Halil İbrahim Kalkan 5 years ago
parent
commit
39ac36e6af
  1. 186
      framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/EntityChangeEventHelper.cs
  2. 6
      framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/IEntityChangeEventHelper.cs
  3. 15
      framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/NullEntityChangeEventHelper.cs
  4. 40
      framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/UnitOfWorkEventPublisher.cs
  5. 6
      framework/src/Volo.Abp.MemoryDb/Volo/Abp/Domain/Repositories/MemoryDb/MemoryDbRepository.cs
  6. 6
      framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs
  7. 14
      framework/src/Volo.Abp.Uow/Volo/Abp/Uow/ChildUnitOfWork.cs
  8. 12
      framework/src/Volo.Abp.Uow/Volo/Abp/Uow/IUnitOfWork.cs
  9. 4
      framework/src/Volo.Abp.Uow/Volo/Abp/Uow/IUnitOfWorkEventPublisher.cs
  10. 4
      framework/src/Volo.Abp.Uow/Volo/Abp/Uow/NullUnitOfWorkEventPublisher.cs
  11. 40
      framework/src/Volo.Abp.Uow/Volo/Abp/Uow/UnitOfWork.cs
  12. 25
      framework/src/Volo.Abp.Uow/Volo/Abp/Uow/UnitOfWorkEventRecord.cs

186
framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/EntityChangeEventHelper.cs

@ -1,6 +1,5 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
@ -21,6 +20,8 @@ namespace Volo.Abp.Domain.Entities.Events
/// </summary>
public class EntityChangeEventHelper : IEntityChangeEventHelper, ITransientDependency
{
private const string UnitOfWorkEventRecordEntityPropName = "_Abp_Entity";
public ILogger<EntityChangeEventHelper> Logger { get; set; }
public ILocalEventBus LocalEventBus { get; set; }
public IDistributedEventBus DistributedEventBus { get; set; }
@ -45,14 +46,9 @@ namespace Volo.Abp.Domain.Entities.Events
public async Task TriggerEventsAsync(EntityChangeReport changeReport)
{
await TriggerEventsInternalAsync(changeReport);
if (changeReport.IsEmpty() || UnitOfWorkManager.Current == null)
{
return;
}
await UnitOfWorkManager.Current.SaveChangesAsync();
await TriggerEntityChangeEvents(changeReport.ChangedEntities);
await TriggerLocalEvents(changeReport.DomainEvents);
await TriggerDistributedEvents(changeReport.DistributedEvents);
}
public virtual async Task TriggerEntityCreatingEventAsync(object entity)
@ -61,19 +57,17 @@ namespace Volo.Abp.Domain.Entities.Events
LocalEventBus,
typeof(EntityCreatingEventData<>),
entity,
entity,
true
entity
);
}
public virtual async Task TriggerEntityCreatedEventOnUowCompletedAsync(object entity)
public virtual async Task TriggerEntityCreatedEventAsync(object entity)
{
await TriggerEventWithEntity(
LocalEventBus,
typeof(EntityCreatedEventData<>),
entity,
entity,
false
entity
);
if (ShouldPublishDistributedEventForEntity(entity))
@ -85,8 +79,7 @@ namespace Volo.Abp.Domain.Entities.Events
DistributedEventBus,
typeof(EntityCreatedEto<>),
eto,
entity,
false
entity
);
}
}
@ -109,19 +102,17 @@ namespace Volo.Abp.Domain.Entities.Events
LocalEventBus,
typeof(EntityUpdatingEventData<>),
entity,
entity,
true
entity
);
}
public virtual async Task TriggerEntityUpdatedEventOnUowCompletedAsync(object entity)
public virtual async Task TriggerEntityUpdatedEventAsync(object entity)
{
await TriggerEventWithEntity(
LocalEventBus,
typeof(EntityUpdatedEventData<>),
entity,
entity,
false
entity
);
if (ShouldPublishDistributedEventForEntity(entity))
@ -133,8 +124,7 @@ namespace Volo.Abp.Domain.Entities.Events
DistributedEventBus,
typeof(EntityUpdatedEto<>),
eto,
entity,
false
entity
);
}
}
@ -146,19 +136,17 @@ namespace Volo.Abp.Domain.Entities.Events
LocalEventBus,
typeof(EntityDeletingEventData<>),
entity,
entity,
true
entity
);
}
public virtual async Task TriggerEntityDeletedEventOnUowCompletedAsync(object entity)
public virtual async Task TriggerEntityDeletedEventAsync(object entity)
{
await TriggerEventWithEntity(
LocalEventBus,
typeof(EntityDeletedEventData<>),
entity,
entity,
false
entity
);
if (ShouldPublishDistributedEventForEntity(entity))
@ -170,20 +158,12 @@ namespace Volo.Abp.Domain.Entities.Events
DistributedEventBus,
typeof(EntityDeletedEto<>),
eto,
entity,
false
entity
);
}
}
}
protected virtual async Task TriggerEventsInternalAsync(EntityChangeReport changeReport)
{
await TriggerEntityChangeEvents(changeReport.ChangedEntities);
await TriggerLocalEvents(changeReport.DomainEvents);
await TriggerDistributedEvents(changeReport.DistributedEvents);
}
protected virtual async Task TriggerEntityChangeEvents(List<EntityChangeEntry> changedEntities)
{
foreach (var changedEntity in changedEntities)
@ -192,15 +172,15 @@ namespace Volo.Abp.Domain.Entities.Events
{
case EntityChangeType.Created:
await TriggerEntityCreatingEventAsync(changedEntity.Entity);
await TriggerEntityCreatedEventOnUowCompletedAsync(changedEntity.Entity);
await TriggerEntityCreatedEventAsync(changedEntity.Entity);
break;
case EntityChangeType.Updated:
await TriggerEntityUpdatingEventAsync(changedEntity.Entity);
await TriggerEntityUpdatedEventOnUowCompletedAsync(changedEntity.Entity);
await TriggerEntityUpdatedEventAsync(changedEntity.Entity);
break;
case EntityChangeType.Deleted:
await TriggerEntityDeletingEventAsync(changedEntity.Entity);
await TriggerEntityDeletedEventOnUowCompletedAsync(changedEntity.Entity);
await TriggerEntityDeletedEventAsync(changedEntity.Entity);
break;
default:
throw new AbpException("Unknown EntityChangeType: " + changedEntity.ChangeType);
@ -229,120 +209,68 @@ namespace Volo.Abp.Domain.Entities.Events
IEventBus eventPublisher,
Type genericEventType,
object entityOrEto,
object originalEntity,
bool triggerInCurrentUnitOfWork)
object originalEntity)
{
var entityType = ProxyHelper.UnProxy(entityOrEto).GetType();
var eventType = genericEventType.MakeGenericType(entityType);
var eventData = Activator.CreateInstance(eventType, entityOrEto);
var currentUow = UnitOfWorkManager.Current;
if (triggerInCurrentUnitOfWork || currentUow == null)
if (currentUow == null)
{
await eventPublisher.PublishAsync(
eventType,
Activator.CreateInstance(eventType, entityOrEto)
eventData
);
return;
}
var eventList = GetEventList(currentUow);
var isFirstEvent = !eventList.Any();
eventList.AddUniqueEvent(eventPublisher, eventType, entityOrEto, originalEntity);
/* Register to OnCompleted if this is the first item.
* Other items will already be in the list once the UOW completes.
var eventRecord = new UnitOfWorkEventRecord(eventType, eventData)
{
Properties =
{
{ UnitOfWorkEventRecordEntityPropName, originalEntity },
}
};
/* We are trying to eliminate same events for the same entity.
* In this way, for example, we don't trigger update event for an entity multiple times
* even if it is updated multiple times in the current UOW.
*/
if (isFirstEvent)
if (eventPublisher == DistributedEventBus)
{
currentUow.OnCompleted(
async () =>
{
foreach (var eventEntry in eventList)
{
try
{
await eventEntry.EventBus.PublishAsync(
eventEntry.EventType,
Activator.CreateInstance(eventEntry.EventType, eventEntry.EntityOrEto)
);
}
catch (Exception ex)
{
Logger.LogError(
$"Caught an exception while publishing the event '{eventType.FullName}' for the entity '{entityOrEto}'");
Logger.LogException(ex);
}
}
}
currentUow.AddOrReplaceDistributedEvent(
eventRecord,
otherRecord => IsSameEntityEventRecord(eventRecord, otherRecord)
);
}
}
private EntityChangeEventList GetEventList(IUnitOfWork currentUow)
{
return (EntityChangeEventList) currentUow.Items.GetOrAdd(
"AbpEntityChangeEventList",
() => new EntityChangeEventList()
);
}
private class EntityChangeEventList : List<EntityChangeEventEntry>
{
public void AddUniqueEvent(IEventBus eventBus, Type eventType, object entityOrEto, object originalEntity)
else
{
var newEntry = new EntityChangeEventEntry(eventBus, eventType, entityOrEto, originalEntity);
//Latest "same" event overrides the previous events.
for (var i = 0; i < Count; i++)
{
if (this[i].IsSameEvent(newEntry))
{
this[i] = newEntry;
return;
}
}
//If this is a "new" event, add to the end
Add(newEntry);
currentUow.AddOrReplaceLocalEvent(
eventRecord,
otherRecord => IsSameEntityEventRecord(eventRecord, otherRecord)
);
}
}
private class EntityChangeEventEntry
public bool IsSameEntityEventRecord(UnitOfWorkEventRecord record1, UnitOfWorkEventRecord record2)
{
public IEventBus EventBus { get; }
public Type EventType { get; }
public object EntityOrEto { get; }
public object OriginalEntity { get; }
public EntityChangeEventEntry(IEventBus eventBus, Type eventType, object entityOrEto, object originalEntity)
if (record1.EventType != record2.EventType)
{
EventType = eventType;
EntityOrEto = entityOrEto;
OriginalEntity = originalEntity;
EventBus = eventBus;
return false;
}
public bool IsSameEvent(EntityChangeEventEntry otherEntry)
{
if (EventBus != otherEntry.EventBus || EventType != otherEntry.EventType)
{
return false;
}
var originalEntityRef = OriginalEntity as IEntity;
var otherOriginalEntityRef = otherEntry.OriginalEntity as IEntity;
if (originalEntityRef == null || otherOriginalEntityRef == null)
{
return false;
}
var record1OriginalEntity = record1.Properties.GetOrDefault(UnitOfWorkEventRecordEntityPropName) as IEntity;
var record2OriginalEntity = record2.Properties.GetOrDefault(UnitOfWorkEventRecordEntityPropName) as IEntity;
return EntityHelper.EntityEquals(originalEntityRef, otherOriginalEntityRef);
if (record1OriginalEntity == null || record2OriginalEntity == null)
{
return false;
}
return EntityHelper.EntityEquals(record1OriginalEntity, record2OriginalEntity);
}
}
}

6
framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/IEntityChangeEventHelper.cs

@ -10,12 +10,12 @@ namespace Volo.Abp.Domain.Entities.Events
Task TriggerEventsAsync(EntityChangeReport changeReport);
Task TriggerEntityCreatingEventAsync(object entity);
Task TriggerEntityCreatedEventOnUowCompletedAsync(object entity);
Task TriggerEntityCreatedEventAsync(object entity);
Task TriggerEntityUpdatingEventAsync(object entity);
Task TriggerEntityUpdatedEventOnUowCompletedAsync(object entity);
Task TriggerEntityUpdatedEventAsync(object entity);
Task TriggerEntityDeletingEventAsync(object entity);
Task TriggerEntityDeletedEventOnUowCompletedAsync(object entity);
Task TriggerEntityDeletedEventAsync(object entity);
}
}

15
framework/src/Volo.Abp.Ddd.Domain/Volo/Abp/Domain/Entities/Events/NullEntityChangeEventHelper.cs

@ -27,11 +27,6 @@ namespace Volo.Abp.Domain.Entities.Events
return Task.CompletedTask;
}
public Task TriggerEntityCreatedEventOnUowCompletedAsync(object entity)
{
return Task.CompletedTask;
}
public Task TriggerEntityUpdatingEventAsync(object entity)
{
return Task.CompletedTask;
@ -42,11 +37,6 @@ namespace Volo.Abp.Domain.Entities.Events
return Task.CompletedTask;
}
public Task TriggerEntityUpdatedEventOnUowCompletedAsync(object entity)
{
return Task.CompletedTask;
}
public Task TriggerEntityDeletingEventAsync(object entity)
{
return Task.CompletedTask;
@ -57,11 +47,6 @@ namespace Volo.Abp.Domain.Entities.Events
return Task.CompletedTask;
}
public Task TriggerEntityDeletedEventOnUowCompletedAsync(object entity)
{
return Task.CompletedTask;
}
public Task TriggerEventsAsync(EntityChangeReport changeReport)
{
return Task.CompletedTask;

40
framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/UnitOfWorkEventPublisher.cs

@ -0,0 +1,40 @@
using System.Collections.Generic;
using System.Threading.Tasks;
using Volo.Abp.DependencyInjection;
using Volo.Abp.EventBus.Distributed;
using Volo.Abp.EventBus.Local;
using Volo.Abp.Uow;
namespace Volo.Abp.EventBus
{
[Dependency(ReplaceServices = true)]
public class UnitOfWorkEventPublisher : IUnitOfWorkEventPublisher, ITransientDependency
{
private readonly ILocalEventBus _localEventBus;
private readonly IDistributedEventBus _distributedEventBus;
public UnitOfWorkEventPublisher(
ILocalEventBus localEventBus,
IDistributedEventBus distributedEventBus)
{
_localEventBus = localEventBus;
_distributedEventBus = distributedEventBus;
}
public async Task PublishLocalEventsAsync(IEnumerable<UnitOfWorkEventRecord> localEvents)
{
foreach (var localEvent in localEvents)
{
await _localEventBus.PublishAsync(localEvent.EventType, localEvent.EventData);
}
}
public async Task PublishDistributedEventsAsync(IEnumerable<UnitOfWorkEventRecord> distributedEvents)
{
foreach (var distributedEvent in distributedEvents)
{
await _distributedEventBus.PublishAsync(distributedEvent.EventType, distributedEvent.EventData);
}
}
}
}

6
framework/src/Volo.Abp.MemoryDb/Volo/Abp/Domain/Repositories/MemoryDb/MemoryDbRepository.cs

@ -145,20 +145,20 @@ namespace Volo.Abp.Domain.Repositories.MemoryDb
protected virtual async Task TriggerEntityCreateEvents(TEntity entity)
{
await EntityChangeEventHelper.TriggerEntityCreatedEventOnUowCompletedAsync(entity);
await EntityChangeEventHelper.TriggerEntityCreatingEventAsync(entity);
await EntityChangeEventHelper.TriggerEntityCreatedEventAsync(entity);
}
protected virtual async Task TriggerEntityUpdateEventsAsync(TEntity entity)
{
await EntityChangeEventHelper.TriggerEntityUpdatedEventOnUowCompletedAsync(entity);
await EntityChangeEventHelper.TriggerEntityUpdatingEventAsync(entity);
await EntityChangeEventHelper.TriggerEntityUpdatedEventAsync(entity);
}
protected virtual async Task TriggerEntityDeleteEventsAsync(TEntity entity)
{
await EntityChangeEventHelper.TriggerEntityDeletedEventOnUowCompletedAsync(entity);
await EntityChangeEventHelper.TriggerEntityDeletingEventAsync(entity);
await EntityChangeEventHelper.TriggerEntityDeletedEventAsync(entity);
}
protected virtual async Task ApplyAbpConceptsForAddedEntityAsync(TEntity entity)

6
framework/src/Volo.Abp.MongoDB/Volo/Abp/Domain/Repositories/MongoDB/MongoDbRepository.cs

@ -579,14 +579,14 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
private async Task TriggerEntityCreateEvents(TEntity entity)
{
await EntityChangeEventHelper.TriggerEntityCreatedEventOnUowCompletedAsync(entity);
await EntityChangeEventHelper.TriggerEntityCreatingEventAsync(entity);
await EntityChangeEventHelper.TriggerEntityCreatedEventAsync(entity);
}
protected virtual async Task TriggerEntityUpdateEventsAsync(TEntity entity)
{
await EntityChangeEventHelper.TriggerEntityUpdatedEventOnUowCompletedAsync(entity);
await EntityChangeEventHelper.TriggerEntityUpdatingEventAsync(entity);
await EntityChangeEventHelper.TriggerEntityUpdatedEventAsync(entity);
}
protected virtual async Task ApplyAbpConceptsForDeletedEntityAsync(TEntity entity)
@ -598,8 +598,8 @@ namespace Volo.Abp.Domain.Repositories.MongoDB
protected virtual async Task TriggerEntityDeleteEventsAsync(TEntity entity)
{
await EntityChangeEventHelper.TriggerEntityDeletedEventOnUowCompletedAsync(entity);
await EntityChangeEventHelper.TriggerEntityDeletingEventAsync(entity);
await EntityChangeEventHelper.TriggerEntityDeletedEventAsync(entity);
}
protected virtual void CheckAndSetId(TEntity entity)

14
framework/src/Volo.Abp.Uow/Volo/Abp/Uow/ChildUnitOfWork.cs

@ -76,16 +76,20 @@ namespace Volo.Abp.Uow
_parent.OnCompleted(handler);
}
public void AddLocalEvent(object eventData)
public void AddOrReplaceLocalEvent(
UnitOfWorkEventRecord eventRecord,
Predicate<UnitOfWorkEventRecord> replacementSelector = null)
{
_parent.AddLocalEvent(eventData);
_parent.AddOrReplaceLocalEvent(eventRecord, replacementSelector);
}
public void AddDistributedEvent(object eventData)
public void AddOrReplaceDistributedEvent(
UnitOfWorkEventRecord eventRecord,
Predicate<UnitOfWorkEventRecord> replacementSelector = null)
{
_parent.AddDistributedEvent(eventData);
_parent.AddOrReplaceDistributedEvent(eventRecord, replacementSelector);
}
public IDatabaseApi FindDatabaseApi(string key)
{
return _parent.FindDatabaseApi(key);

12
framework/src/Volo.Abp.Uow/Volo/Abp/Uow/IUnitOfWork.cs

@ -43,8 +43,14 @@ namespace Volo.Abp.Uow
void OnCompleted(Func<Task> handler);
void AddLocalEvent(object eventData);
void AddDistributedEvent(object eventData);
void AddOrReplaceLocalEvent(
UnitOfWorkEventRecord eventRecord,
Predicate<UnitOfWorkEventRecord> replacementSelector = null
);
void AddOrReplaceDistributedEvent(
UnitOfWorkEventRecord eventRecord,
Predicate<UnitOfWorkEventRecord> replacementSelector = null
);
}
}

4
framework/src/Volo.Abp.Uow/Volo/Abp/Uow/IUnitOfWorkEventPublisher.cs

@ -5,7 +5,7 @@ namespace Volo.Abp.Uow
{
public interface IUnitOfWorkEventPublisher
{
Task PublishLocalEventsAsync(IEnumerable<object> localEvents);
Task PublishDistributedEventsAsync(IEnumerable<object> distributedEvents);
Task PublishLocalEventsAsync(IEnumerable<UnitOfWorkEventRecord> localEvents);
Task PublishDistributedEventsAsync(IEnumerable<UnitOfWorkEventRecord> distributedEvents);
}
}

4
framework/src/Volo.Abp.Uow/Volo/Abp/Uow/NullUnitOfWorkEventPublisher.cs

@ -6,12 +6,12 @@ namespace Volo.Abp.Uow
{
public class NullUnitOfWorkEventPublisher : IUnitOfWorkEventPublisher, ISingletonDependency
{
public Task PublishLocalEventsAsync(IEnumerable<object> localEvents)
public Task PublishLocalEventsAsync(IEnumerable<UnitOfWorkEventRecord> localEvents)
{
return Task.CompletedTask;
}
public Task PublishDistributedEventsAsync(IEnumerable<object> distributedEvents)
public Task PublishDistributedEventsAsync(IEnumerable<UnitOfWorkEventRecord> distributedEvents)
{
return Task.CompletedTask;
}

40
framework/src/Volo.Abp.Uow/Volo/Abp/Uow/UnitOfWork.cs

@ -1,7 +1,6 @@
using System;
using System.Collections.Generic;
using System.Collections.Immutable;
using System.Collections.ObjectModel;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
@ -35,8 +34,8 @@ namespace Volo.Abp.Uow
public string ReservationName { get; set; }
protected List<Func<Task>> CompletedHandlers { get; } = new List<Func<Task>>();
protected ICollection<object> DistributedEvents { get; } = new Collection<object>();
protected ICollection<object> LocalEvents { get; } = new Collection<object>();
protected List<UnitOfWorkEventRecord> DistributedEvents { get; } = new List<UnitOfWorkEventRecord>();
protected List<UnitOfWorkEventRecord> LocalEvents { get; } = new List<UnitOfWorkEventRecord>();
public event EventHandler<UnitOfWorkFailedEventArgs> Failed;
public event EventHandler<UnitOfWorkEventArgs> Disposed;
@ -241,14 +240,41 @@ namespace Volo.Abp.Uow
CompletedHandlers.Add(handler);
}
public virtual void AddLocalEvent(object eventData)
public virtual void AddOrReplaceLocalEvent(
UnitOfWorkEventRecord eventRecord,
Predicate<UnitOfWorkEventRecord> replacementSelector = null)
{
LocalEvents.Add(eventData);
AddOrReplaceEvent(LocalEvents, eventRecord, replacementSelector);
}
public virtual void AddDistributedEvent(object eventData)
public virtual void AddOrReplaceDistributedEvent(
UnitOfWorkEventRecord eventRecord,
Predicate<UnitOfWorkEventRecord> replacementSelector = null)
{
DistributedEvents.Add(eventData);
AddOrReplaceEvent(DistributedEvents, eventRecord, replacementSelector);
}
public virtual void AddOrReplaceEvent(
List<UnitOfWorkEventRecord> eventRecords,
UnitOfWorkEventRecord eventRecord,
Predicate<UnitOfWorkEventRecord> replacementSelector = null)
{
if (replacementSelector == null)
{
eventRecords.Add(eventRecord);
}
else
{
var foundIndex = eventRecords.FindIndex(replacementSelector);
if (foundIndex < 0)
{
eventRecords.Add(eventRecord);
}
else
{
eventRecords[foundIndex] = eventRecord;
}
}
}
protected virtual async Task OnCompletedAsync()

25
framework/src/Volo.Abp.Uow/Volo/Abp/Uow/UnitOfWorkEventRecord.cs

@ -0,0 +1,25 @@
using System;
using System.Collections.Generic;
namespace Volo.Abp.Uow
{
public class UnitOfWorkEventRecord
{
public object EventData { get; }
public Type EventType { get; }
/// <summary>
/// Extra properties can be used if needed.
/// </summary>
public Dictionary<string, object> Properties { get; } = new Dictionary<string, object>();
public UnitOfWorkEventRecord(
Type eventType,
object eventData)
{
EventType = eventType;
EventData = eventData;
}
}
}
Loading…
Cancel
Save