mirror of https://github.com/abpframework/abp.git
Browse Source
Introduce AnonymousEventData and add support for anonymous (name-based) events across the event bus implementations. Adds string-based Subscribe/Unsubscribe APIs, anonymous handler factories, and handling in distributed providers (Azure, Dapr, Kafka, RabbitMQ, Rebus) and local buses. Update EventBusBase and DistributedEventBusBase to resolve event names/data (GetEventName/GetEventData/ResolveEventForPublishing) and route/serialize/deserialize anonymous payloads. Also add AnonymousEventHandlerFactoryUnregistrar and minimal NullDistributedEventBus implementations, plus tests for anonymous local events.pull/25023/head
17 changed files with 1140 additions and 84 deletions
@ -0,0 +1,103 @@ |
|||||
|
using System; |
||||
|
using System.Collections.Generic; |
||||
|
using System.Linq; |
||||
|
using System.Text.Json; |
||||
|
|
||||
|
namespace Volo.Abp.EventBus; |
||||
|
|
||||
|
[Serializable] |
||||
|
public class AnonymousEventData |
||||
|
{ |
||||
|
public string EventName { get; } |
||||
|
public object Data { get; } |
||||
|
|
||||
|
private JsonElement? _cachedJsonElement; |
||||
|
|
||||
|
public AnonymousEventData(string eventName, object data) |
||||
|
{ |
||||
|
EventName = eventName; |
||||
|
Data = data; |
||||
|
} |
||||
|
|
||||
|
public object ConvertToTypedObject() |
||||
|
{ |
||||
|
return ConvertElement(GetJsonElement()); |
||||
|
} |
||||
|
|
||||
|
public T ConvertToTypedObject<T>() |
||||
|
{ |
||||
|
if (Data is T typedData) |
||||
|
{ |
||||
|
return typedData; |
||||
|
} |
||||
|
|
||||
|
return GetJsonElement().Deserialize<T>() |
||||
|
?? throw new InvalidOperationException($"Failed to deserialize AnonymousEventData to {typeof(T).FullName}."); |
||||
|
} |
||||
|
|
||||
|
public object ConvertToTypedObject(Type type) |
||||
|
{ |
||||
|
if (type.IsInstanceOfType(Data)) |
||||
|
{ |
||||
|
return Data; |
||||
|
} |
||||
|
|
||||
|
return GetJsonElement().Deserialize(type) |
||||
|
?? throw new InvalidOperationException($"Failed to deserialize AnonymousEventData to {type.FullName}."); |
||||
|
} |
||||
|
|
||||
|
private JsonElement GetJsonElement() |
||||
|
{ |
||||
|
if (_cachedJsonElement.HasValue) |
||||
|
{ |
||||
|
return _cachedJsonElement.Value; |
||||
|
} |
||||
|
|
||||
|
if (Data is JsonElement existingElement) |
||||
|
{ |
||||
|
_cachedJsonElement = existingElement; |
||||
|
return existingElement; |
||||
|
} |
||||
|
|
||||
|
_cachedJsonElement = JsonSerializer.SerializeToElement(Data); |
||||
|
return _cachedJsonElement.Value; |
||||
|
} |
||||
|
|
||||
|
private static object ConvertElement(JsonElement element) |
||||
|
{ |
||||
|
switch (element.ValueKind) |
||||
|
{ |
||||
|
case JsonValueKind.Object: |
||||
|
{ |
||||
|
var obj = new Dictionary<string, object?>(); |
||||
|
foreach (var property in element.EnumerateObject()) |
||||
|
{ |
||||
|
obj[property.Name] = property.Value.ValueKind == JsonValueKind.Null |
||||
|
? null |
||||
|
: ConvertElement(property.Value); |
||||
|
} |
||||
|
return obj; |
||||
|
} |
||||
|
case JsonValueKind.Array: |
||||
|
return element.EnumerateArray() |
||||
|
.Select(item => item.ValueKind == JsonValueKind.Null ? null : (object?)ConvertElement(item)) |
||||
|
.ToList(); |
||||
|
case JsonValueKind.String: |
||||
|
return element.GetString()!; |
||||
|
case JsonValueKind.Number when element.TryGetInt64(out var longValue): |
||||
|
return longValue; |
||||
|
case JsonValueKind.Number when element.TryGetDecimal(out var decimalValue): |
||||
|
return decimalValue; |
||||
|
case JsonValueKind.Number when element.TryGetDouble(out var doubleValue): |
||||
|
return doubleValue; |
||||
|
case JsonValueKind.True: |
||||
|
return true; |
||||
|
case JsonValueKind.False: |
||||
|
return false; |
||||
|
case JsonValueKind.Null: |
||||
|
case JsonValueKind.Undefined: |
||||
|
default: |
||||
|
return null!; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,162 @@ |
|||||
|
using System.Collections.Generic; |
||||
|
using System.Threading.Tasks; |
||||
|
using Shouldly; |
||||
|
using Xunit; |
||||
|
|
||||
|
namespace Volo.Abp.EventBus.Local; |
||||
|
|
||||
|
public class LocalEventBus_Anonymous_Test : EventBusTestBase |
||||
|
{ |
||||
|
[Fact] |
||||
|
public async Task Should_Handle_Anonymous_Handler_With_EventName() |
||||
|
{ |
||||
|
var handleCount = 0; |
||||
|
|
||||
|
LocalEventBus.Subscribe("TestEvent", |
||||
|
new SingleInstanceHandlerFactory(new ActionEventHandler<AnonymousEventData>(async (d) => |
||||
|
{ |
||||
|
handleCount++; |
||||
|
d.EventName.ShouldBe("TestEvent"); |
||||
|
await Task.CompletedTask; |
||||
|
}))); |
||||
|
|
||||
|
await LocalEventBus.PublishAsync("TestEvent", new { Value = 1 }); |
||||
|
await LocalEventBus.PublishAsync("TestEvent", new { Value = 2 }); |
||||
|
|
||||
|
handleCount.ShouldBe(2); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Handle_Typed_Handler_When_Published_With_EventName() |
||||
|
{ |
||||
|
var handleCount = 0; |
||||
|
|
||||
|
LocalEventBus.Subscribe<MySimpleEventData>(async (data) => |
||||
|
{ |
||||
|
handleCount++; |
||||
|
await Task.CompletedTask; |
||||
|
}); |
||||
|
|
||||
|
var eventName = EventNameAttribute.GetNameOrDefault<MySimpleEventData>(); |
||||
|
await LocalEventBus.PublishAsync(eventName, new MySimpleEventData(42)); |
||||
|
|
||||
|
handleCount.ShouldBe(1); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Convert_Dictionary_To_Typed_Handler() |
||||
|
{ |
||||
|
MySimpleEventData? receivedData = null; |
||||
|
|
||||
|
LocalEventBus.Subscribe<MySimpleEventData>(async (data) => |
||||
|
{ |
||||
|
receivedData = data; |
||||
|
await Task.CompletedTask; |
||||
|
}); |
||||
|
|
||||
|
var eventName = EventNameAttribute.GetNameOrDefault<MySimpleEventData>(); |
||||
|
await LocalEventBus.PublishAsync(eventName, new Dictionary<string, object> |
||||
|
{ |
||||
|
{ "Value", 42 } |
||||
|
}); |
||||
|
|
||||
|
receivedData.ShouldNotBeNull(); |
||||
|
receivedData.Value.ShouldBe(42); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Trigger_Both_Typed_And_Anonymous_Handlers() |
||||
|
{ |
||||
|
var typedHandleCount = 0; |
||||
|
var anonymousHandleCount = 0; |
||||
|
|
||||
|
LocalEventBus.Subscribe<MySimpleEventData>(async (data) => |
||||
|
{ |
||||
|
typedHandleCount++; |
||||
|
await Task.CompletedTask; |
||||
|
}); |
||||
|
|
||||
|
var eventName = EventNameAttribute.GetNameOrDefault<MySimpleEventData>(); |
||||
|
|
||||
|
LocalEventBus.Subscribe(eventName, |
||||
|
new SingleInstanceHandlerFactory(new ActionEventHandler<AnonymousEventData>(async (d) => |
||||
|
{ |
||||
|
anonymousHandleCount++; |
||||
|
await Task.CompletedTask; |
||||
|
}))); |
||||
|
|
||||
|
await LocalEventBus.PublishAsync(new MySimpleEventData(1)); |
||||
|
|
||||
|
typedHandleCount.ShouldBe(1); |
||||
|
anonymousHandleCount.ShouldBe(1); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Unsubscribe_Anonymous_Handler() |
||||
|
{ |
||||
|
var handleCount = 0; |
||||
|
|
||||
|
var handler = new ActionEventHandler<AnonymousEventData>(async (d) => |
||||
|
{ |
||||
|
handleCount++; |
||||
|
await Task.CompletedTask; |
||||
|
}); |
||||
|
var factory = new SingleInstanceHandlerFactory(handler); |
||||
|
|
||||
|
var disposable = LocalEventBus.Subscribe("TestEvent", factory); |
||||
|
|
||||
|
await LocalEventBus.PublishAsync("TestEvent", new { Value = 1 }); |
||||
|
handleCount.ShouldBe(1); |
||||
|
|
||||
|
disposable.Dispose(); |
||||
|
|
||||
|
await Assert.ThrowsAsync<AbpException>(() => |
||||
|
LocalEventBus.PublishAsync("TestEvent", new { Value = 2 })); |
||||
|
handleCount.ShouldBe(1); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_Throw_For_Unknown_Event_Name() |
||||
|
{ |
||||
|
await Assert.ThrowsAsync<AbpException>(() => |
||||
|
LocalEventBus.PublishAsync("NonExistentEvent", new { Value = 1 })); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_ConvertToTypedObject_In_Anonymous_Handler() |
||||
|
{ |
||||
|
object? receivedData = null; |
||||
|
|
||||
|
LocalEventBus.Subscribe("TestEvent", |
||||
|
new SingleInstanceHandlerFactory(new ActionEventHandler<AnonymousEventData>(async (d) => |
||||
|
{ |
||||
|
receivedData = d.ConvertToTypedObject(); |
||||
|
await Task.CompletedTask; |
||||
|
}))); |
||||
|
|
||||
|
await LocalEventBus.PublishAsync("TestEvent", new { Name = "Hello", Count = 42 }); |
||||
|
|
||||
|
receivedData.ShouldNotBeNull(); |
||||
|
var dict = receivedData.ShouldBeOfType<Dictionary<string, object?>>(); |
||||
|
dict["Name"].ShouldBe("Hello"); |
||||
|
dict["Count"].ShouldBe(42L); |
||||
|
} |
||||
|
|
||||
|
[Fact] |
||||
|
public async Task Should_ConvertToTypedObject_Generic_In_Anonymous_Handler() |
||||
|
{ |
||||
|
MySimpleEventData? receivedData = null; |
||||
|
|
||||
|
LocalEventBus.Subscribe("TestEvent", |
||||
|
new SingleInstanceHandlerFactory(new ActionEventHandler<AnonymousEventData>(async (d) => |
||||
|
{ |
||||
|
receivedData = d.ConvertToTypedObject<MySimpleEventData>(); |
||||
|
await Task.CompletedTask; |
||||
|
}))); |
||||
|
|
||||
|
await LocalEventBus.PublishAsync("TestEvent", new MySimpleEventData(99)); |
||||
|
|
||||
|
receivedData.ShouldNotBeNull(); |
||||
|
receivedData.Value.ShouldBe(99); |
||||
|
} |
||||
|
} |
||||
Loading…
Reference in new issue