Browse Source

Add more properties to `AbpDaprEventData`.

pull/16795/head
maliming 3 years ago
parent
commit
75e3383008
No known key found for this signature in database GPG Key ID: A646B9CB645ECEA4
  1. 20
      framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs
  2. 2
      framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/IDaprSerializer.cs
  3. 5
      framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/Utf8JsonDaprSerializer.cs
  4. 30
      framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/AbpDaprEventData.cs
  5. 16
      framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs

20
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<IDaprSerializer>(); var daprSerializer = HttpContext.RequestServices.GetRequiredService<IDaprSerializer>();
var body = (await JsonDocument.ParseAsync(HttpContext.Request.Body)); var body = (await JsonDocument.ParseAsync(HttpContext.Request.Body));
var id = body.RootElement.GetProperty("id").GetString();
var pubSubName = body.RootElement.GetProperty("pubsubname").GetString(); var pubSubName = body.RootElement.GetProperty("pubsubname").GetString();
var topic = body.RootElement.GetProperty("topic").GetString(); var topic = body.RootElement.GetProperty("topic").GetString();
var data = body.RootElement.GetProperty("data").GetRawText(); var data = body.RootElement.GetProperty("data").GetRawText();
@ -36,16 +35,16 @@ public class AbpAspNetCoreMvcDaprEventsController : AbpController
if (IsAbpDaprEventData(data)) if (IsAbpDaprEventData(data))
{ {
var abpDaprEventData = daprSerializer.Deserialize(data, typeof(AbpDaprEventData<>).MakeGenericType(distributedEventBus.GetEventType(topic))); var daprEventData = daprSerializer.Deserialize(data, typeof(AbpDaprEventData)).As<AbpDaprEventData>();
var eventData = abpDaprEventData.GetType().GetProperties().First(x => x.Name == "Data").GetValue(abpDaprEventData); var eventData = daprSerializer.Deserialize(daprEventData.JsonData, distributedEventBus.GetEventType(daprEventData.Topic));
var correlationId = abpDaprEventData.GetType().GetProperties().First(x => x.Name == "CorrelationId").GetValue(abpDaprEventData) as string; await distributedEventBus.TriggerHandlersAsync(distributedEventBus.GetEventType(daprEventData.Topic), eventData, daprEventData.MessageId, daprEventData.CorrelationId);
await distributedEventBus.TriggerHandlersAsync(id, distributedEventBus.GetEventType(topic), eventData, correlationId);
} }
else else
{ {
var eventData = daprSerializer.Deserialize(data, distributedEventBus.GetEventType(topic)); 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(); return Ok();
} }
@ -53,8 +52,11 @@ public class AbpAspNetCoreMvcDaprEventsController : AbpController
{ {
var document = JsonDocument.Parse(data); var document = JsonDocument.Parse(data);
var objects = document.RootElement.EnumerateObject().ToList(); var objects = document.RootElement.EnumerateObject().ToList();
return objects.Count == 2 && return objects.Count == 5 &&
objects.Any(x => x.Name.Equals("data", StringComparison.CurrentCultureIgnoreCase)) && objects.Any(x => x.Name.Equals("PubSubName", StringComparison.CurrentCultureIgnoreCase)) &&
objects.Any(x => x.Name.Equals("correlationId", 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));
} }
} }

2
framework/src/Volo.Abp.Dapr/Volo/Abp/Dapr/IDaprSerializer.cs

@ -6,6 +6,8 @@ public interface IDaprSerializer
{ {
byte[] Serialize(object obj); byte[] Serialize(object obj);
string SerializeToString(object obj);
object Deserialize(byte[] value, Type type); object Deserialize(byte[] value, Type type);
object Deserialize(string value, Type type); object Deserialize(string value, Type type);

5
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)); return Encoding.UTF8.GetBytes(_jsonSerializer.Serialize(obj));
} }
public string SerializeToString(object obj)
{
return _jsonSerializer.Serialize(obj);
}
public object Deserialize(byte[] value, Type type) public object Deserialize(byte[] value, Type type)
{ {
return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value)); return _jsonSerializer.Deserialize(type, Encoding.UTF8.GetString(value));

30
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; namespace Volo.Abp.EventBus.Dapr;
public class AbpDaprEventData<TData> 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 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; 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)!;
}
} }

16
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) 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) 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<OutgoingEventInfo> outgoingEvents, OutboxConfig outboxConfig) public async override Task PublishManyFromOutboxAsync(IEnumerable<OutgoingEventInfo> 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)) if (await AddToInboxAsync(messageId, EventNameAttribute.GetNameOrDefault(eventType), eventType, eventData, correlationId))
{ {
@ -248,15 +248,15 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
return EventTypes.GetOrDefault(eventName); 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 client = DaprClientFactory.Create();
var data = AbpDaprEventData<object>.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); await client.PublishEventAsync(pubsubName: DaprEventBusOptions.PubSubName, topicName: eventName, data: data);
} }

Loading…
Cancel
Save