diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs index 5238cdd8bb..4c39b9bf1d 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/AbpRebusEventBusOptions.cs @@ -20,24 +20,13 @@ public class AbpRebusEventBusOptions } private Action _configurer; - [NotNull] - public Func Publish { - get => _publish; - set => _publish = Check.NotNull(value, nameof(value)); - } - private Func _publish; + public Func Publish { get; set; } public AbpRebusEventBusOptions() { - _publish = DefaultPublish; _configurer = DefaultConfigure; } - private async Task DefaultPublish(IBus bus, Type eventType, object eventData) - { - await bus.Advanced.Routing.Send(InputQueueName, eventData); - } - private void DefaultConfigure(RebusConfigurer configure) { configure.Subscriptions(s => s.StoreInMemory()); diff --git a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs index aad844a4d7..fc6a0667af 100644 --- a/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs @@ -148,7 +148,18 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen protected async override Task PublishToEventBusAsync(Type eventType, object eventData) { - await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData); + await PublishAsync(eventType, eventData); + } + + protected virtual async Task PublishAsync(Type eventType, object eventData) + { + if (AbpRebusEventBusOptions.Publish != null) + { + await AbpRebusEventBusOptions.Publish(Rebus, eventType, eventData); + return; + } + + await Rebus.Advanced.Routing.Send(AbpRebusEventBusOptions.InputQueueName, eventData); } protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) @@ -210,9 +221,16 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen return PublishToEventBusAsync(eventType, eventData); } - public override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) + public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) { - throw new NotImplementedException(); + var outgoingEventArray = outgoingEvents.ToArray(); + + foreach (var outgoingEvent in outgoingEventArray) + { + await PublishFromOutboxAsync(outgoingEvent, outboxConfig); + } + + return new MultipleOutgoingEventPublishResult(outgoingEventArray); } public async override Task ProcessFromInboxAsync(