Browse Source

Add event to all Inbox/Outbox.

pull/21965/head
maliming 2 years ago
parent
commit
95bd271c53
No known key found for this signature in database GPG Key ID: A646B9CB645ECEA4
  1. 11
      framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs

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

@ -117,6 +117,8 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
return false; return false;
} }
var addedToOutbox = false;
foreach (var outboxConfig in AbpDistributedEventBusOptions.Outboxes.Values.OrderBy(x => x.Selector is null)) 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))
@ -140,11 +142,11 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
} }
await eventOutbox.EnqueueAsync(outgoingEventInfo); await eventOutbox.EnqueueAsync(outgoingEventInfo);
return true; addedToOutbox = true;
} }
} }
return false; return addedToOutbox;
} }
protected virtual Task OnAddToOutboxAsync(string eventName, Type eventType, object eventData) protected virtual Task OnAddToOutboxAsync(string eventName, Type eventType, object eventData)
@ -164,6 +166,8 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
return false; return false;
} }
var addToInbox = false;
using (var scope = ServiceScopeFactory.CreateScope()) using (var scope = ServiceScopeFactory.CreateScope())
{ {
foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values.OrderBy(x => x.EventSelector is null)) foreach (var inboxConfig in AbpDistributedEventBusOptions.Inboxes.Values.OrderBy(x => x.EventSelector is null))
@ -190,11 +194,12 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB
); );
incomingEventInfo.SetCorrelationId(correlationId!); incomingEventInfo.SetCorrelationId(correlationId!);
await eventInbox.EnqueueAsync(incomingEventInfo); await eventInbox.EnqueueAsync(incomingEventInfo);
addToInbox = true;
} }
} }
} }
return true; return addToInbox;
} }
protected abstract byte[] Serialize(object eventData); protected abstract byte[] Serialize(object eventData);

Loading…
Cancel
Save