Browse Source

Use `AbpDaprEventData ` to pass `CorrelationId`.

pull/16795/head
maliming 3 years ago
parent
commit
26f20deb0c
No known key found for this signature in database GPG Key ID: A646B9CB645ECEA4
  1. 17
      framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs
  2. 27
      framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/AbpDaprEventData.cs
  3. 3
      framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/DaprDistributedEventBus.cs

17
framework/src/Volo.Abp.AspNetCore.Mvc.Dapr.EventBus/Volo/Abp/AspNetCore/Mvc/Dapr/EventBus/Controllers/AbpAspNetCoreMvcDaprEventsController.cs

@ -1,4 +1,5 @@
using System;
using System.Linq;
using System.Text.Json;
using System.Threading.Tasks;
using Microsoft.AspNetCore.Mvc;
@ -32,9 +33,19 @@ public class AbpAspNetCoreMvcDaprEventsController : AbpController
}
var distributedEventBus = HttpContext.RequestServices.GetRequiredService<DaprDistributedEventBus>();
var eventData = daprSerializer.Deserialize(data, distributedEventBus.GetEventType(topic));
string correlationId = null;
await distributedEventBus.TriggerHandlersAsync(id, distributedEventBus.GetEventType(topic), eventData, correlationId);
if (data.Contains("Data") && data.Contains("CorrelationId")) //TODO: Check the json with JSON Schema.
{
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);
}
else
{
var eventData = daprSerializer.Deserialize(data, distributedEventBus.GetEventType(topic));
await distributedEventBus.TriggerHandlersAsync(id, distributedEventBus.GetEventType(topic), eventData, null);
}
return Ok();
}
}

27
framework/src/Volo.Abp.EventBus.Dapr/Volo/Abp/EventBus/Dapr/AbpDaprEventData.cs

@ -0,0 +1,27 @@
using System;
using System.Reflection;
namespace Volo.Abp.EventBus.Dapr;
public class AbpDaprEventData<TData>
{
public TData Data { get; set; }
public string CorrelationId { get; set; }
public AbpDaprEventData(TData data, string correlationId)
{
Data = data;
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)!;
}
}

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

@ -256,7 +256,8 @@ public class DaprDistributedEventBus : DistributedEventBusBase, ISingletonDepend
protected virtual async Task PublishToDaprAsync(string eventName, object eventData, string correlationId)
{
var client = DaprClientFactory.Create();
await client.PublishEventAsync(pubsubName: DaprEventBusOptions.PubSubName, topicName: eventName, data: eventData);
var data = AbpDaprEventData<object>.Create(eventData, correlationId);
await client.PublishEventAsync(pubsubName: DaprEventBusOptions.PubSubName, topicName: eventName, data: data);
}
private static bool ShouldTriggerEventForHandler(Type targetEventType, Type handlerEventType)

Loading…
Cancel
Save