Browse Source

Make sure `eventType` is not `null`.

pull/21965/head
maliming 2 years ago
parent
commit
69849ead7e
No known key found for this signature in database GPG Key ID: A646B9CB645ECEA4
  1. 16
      framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprEventBusModule.cs
  2. 30
      framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs
  3. 7
      framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs
  4. 9
      framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs

16
framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/AbpAspNetCoreMvcDaprEventBusModule.cs

@ -97,13 +97,21 @@ public class AbpAspNetCoreMvcDaprEventBusModule : AbpModule
if (IsAbpDaprEventData(data)) if (IsAbpDaprEventData(data))
{ {
var daprEventData = daprSerializer.Deserialize(data, typeof(AbpDaprEventData)).As<AbpDaprEventData>(); var daprEventData = daprSerializer.Deserialize(data, typeof(AbpDaprEventData)).As<AbpDaprEventData>();
var eventData = daprSerializer.Deserialize(daprEventData.JsonData, distributedEventBus.GetEventType(daprEventData.Topic)); var eventType = distributedEventBus.GetEventType(daprEventData.Topic);
await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(daprEventData.Topic), eventData, daprEventData.MessageId, daprEventData.CorrelationId); if (eventType != null)
{
var eventData = daprSerializer.Deserialize(daprEventData.JsonData, eventType);
await distributedEventBus.TriggerHandlersAsync(eventType, eventData, daprEventData.MessageId, daprEventData.CorrelationId);
}
} }
else else
{ {
var eventData = daprSerializer.Deserialize(data, distributedEventBus.GetEventType(topic!)); var eventType = distributedEventBus.GetEventType(topic);
await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(topic!), eventData); if (eventType != null)
{
var eventData = daprSerializer.Deserialize(data, eventType);
await distributedEventBus.TriggerHandlersAsync(eventType, eventData);
}
} }
httpContext.Response.StatusCode = 200; httpContext.Response.StatusCode = 200;

30
framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs

@ -153,7 +153,13 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
public async override Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig) public async override Task PublishFromOutboxAsync(OutgoingEventInfo outgoingEvent, OutboxConfig outboxConfig)
{ {
await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName)), outgoingEvent.Id, outgoingEvent.GetCorrelationId()); var eventType = GetEventType(outgoingEvent.EventName);
if (eventType == null)
{
return;
}
await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, eventType), outgoingEvent.Id, outgoingEvent.GetCorrelationId());
using (CorrelationIdProvider.Change(outgoingEvent.GetCorrelationId())) using (CorrelationIdProvider.Change(outgoingEvent.GetCorrelationId()))
{ {
@ -168,21 +174,9 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig) public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig)
{ {
var outgoingEventArray = outgoingEvents.ToArray(); foreach (var outgoingEvent in outgoingEvents)
foreach (var outgoingEvent in outgoingEventArray)
{ {
await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName)), outgoingEvent.Id, outgoingEvent.GetCorrelationId()); await PublishFromOutboxAsync(outgoingEvent, outboxConfig);
using (CorrelationIdProvider.Change(outgoingEvent.GetCorrelationId()))
{
await TriggerDistributedEventSentAsync(new DistributedEventSent()
{
Source = DistributedEventSource.Outbox,
EventName = outgoingEvent.EventName,
EventData = outgoingEvent.EventData
});
}
} }
} }
@ -201,7 +195,7 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig) public async override Task ProcessFromInboxAsync(IncomingEventInfo incomingEvent, InboxConfig inboxConfig)
{ {
var eventType = EventTypes.GetOrDefault(incomingEvent.EventName); var eventType = GetEventType(incomingEvent.EventName);
if (eventType == null) if (eventType == null)
{ {
return; return;
@ -243,9 +237,9 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
); );
} }
public Type GetEventType(string eventName) public Type? GetEventType(string eventName)
{ {
return EventTypes.GetOrDefault(eventName)!; return EventTypes.GetOrDefault(eventName);
} }
protected virtual async Task PublishToDaprAsync(Type eventType, object eventData, Guid? messageId = null, string? correlationId = null) protected virtual async Task PublishToDaprAsync(Type eventType, object eventData, Guid? messageId = null, string? correlationId = null)

7
framework/src/Volo.Abp.EventBus.Rebus/Volo/Abp/EventBus/Rebus/RebusDistributedEventBus.cs

@ -250,7 +250,12 @@ public class RebusDistributedEventBus : DistributedEventBusBase, ISingletonDepen
OutgoingEventInfo outgoingEvent, OutgoingEventInfo outgoingEvent,
OutboxConfig outboxConfig) OutboxConfig outboxConfig)
{ {
var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName)!; var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName);
if (eventType == null)
{
return;
}
var eventData = Serializer.Deserialize(outgoingEvent.EventData, eventType); var eventData = Serializer.Deserialize(outgoingEvent.EventData, eventType);
var headers = new Dictionary<string, string>(); var headers = new Dictionary<string, string>();

9
framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/LocalDistributedEventBus.cs

@ -72,6 +72,8 @@ public class LocalDistributedEventBus : DistributedEventBusBase, ISingletonDepen
public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory) public override IDisposable Subscribe(Type eventType, IEventHandlerFactory factory)
{ {
var eventName = EventNameAttribute.GetNameOrDefault(eventType);
EventTypes.GetOrAdd(eventName, eventType);
return LocalEventBus.Subscribe(eventType, factory); return LocalEventBus.Subscribe(eventType, factory);
} }
@ -157,7 +159,12 @@ public class LocalDistributedEventBus : DistributedEventBusBase, ISingletonDepen
EventData = outgoingEvent.EventData EventData = outgoingEvent.EventData
}); });
var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName)!; var eventType = EventTypes.GetOrDefault(outgoingEvent.EventName);
if (eventType == null)
{
return;
}
var eventData = JsonSerializer.Deserialize(Encoding.UTF8.GetString(outgoingEvent.EventData), eventType)!; var eventData = JsonSerializer.Deserialize(Encoding.UTF8.GetString(outgoingEvent.EventData), eventType)!;
await LocalEventBus.PublishAsync(eventType, eventData, false); await LocalEventBus.PublishAsync(eventType, eventData, false);
} }

Loading…
Cancel
Save