Browse Source

Merge branch 'dev' into cmskit-widget-render

pull/13082/head
malik masis 4 years ago
parent
commit
529ac58a6c
  1. 17
      framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs
  2. 5
      framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs

17
framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs

@ -6,6 +6,7 @@ using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using Rebus.Bus; using Rebus.Bus;
using Rebus.Messages;
using Rebus.Pipeline; using Rebus.Pipeline;
using Rebus.Transport; using Rebus.Transport;
using Volo.Abp.DependencyInjection; using Volo.Abp.DependencyInjection;
@ -154,7 +155,11 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
await PublishAsync(eventType, eventData); await PublishAsync(eventType, eventData);
} }
protected virtual async Task PublishAsync(Type eventType, object eventData) protected virtual async Task PublishAsync(
Type eventType,
object eventData,
Guid? eventId = null,
Dictionary<string, string> headersArguments = null)
{ {
if (AbpRebusEventBusOptions.Publish != null) if (AbpRebusEventBusOptions.Publish != null)
{ {
@ -162,7 +167,13 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
return; return;
} }
await Rebus.Publish(eventData); headersArguments ??= new Dictionary<string, string>();
if (!headersArguments.ContainsKey(Headers.MessageId))
{
headersArguments[Headers.MessageId] = (eventId ?? GuidGenerator.Create()).ToString("N");
}
await Rebus.Publish(eventData, headersArguments);
} }
protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord)
@ -221,7 +232,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName); var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName);
var eventData = Serializer.Deserialize(outgoingEvent.EventData, eventType); var eventData = Serializer.Deserialize(outgoingEvent.EventData, eventType);
return PublishToEventBusAsync(eventType, eventData); return PublishAsync(eventType, eventData, eventId: outgoingEvent.Id);
} }
public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig) public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)

5
framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs

@ -1,5 +1,6 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks; using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
@ -102,7 +103,7 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
return false; return false;
} }
foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes.Values) foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes.Values.OrderBy(x => x.Selector is null))
{ {
if (outboxConfig.Selector == null || outboxConfig.Selector(eventType)) if (outboxConfig.Selector == null || outboxConfig.Selector(eventType))
{ {
@ -137,7 +138,7 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
using (var scope = ServiceScopeFactory.CreateScope()) using (var scope = ServiceScopeFactory.CreateScope())
{ {
foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values) foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values.OrderBy(x => x.EventSelector is null))
{ {
if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType)) if (inboxConfig.EventSelector == null || inboxConfig.EventSelector(eventType))
{ {

Loading…
Cancel
Save