Browse Source

Change the order of events.

pull/16333/head
maliming 4 years ago
parent
commit
f3589878fa
No known key found for this signature in database GPG Key ID: A646B9CB645ECEA4
  1. 18
      framework/src/Volo.Abp.EventBus.Azure/Volo/Abp/EventBus/Azure/AzureDistributedEventBus.cs
  2. 11
      framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs
  3. 34
      framework/src/Volo.Abp.EventBus.Kafka/Volo/Abp/EventBus/Kafka/KafkaDistributedEventBus.cs
  4. 22
      framework/src/Volo.Abp.EventBus.RabbitMQ/Volo/Abp/EventBus/RabbitMq/RabbitMqDistributedEventBus.cs
  5. 10
      framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs
  6. 30
      framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/DistributedEventBusBase.cs

18
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<OutgoingEventInfo> 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)

11
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<OutgoingEventInfo> 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);
}

34
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<OutgoingEventInfo> 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<string, byte[]>
@ -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
});
}
}

22
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();

10
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<OutgoingEventInfo> 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();

30
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<Exception> 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)

Loading…
Cancel
Save