From 75e3383008a6337a4b2ca0a6179b8548fd7df71f Mon Sep 17 00:00:00 2001 From: maliming Date: Tue, 13 Jun 2023 13:48:15 +0800 Subject: [PATCH] Add more properties to `AbpDaprEventData`. --- .../AbpAspNetCoreMvcDaprEventsController.cs | 20 +++++++------ .../Volo/Abp/Dapr/IDaprSerializer.cs | 2 ++ .../Volo/Abp/Dapr/Utf8JsonDaprSerializer.cs | 5 ++++ .../Abp/EventBus/Dapr/AbpDaprEventData.cs | 30 ++++++++----------- .../EventBus/Dapr/DaprDistributedEventBus.cs | 16 +++++----- 5 files changed, 39 insertions(+), 34 deletions(-) diff --git a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs b/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs index 706bb8a56c..456a118d4e 100644 --- a/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs +++ b/framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs @@ -22,7 +22,6 @@ public class AbpAspNetCoreMvcDaprEventsController : AbpController var daprSerializer = HttpContext.RequestServices.GetRequiredService(); var body = (await JsonDocument.ParseAsync(HttpContext.Request.Body)); - var id = body.RootElement.GetProperty("id").GetString(); var pubSubName = body.RootElement.GetProperty("pubsubname").GetString(); var topic = body.RootElement.GetProperty("topic").GetString(); var data = body.RootElement.GetProperty("data").GetRawText(); @@ -36,16 +35,16 @@ public class AbpAspNetCoreMvcDaprEventsController : AbpController if (IsAbpDaprEventData(data)) { - var abpDaprEventData = daprSerializer.Deserialize(data, typeof(AbpDaprEventData<>).MakeGenericType(distributedEventBus.GetEventType(topic))); - var eventData = abpDaprEventData.GetType().GetProperties().First(x => x.Name == "Data").GetValue(abpDaprEventData); - var correlationId = abpDaprEventData.GetType().GetProperties().First(x => x.Name == "CorrelationId").GetValue(abpDaprEventData) as string; - await distributedEventBus.TriggerHandlersAsync(id, distributedEventBus.GetEventType(topic), eventData, correlationId); + var daprEventData = daprSerializer.Deserialize(data, typeof(AbpDaprEventData)).As(); + var eventData = daprSerializer.Deserialize(daprEventData.JsonData, distributedEventBus.GetEventType(daprEventData.Topic)); + await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(daprEventData.Topic), eventData, daprEventData.MessageId, daprEventData.CorrelationId); } else { var eventData = daprSerializer.Deserialize(data, distributedEventBus.GetEventType(topic)); - await distributedEventBus.TriggerHandlersAsync(id, distributedEventBus.GetEventType(topic), eventData, null); + await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(topic), eventData); } + return Ok(); } @@ -53,8 +52,11 @@ public class AbpAspNetCoreMvcDaprEventsController : AbpController { var document = JsonDocument.Parse(data); var objects = document.RootElement.EnumerateObject().ToList(); - return objects.Count == 2 && - objects.Any(x => x.Name.Equals("data", StringComparison.CurrentCultureIgnoreCase)) && - objects.Any(x => x.Name.Equals("correlationId", StringComparison.CurrentCultureIgnoreCase)); + return objects.Count == 5 && + objects.Any(x => x.Name.Equals("PubSubName", StringComparison.CurrentCultureIgnoreCase)) && + objects.Any(x => x.Name.Equals("Topic", StringComparison.CurrentCultureIgnoreCase)) && + objects.Any(x => x.Name.Equals("MessageId", StringComparison.CurrentCultureIgnoreCase)) && + objects.Any(x => x.Name.Equals("jsonData", StringComparison.CurrentCultureIgnoreCase)) && + objects.Any(x => x.Name.Equals("CorrelationId", StringComparison.CurrentCultureIgnoreCase)); } } diff --git a/framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/IDaprSerializer.cs b/framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/IDaprSerializer.cs index 9a8b4c9520..1d1861cd97 100644 --- a/framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/IDaprSerializer.cs +++ b/framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/IDaprSerializer.cs @@ -6,6 +6,8 @@ public interface IDaprSerializer { byte[] Serialize(object obj); + string SerializeToString(object obj); + object Deserialize(byte[] value, Type type); object Deserialize(string value, Type type); diff --git a/framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/Utf8JsonDaprSerializer.cs b/framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/Utf8JsonDaprSerializer.cs index c0924f775b..a1a8324598 100644 --- a/framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/Utf8JsonDaprSerializer.cs +++ b/framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/Utf8JsonDaprSerializer.cs @@ -19,6 +19,11 @@ public class Utf8JsonDaprSerializer : IDaprSerializer, ITransientDependency return Encoding.UTF8.GetBytes(_jsonSerializer.Serialize(obj)); } + public string SerializeToString(object obj) + { + return _jsonSerializer.Serialize(obj); + } + public object Deserialize(byte[] value, Type type) { return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value)); diff --git a/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/AbpDaprEventData.cs b/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/AbpDaprEventData.cs index 25e6bc1b0b..ee08586b8d 100644 --- a/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/AbpDaprEventData.cs +++ b/framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/AbpDaprEventData.cs @@ -1,27 +1,23 @@ -using System; -using System.Reflection; - namespace Volo.Abp.EventBus.Dapr; -public class AbpDaprEventData +public class AbpDaprEventData { - public TData Data { get; set; } + public string PubSubName { get; set; } + + public string Topic { get; set; } + + public string MessageId { get; set; } + + public string JsonData { get; set; } public string CorrelationId { get; set; } - public AbpDaprEventData(TData data, string correlationId) + public AbpDaprEventData(string pubSubName, string topic, string messageId, string jsonData, string correlationId) { - Data = data; + PubSubName = pubSubName; + Topic = topic; + MessageId = messageId; + JsonData = jsonData; CorrelationId = correlationId; } - - public static object Create(object data, string correlationId) - { - return Activator.CreateInstance( - typeof(AbpDaprEventData<>).MakeGenericType(data.GetType()), - BindingFlags.Instance | BindingFlags.Public, - binder: null, - new object[] { data, correlationId }, - culture: null)!; - } } 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 b7b6efcb92..4912a58388 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 @@ -131,7 +131,7 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend protected async override Task PublishToEventBusAsync(Type eventType, object eventData) { - await PublishToDaprAsync(eventType, eventData, CorrelationIdProvider.Get()); + await PublishToDaprAsync(eventType, eventData, null, CorrelationIdProvider.Get()); } protected override void AddToUnitOfWork(IUnitOfWork unitOfWork, UnitOfWorkEventRecord eventRecord) @@ -163,7 +163,7 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend }); } - await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName)), outgoingEvent.GetCorrelationId()); + await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName)), outgoingEvent.Id, outgoingEvent.GetCorrelationId()); } public async override Task PublishManyFromOutboxAsync(IEnumerable outgoingEvents, OutboxConfig outboxConfig) @@ -182,11 +182,11 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend }); } - await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName)), outgoingEvent.GetCorrelationId()); + await PublishToDaprAsync(outgoingEvent.EventName, Serializer.Deserialize(outgoingEvent.EventData, GetEventType(outgoingEvent.EventName)), outgoingEvent.Id, outgoingEvent.GetCorrelationId()); } } - public virtual async Task TriggerHandlersAsync(string messageId, Type eventType, object eventData, string correlationId) + public virtual async Task TriggerHandlersAsync(Type eventType, object eventData, string messageId = null, string correlationId = null) { if (await AddToInboxAsync(messageId, EventNameAttribute.GetNameOrDefault(eventType), eventType, eventData, correlationId)) { @@ -248,15 +248,15 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend return EventTypes.GetOrDefault(eventName); } - protected virtual async Task PublishToDaprAsync(Type eventType, object eventData, string correlationId) + protected virtual async Task PublishToDaprAsync(Type eventType, object eventData, Guid? messageId = null, string correlationId = null) { - await PublishToDaprAsync(EventNameAttribute.GetNameOrDefault(eventType), eventData, correlationId); + await PublishToDaprAsync(EventNameAttribute.GetNameOrDefault(eventType), eventData, messageId, correlationId); } - protected virtual async Task PublishToDaprAsync(string eventName, object eventData, string correlationId) + protected virtual async Task PublishToDaprAsync(string eventName, object eventData, Guid? messageId = null, string correlationId = null) { var client = DaprClientFactory.Create(); - var data = AbpDaprEventData.Create(eventData, correlationId); + var data = new AbpDaprEventData(DaprEventBusOptions.PubSubName, eventName, (messageId ?? GuidGenerator.Create()).ToString("N"), Serializer.SerializeToString(eventData), correlationId); await client.PublishEventAsync(pubsubName: DaprEventBusOptions.PubSubName, topicName: eventName, data: data); }