diff --git a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs index 0c45215c8d..d9093e6336 100644 --- a/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs @@ -84,26 +84,27 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen return; } - if (await AddToInboxAsync(message.MessageId, eventName, eventType, message.Body.ToArray())) + var eventBytes = message.Body.ToArray(); + var eventData = _serializer.Deserialize(eventBytes, eventType); + + if (await AddToInboxAsync(message.MessageId, eventName, eventType, eventBytes, eventData)) { return; } - var eventData = _serializer.Deserialize(message.Body.ToArray(), eventType); - await TriggerHandlersDirectAsync(eventType, eventData); } public async override Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) { - await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, outgoingEvent.Id); - await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Outbox, EventName = outgoingEvent.EventName, EventData = outgoingEvent.EventData }); + + await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, outgoingEvent.Id); } public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) @@ -130,12 +131,7 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen throw new AbpException( "The message is too large to fit in the batch. Set AbpEventBusBoxesOptions.OutboxWaitingEventMaxCount to reduce the number"); } - } - await publisher.SendMessagesAsync(messageBatch); - - foreach (var outgoingEvent in outgoingEventArray) - { await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Outbox, @@ -143,6 +139,8 @@ public class AzureDistributedEventBus : DistributedEventBusBase, ISingletonDepen EventData = outgoingEvent.EventData }); } + + await publisher.SendMessagesAsync(messageBatch); } public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) diff --git a/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs index f479370c8e..06e11ac44e 100644 --- a/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs @@ -143,13 +143,14 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend public override async Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) { - await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName))); await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Outbox, EventName = outgoingEvent.EventName, EventData = outgoingEvent.EventData }); + + await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName))); } public override async Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) @@ -158,18 +159,24 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend foreach (var outgoingEvent in outgoingEventArray) { - await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName))); await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Outbox, EventName = outgoingEvent.EventName, EventData = outgoingEvent.EventData }); + await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName))); } } public virtual async Task DaprTriggerHandlersDirectAsync(Type eventType, object eventData) { + // TODO: Implement inbox + // if (await AddToInboxAsync(message.MessageId, EventNameAttribute.GetNameOrDefault(eventType), eventType, message.Body.ToArray())) + // { + // return; + // } + await TriggerHandlersDirectAsync(eventType, eventData); } diff --git a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs index f66f7cf60f..1ab1422aed 100644 --- a/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs @@ -83,14 +83,14 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen } var messageId = message.GetMessageId(); + var eventBytes = message.Value; + var eventData = Serializer.Deserialize(message.Value, eventType); - if (await AddToInboxAsync(messageId, eventName, eventType, message.Value)) + if (await AddToInboxAsync(messageId, eventName, eventType, eventBytes, eventData)) { return; } - var eventData = Serializer.Deserialize(message.Value, eventType); - await TriggerHandlersDirectAsync(eventType, eventData); } @@ -184,6 +184,13 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) { + await TriggerDistributedEventSentAsync(new DistributedEventSent() + { + Source = DistributedEventSource.Outbox, + EventName = outgoingEvent.EventName, + EventData = outgoingEvent.EventData + }); + await PublishAsync( AbpKafkaEventBusOptions.TopicName, outgoingEvent.EventName, @@ -193,13 +200,6 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen { "messageId", System.Text.Encoding.UTF8.GetBytes(outgoingEvent.Id.ToString("N")) } } ); - - await TriggerDistributedEventSentAsync(new DistributedEventSent() - { - Source = DistributedEventSource.Outbox, - EventName = outgoingEvent.EventName, - EventData = outgoingEvent.EventData - }); } public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) @@ -215,6 +215,13 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen { "messageId", System.Text.Encoding.UTF8.GetBytes(messageId)} }; + await TriggerDistributedEventSentAsync(new DistributedEventSent() + { + Source = DistributedEventSource.Outbox, + EventName = outgoingEvent.EventName, + EventData = outgoingEvent.EventData + }); + producer.Produce( AbpKafkaEventBusOptions.TopicName, new Message @@ -223,13 +230,6 @@ public class KafkaDistributedEventBus : DistributedEventBusBase, ISingletonDepen Value = outgoingEvent.EventData, Headers = headers }); - - await TriggerDistributedEventSentAsync(new DistributedEventSent() - { - Source = DistributedEventSource.Outbox, - EventName = outgoingEvent.EventName, - EventData = outgoingEvent.EventData - }); } } diff --git a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs index e085cc2150..b586f44c26 100644 --- a/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs +++ b/framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs @@ -102,14 +102,13 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe } var eventBytes = ea.Body.ToArray(); + var eventData = Serializer.Deserialize(eventBytes, eventType); - if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventBytes)) + if (await AddToInboxAsync(ea.BasicProperties.MessageId, eventName, eventType, eventBytes, eventData)) { return; } - var eventData = Serializer.Deserialize(eventBytes, eventType); - await TriggerHandlersDirectAsync(eventType, eventData); } @@ -200,13 +199,14 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) { - await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Outbox, EventName = outgoingEvent.EventName, EventData = outgoingEvent.EventData }); + + await PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); } public async override Task PublishManyFromOutboxAsync( @@ -220,19 +220,19 @@ public class RabbitMqDistributedEventBus : DistributedEventBusBase, ISingletonDe foreach (var outgoingEvent in outgoingEventArray) { - await PublishAsync( - channel, - outgoingEvent.EventName, - outgoingEvent.EventData, - properties: null, - eventId: outgoingEvent.Id); - await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Outbox, EventName = outgoingEvent.EventName, EventData = outgoingEvent.EventData }); + + await PublishAsync( + channel, + outgoingEvent.EventName, + outgoingEvent.EventData, + properties: null, + eventId: outgoingEvent.Id); } channel.WaitForConfirmsOrDie(); 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 6540ef242a..bbb3866f04 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 @@ -145,7 +145,7 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen var messageId = MessageContext.Current.TransportMessage.GetMessageId(); var eventName = EventNameAttribute.GetNameOrDefault(eventType); - if (await AddToInboxAsync(messageId, eventName, eventType, MessageContext.Current.TransportMessage.Body)) + if (await AddToInboxAsync(messageId, eventName, eventType, MessageContext.Current.TransportMessage.Body, eventData)) { return; } @@ -235,14 +235,14 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName); var eventData = Serializer.Deserialize(outgoingEvent.EventData, eventType); - await PublishAsync(eventType, eventData, eventId: outgoingEvent.Id); - await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Outbox, EventName = outgoingEvent.EventName, EventData = outgoingEvent.EventData }); + + await PublishAsync(eventType, eventData, eventId: outgoingEvent.Id); } public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) @@ -253,14 +253,14 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen { foreach (var outgoingEvent in outgoingEventArray) { - await PublishFromOutboxAsync(outgoingEvent, outboxConfig); - await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Outbox, EventName = outgoingEvent.EventName, EventData = outgoingEvent.EventData }); + + await PublishFromOutboxAsync(outgoingEvent, outboxConfig); } await scope.CompleteAsync(); diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs index d35377db2d..8cca01485d 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs @@ -81,14 +81,14 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB } } - await PublishToEventBusAsync(eventType, eventData); - await TriggerDistributedEventSentAsync(new DistributedEventSent() { Source = DistributedEventSource.Direct, EventName = EventNameAttribute.GetNameOrDefault(eventType), EventData = eventData }); + + await PublishToEventBusAsync(eventType, eventData); } public abstract Task PublishFromOutboxAsync( @@ -120,6 +120,14 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB var eventOutbox = (IEventOutbox)unitOfWork.ServiceProvider.GetRequiredService(outboxConfig.ImplementationType); var eventName = EventNameAttribute.GetNameOrDefault(eventType); + + await TriggerDistributedEventSentAsync(new DistributedEventSent() + { + Source = DistributedEventSource.Direct, + EventName = eventName, + EventData = eventData + }); + await eventOutbox.EnqueueAsync( new OutgoingEventInfo( GuidGenerator.Create(), @@ -139,7 +147,8 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB string messageId, string eventName, Type eventType, - byte[] eventBytes) + byte[] eventBytes, + object eventData) { if (AbpDistributedEventBusOptions.Inboxes.Count <= 0) { @@ -163,6 +172,13 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB } } + await TriggerDistributedEventReceivedAsync(new DistributedEventReceived + { + Source = DistributedEventSource.Direct, + EventName = EventNameAttribute.GetNameOrDefault(eventType), + EventData = eventData + }); + await eventInbox.EnqueueAsync( new IncomingEventInfo( GuidGenerator.Create(), @@ -183,26 +199,26 @@ public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventB protected virtual async Task TriggerHandlersDirectAsync(Type eventType, object eventData) { - await TriggerHandlersAsync(eventType, eventData); - await TriggerDistributedEventReceivedAsync(new DistributedEventReceived { Source = DistributedEventSource.Direct, EventName = EventNameAttribute.GetNameOrDefault(eventType), EventData = eventData }); + + await TriggerHandlersAsync(eventType, eventData); } protected virtual async Task TriggerHandlersFromInboxAsync(Type eventType, object eventData, List exceptions, InboxConfig inboxConfig = null) { - await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig); - await TriggerDistributedEventReceivedAsync(new DistributedEventReceived { Source = DistributedEventSource.Inbox, EventName = EventNameAttribute.GetNameOrDefault(eventType), EventData = eventData }); + + await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig); } public virtual async Task TriggerDistributedEventSentAsync(DistributedEventSent distributedEvent)