diff --git a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/AbpDistributedEventBusExtensions.cs b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/AbpDistributedEventBusExtensions.cs index dda6f71245..ed652f677f 100644 --- a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/AbpDistributedEventBusExtensions.cs +++ b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/AbpDistributedEventBusExtensions.cs @@ -4,12 +4,12 @@ namespace Volo.Abp.EventBus.Boxes { public static class AbpDistributedEventBusExtensions { - public static IRawEventPublisher AsRawEventPublisher(this IDistributedEventBus eventBus) + public static ISupportsEventBoxes AsSupportsEventBoxes(this IDistributedEventBus eventBus) { - var rawPublisher = eventBus as IRawEventPublisher; + var rawPublisher = eventBus as ISupportsEventBoxes; if (rawPublisher == null) { - throw new AbpException($"Given type ({eventBus.GetType().AssemblyQualifiedName}) should implement {nameof(IRawEventPublisher)}!"); + throw new AbpException($"Given type ({eventBus.GetType().AssemblyQualifiedName}) should implement {nameof(ISupportsEventBoxes)}!"); } return rawPublisher; diff --git a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs index c9b45a3ea8..7f238728f7 100644 --- a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs +++ b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/InboxProcessor.cs @@ -103,8 +103,8 @@ namespace Volo.Abp.EventBus.Boxes using (var uow = UnitOfWorkManager.Begin(isTransactional: true, requiresNew: true)) { await DistributedEventBus - .AsRawEventPublisher() - .ProcessRawAsync(InboxConfig, waitingEvent.EventName, waitingEvent.EventData); + .AsSupportsEventBoxes() + .ProcessFromInboxAsync(waitingEvent, InboxConfig); await Inbox.MarkAsProcessedAsync(waitingEvent.Id); diff --git a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSender.cs b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSender.cs index 1a2e0b9d79..1be2374268 100644 --- a/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSender.cs +++ b/framework/src/Volo.Abp.EventBus.Boxes/Volo/Abp/EventBus/Boxes/OutboxSender.cs @@ -82,11 +82,13 @@ namespace Volo.Abp.EventBus.Boxes foreach (var waitingEvent in waitingEvents) { await DistributedEventBus - .AsRawEventPublisher() - .PublishRawAsync(waitingEvent.Id, waitingEvent.EventName, waitingEvent.EventData); + .AsSupportsEventBoxes() + .PublishFromOutboxAsync( + waitingEvent, + OutboxConfig + ); await Outbox.DeleteAsync(waitingEvent.Id); - Logger.LogInformation($"Sent the event to the message broker with id = {waitingEvent.Id:N}"); } } 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 3ffe3bb880..b747a858d2 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 @@ -201,32 +201,33 @@ namespace Volo.Abp.EventBus.Kafka unitOfWork.AddOrReplaceDistributedEvent(eventRecord); } - public override Task PublishRawAsync( - Guid eventId, - string eventName, - byte[] eventData) + public override Task PublishFromOutboxAsync( + OutgoingEventInfo outgoingEvent, + OutboxConfig outboxConfig) { return PublishAsync( AbpKafkaEventBusOptions.TopicName, - eventName, - eventData, + outgoingEvent.EventName, + outgoingEvent.EventData, new Headers { - { "messageId", System.Text.Encoding.UTF8.GetBytes(eventId.ToString("N")) } + { "messageId", System.Text.Encoding.UTF8.GetBytes(outgoingEvent.Id.ToString("N")) } }, null ); } - public override async Task ProcessRawAsync(InboxConfig inboxConfig, string eventName, byte[] eventDataBytes) + public override async Task ProcessFromInboxAsync( + IncomingEventInfo incomingEvent, + InboxConfig inboxConfig) { - var eventType = EventTypes.GetOrDefault(eventName); + var eventType = EventTypes.GetOrDefault(incomingEvent.EventName); if (eventType == null) { return; } - var eventData = Serializer.Deserialize(eventDataBytes, eventType); + var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType); var exceptions = new List(); await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig); if (exceptions.Any()) 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 deb20b5ff3..12eec80ce7 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 @@ -212,22 +212,26 @@ namespace Volo.Abp.EventBus.RabbitMq unitOfWork.AddOrReplaceDistributedEvent(eventRecord); } - public override Task PublishRawAsync(Guid eventId, string eventName, byte[] eventData) + public override Task PublishFromOutboxAsync( + OutgoingEventInfo outgoingEvent, + OutboxConfig outboxConfig) { - return PublishAsync(eventName, eventData, null, eventId: eventId); + return PublishAsync(outgoingEvent.EventName, outgoingEvent.EventData, null, eventId: outgoingEvent.Id); } - public override async Task ProcessRawAsync(InboxConfig inboxConfig, string eventName, byte[] eventDataBytes) + public override async Task ProcessFromInboxAsync( + IncomingEventInfo incomingEvent, + InboxConfig inboxConfig) { //TODO: We have a duplication in logic and also with the kafka side! - var eventType = EventTypes.GetOrDefault(eventName); + var eventType = EventTypes.GetOrDefault(incomingEvent.EventName); if (eventType == null) { return; } - var eventData = Serializer.Deserialize(eventDataBytes, eventType); + var eventData = Serializer.Deserialize(incomingEvent.EventData, eventType); var exceptions = new List(); await TriggerHandlersAsync(eventType, eventData, exceptions, inboxConfig); if (exceptions.Any()) 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 af3d148ecf..8dac885036 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 @@ -188,13 +188,17 @@ namespace Volo.Abp.EventBus.Rebus return false; } - public override Task PublishRawAsync(Guid eventId, string eventName, byte[] eventData) + public override Task PublishFromOutboxAsync( + OutgoingEventInfo outgoingEvent, + OutboxConfig outboxConfig) { /* TODO: IMPLEMENT! */ throw new NotImplementedException(); } - public override Task ProcessRawAsync(InboxConfig inboxConfig, string eventName, byte[] eventDataBytes) + public override Task ProcessFromInboxAsync( + IncomingEventInfo incomingEvent, + InboxConfig inboxConfig) { /* TODO: IMPLEMENT! */ throw new NotImplementedException(); 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 1fdaf0d26f..70c5bd5533 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 @@ -10,7 +10,7 @@ using Volo.Abp.Uow; namespace Volo.Abp.EventBus.Distributed { - public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventBus, IRawEventPublisher + public abstract class DistributedEventBusBase : EventBusBase, IDistributedEventBus, ISupportsEventBoxes { protected IGuidGenerator GuidGenerator { get; } protected IClock Clock { get; } @@ -80,8 +80,14 @@ namespace Volo.Abp.EventBus.Distributed await PublishToEventBusAsync(eventType, eventData); } - public abstract Task PublishRawAsync(Guid eventId, string eventName, byte[] eventData); - public abstract Task ProcessRawAsync(InboxConfig inboxConfig, string eventName, byte[] eventDataBytes); + public abstract Task PublishFromOutboxAsync( + OutgoingEventInfo outgoingEvent, + OutboxConfig outboxConfig + ); + + public abstract Task ProcessFromInboxAsync( + IncomingEventInfo incomingEvent, + InboxConfig inboxConfig); private async Task AddToOutboxAsync(Type eventType, object eventData) { diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IRawEventPublisher.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IRawEventPublisher.cs deleted file mode 100644 index 4505239ade..0000000000 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/IRawEventPublisher.cs +++ /dev/null @@ -1,18 +0,0 @@ -using System; -using System.Threading.Tasks; - -namespace Volo.Abp.EventBus.Distributed -{ - public interface IRawEventPublisher //TODO: Rename: ISupportsEventBoxes - { - Task PublishRawAsync( - Guid eventId, - string eventName, - byte[] eventData); - - Task ProcessRawAsync( - InboxConfig inboxConfig, - string eventName, - byte[] eventDataBytes); - } -} \ No newline at end of file diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs new file mode 100644 index 0000000000..7a13696bc0 --- /dev/null +++ b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/ISupportsEventBoxes.cs @@ -0,0 +1,17 @@ +using System.Threading.Tasks; + +namespace Volo.Abp.EventBus.Distributed +{ + public interface ISupportsEventBoxes + { + Task PublishFromOutboxAsync( + OutgoingEventInfo outgoingEvent, + OutboxConfig outboxConfig + ); + + Task ProcessFromInboxAsync( + IncomingEventInfo incomingEvent, + InboxConfig inboxConfig + ); + } +} \ No newline at end of file