mirror of
https://github.com/GeWuYou/GFramework.git
synced 2026-05-07 00:39:00 +08:00
feat(cqrs): 添加CQRS命令查询责任分离架构支持
- 实现抽象命令处理器基类支持命令处理 - 添加流式命令处理器基类支持异步流式响应 - 创建查询处理器基类提供统一查询处理接口 - 实现查询基类提供通用查询结构定义 - 扩展架构上下文接口集成CQRS运行时入口 - 定义消息处理器委托支持管道行为处理 - 实现CQRS处理器注册器扫描并注册处理器 - 添加架构模块行为测试验证模块安装功能 - 创建中介器高级特性测试覆盖边界场景
This commit is contained in:
GeWuYou
parent
618f07369e
commit
195c8321a1
@ -1,7 +1,5 @@
|
||||
using GFramework.Core.Abstractions.Command;
|
||||
using GFramework.Core.Abstractions.Cqrs;
|
||||
using GFramework.Core.Abstractions.Cqrs.Command;
|
||||
using GFramework.Core.Abstractions.Cqrs.Query;
|
||||
using GFramework.Core.Abstractions.Environment;
|
||||
using GFramework.Core.Abstractions.Events;
|
||||
using GFramework.Core.Abstractions.Model;
|
||||
@ -13,8 +11,13 @@ using ICommand = GFramework.Core.Abstractions.Command.ICommand;
|
||||
namespace GFramework.Core.Abstractions.Architectures;
|
||||
|
||||
/// <summary>
|
||||
/// 架构上下文接口，提供对系统、模型、工具类的访问以及命令、查询、事件的发送和注册功能
|
||||
/// 架构上下文接口，统一暴露框架组件访问、兼容旧命令/查询总线，以及当前推荐的 CQRS 运行时入口。
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>旧的 <c>GFramework.Core.Abstractions.Command</c> 与 <c>GFramework.Core.Abstractions.Query</c> 契约会继续通过原有 Command/Query Executor 路径执行，以保证存量代码兼容。</para>
|
||||
/// <para>新的 <c>GFramework.Core.Abstractions.Cqrs</c> 契约由内置 CQRS dispatcher 统一处理，支持 request pipeline、notification publish 与 stream request。</para>
|
||||
/// <para>新功能优先使用 <see cref="SendRequestAsync{TResponse}(IRequest{TResponse},CancellationToken)" />、<see cref="SendAsync{TCommand}(TCommand,CancellationToken)" /> 与对应的 CQRS Command/Query 重载；迁移旧代码时可先保留旧入口，再逐步替换为 CQRS 请求模型。</para>
|
||||
/// </remarks>
|
||||
public interface IArchitectureContext
|
||||
{
|
||||
/// <summary>
|
||||
@ -106,85 +109,91 @@ public interface IArchitectureContext
|
||||
IReadOnlyList<TUtility> GetUtilitiesByPriority<TUtility>() where TUtility : class, IUtility;
|
||||
|
||||
/// <summary>
|
||||
/// 发送一个命令
|
||||
/// 发送一个旧版命令。
|
||||
/// </summary>
|
||||
/// <param name="command">要发送的命令</param>
|
||||
/// <param name="command">要发送的旧版命令。</param>
|
||||
void SendCommand(ICommand command);
|
||||
|
||||
/// <summary>
|
||||
/// 发送一个带返回值的命令
|
||||
/// 发送一个旧版带返回值命令。
|
||||
/// </summary>
|
||||
/// <typeparam name="TResult">命令执行结果类型</typeparam>
|
||||
/// <param name="command">要发送的命令</param>
|
||||
/// <returns>命令执行结果</returns>
|
||||
TResult SendCommand<TResult>(Command.ICommand<TResult> command);
|
||||
/// <typeparam name="TResult">命令执行结果类型。</typeparam>
|
||||
/// <param name="command">要发送的旧版命令。</param>
|
||||
/// <returns>命令执行结果。</returns>
|
||||
TResult SendCommand<TResult>(ICommand<TResult> command);
|
||||
|
||||
/// <summary>
|
||||
/// 发送一个 CQRS 命令并返回结果。
|
||||
/// 发送一个新版 CQRS 命令并返回结果。
|
||||
/// </summary>
|
||||
/// <typeparam name="TResponse">命令响应类型。</typeparam>
|
||||
/// <param name="command">要发送的 CQRS 命令。</param>
|
||||
/// <returns>命令执行结果。</returns>
|
||||
TResponse SendCommand<TResponse>(GFramework.Core.Abstractions.Cqrs.Command.ICommand<TResponse> command);
|
||||
/// <remarks>
|
||||
/// 这是迁移后的推荐命令入口。无返回值命令应实现 <c>IRequest<Unit></c>，并优先通过 <see cref="SendAsync{TCommand}(TCommand,CancellationToken)" /> 调用。
|
||||
/// </remarks>
|
||||
TResponse SendCommand<TResponse>(Cqrs.Command.ICommand<TResponse> command);
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// 发送并异步执行一个命令
|
||||
/// 异步发送一个旧版命令。
|
||||
/// </summary>
|
||||
/// <param name="command">要发送的命令</param>
|
||||
/// <param name="command">要发送的旧版命令。</param>
|
||||
Task SendCommandAsync(IAsyncCommand command);
|
||||
|
||||
/// <summary>
|
||||
/// 异步发送一个 CQRS 命令并返回结果。
|
||||
/// 异步发送一个新版 CQRS 命令并返回结果。
|
||||
/// </summary>
|
||||
/// <typeparam name="TResponse">命令响应类型。</typeparam>
|
||||
/// <param name="command">要发送的 CQRS 命令。</param>
|
||||
/// <param name="cancellationToken">取消令牌。</param>
|
||||
/// <returns>包含命令执行结果的值任务。</returns>
|
||||
ValueTask<TResponse> SendCommandAsync<TResponse>(GFramework.Core.Abstractions.Cqrs.Command.ICommand<TResponse> command,
|
||||
ValueTask<TResponse> SendCommandAsync<TResponse>(Cqrs.Command.ICommand<TResponse> command,
|
||||
CancellationToken cancellationToken = default);
|
||||
|
||||
|
||||
/// <summary>
|
||||
/// 发送并异步执行一个带返回值的命令
|
||||
/// 异步发送一个旧版带返回值命令。
|
||||
/// </summary>
|
||||
/// <typeparam name="TResult">命令执行结果类型</typeparam>
|
||||
/// <param name="command">要发送的命令</param>
|
||||
/// <returns>命令执行结果</returns>
|
||||
/// <typeparam name="TResult">命令执行结果类型。</typeparam>
|
||||
/// <param name="command">要发送的旧版命令。</param>
|
||||
/// <returns>命令执行结果。</returns>
|
||||
Task<TResult> SendCommandAsync<TResult>(IAsyncCommand<TResult> command);
|
||||
|
||||
/// <summary>
|
||||
/// 发送一个查询请求
|
||||
/// 发送一个旧版查询请求。
|
||||
/// </summary>
|
||||
/// <typeparam name="TResult">查询结果类型</typeparam>
|
||||
/// <param name="query">要发送的查询</param>
|
||||
/// <returns>查询结果</returns>
|
||||
TResult SendQuery<TResult>(Query.IQuery<TResult> query);
|
||||
/// <typeparam name="TResult">查询结果类型。</typeparam>
|
||||
/// <param name="query">要发送的旧版查询。</param>
|
||||
/// <returns>查询结果。</returns>
|
||||
TResult SendQuery<TResult>(IQuery<TResult> query);
|
||||
|
||||
/// <summary>
|
||||
/// 发送一个 CQRS 查询并返回结果。
|
||||
/// 发送一个新版 CQRS 查询并返回结果。
|
||||
/// </summary>
|
||||
/// <typeparam name="TResponse">查询响应类型。</typeparam>
|
||||
/// <param name="query">要发送的 CQRS 查询。</param>
|
||||
/// <returns>查询结果。</returns>
|
||||
TResponse SendQuery<TResponse>(GFramework.Core.Abstractions.Cqrs.Query.IQuery<TResponse> query);
|
||||
/// <remarks>
|
||||
/// 这是迁移后的推荐查询入口。新查询应优先实现 <c>GFramework.Core.Abstractions.Cqrs.Query.IQuery<TResponse></c>。
|
||||
/// </remarks>
|
||||
TResponse SendQuery<TResponse>(Cqrs.Query.IQuery<TResponse> query);
|
||||
|
||||
/// <summary>
|
||||
/// 异步发送一个查询请求
|
||||
/// 异步发送一个旧版查询请求。
|
||||
/// </summary>
|
||||
/// <typeparam name="TResult">查询结果类型</typeparam>
|
||||
/// <param name="query">要发送的异步查询</param>
|
||||
/// <returns>查询结果</returns>
|
||||
/// <typeparam name="TResult">查询结果类型。</typeparam>
|
||||
/// <param name="query">要发送的旧版异步查询。</param>
|
||||
/// <returns>查询结果。</returns>
|
||||
Task<TResult> SendQueryAsync<TResult>(IAsyncQuery<TResult> query);
|
||||
|
||||
/// <summary>
|
||||
/// 异步发送一个 CQRS 查询并返回结果。
|
||||
/// 异步发送一个新版 CQRS 查询并返回结果。
|
||||
/// </summary>
|
||||
/// <typeparam name="TResponse">查询响应类型。</typeparam>
|
||||
/// <param name="query">要发送的 CQRS 查询。</param>
|
||||
/// <param name="cancellationToken">取消令牌。</param>
|
||||
/// <returns>包含查询结果的值任务。</returns>
|
||||
ValueTask<TResponse> SendQueryAsync<TResponse>(GFramework.Core.Abstractions.Cqrs.Query.IQuery<TResponse> query,
|
||||
ValueTask<TResponse> SendQueryAsync<TResponse>(Cqrs.Query.IQuery<TResponse> query,
|
||||
CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
@ -216,28 +225,40 @@ public interface IArchitectureContext
|
||||
void UnRegisterEvent<TEvent>(Action<TEvent> onEvent);
|
||||
|
||||
/// <summary>
|
||||
/// 发送请求（统一处理 Command/Query）
|
||||
/// 发送新版 CQRS 请求，并统一处理命令与查询。
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 这是自有 CQRS 运行时的主入口。新代码应优先通过该方法或 <see cref="SendAsync{TCommand}(TCommand,CancellationToken)" /> 进入 dispatcher。
|
||||
/// </remarks>
|
||||
ValueTask<TResponse> SendRequestAsync<TResponse>(
|
||||
IRequest<TResponse> request,
|
||||
CancellationToken cancellationToken = default);
|
||||
|
||||
/// <summary>
|
||||
/// 发送请求（同步版本，不推荐）
|
||||
/// 发送新版 CQRS 请求的同步包装版本。
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 仅为兼容同步调用链保留；新代码应优先使用异步入口，避免阻塞当前线程。
|
||||
/// </remarks>
|
||||
TResponse SendRequest<TResponse>(IRequest<TResponse> request);
|
||||
|
||||
/// <summary>
|
||||
/// 发布通知（一对多事件）
|
||||
/// 发布新版 CQRS 通知。
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 该入口用于一对多通知分发，与框架级 <c>EventBus</c> 事件系统并存，适合围绕请求处理过程传播领域通知。
|
||||
/// </remarks>
|
||||
ValueTask PublishAsync<TNotification>(
|
||||
TNotification notification,
|
||||
CancellationToken cancellationToken = default)
|
||||
where TNotification : INotification;
|
||||
|
||||
/// <summary>
|
||||
/// 创建流式请求（用于大数据集）
|
||||
/// 创建新版 CQRS 流式请求。
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// 适用于需要按序惰性产出大量结果的场景。调用方应消费返回的异步序列，而不是回退到旧版查询总线。
|
||||
/// </remarks>
|
||||
IAsyncEnumerable<TResponse> CreateStream<TResponse>(
|
||||
IStreamRequest<TResponse> request,
|
||||
CancellationToken cancellationToken = default);
|
||||
@ -245,7 +266,7 @@ public interface IArchitectureContext
|
||||
// === 便捷扩展方法 ===
|
||||
|
||||
/// <summary>
|
||||
/// 发送命令（无返回值）
|
||||
/// 发送一个无返回值的新版 CQRS 命令。
|
||||
/// </summary>
|
||||
ValueTask SendAsync<TCommand>(
|
||||
TCommand command,
|
||||
@ -253,7 +274,7 @@ public interface IArchitectureContext
|
||||
where TCommand : IRequest<Unit>;
|
||||
|
||||
/// <summary>
|
||||
/// 发送命令（有返回值）
|
||||
/// 发送一个有返回值的新版 CQRS 请求。
|
||||
/// </summary>
|
||||
ValueTask<TResponse> SendAsync<TResponse>(
|
||||
IRequest<TResponse> command,
|
||||
|
||||
@ -3,6 +3,11 @@ namespace GFramework.Core.Abstractions.Cqrs;
|
||||
/// <summary>
|
||||
/// 表示 CQRS 请求在管道中继续向下执行的处理委托。
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>管道行为可以通过不调用该委托来短路请求处理。</para>
|
||||
/// <para>除显式实现重试等高级语义外，行为通常应最多调用一次该委托，以维持单次请求分发的确定性。</para>
|
||||
/// <para>调用方应传递当前收到的 <paramref name="cancellationToken" />，确保取消信号沿整条管道一致传播。</para>
|
||||
/// </remarks>
|
||||
/// <typeparam name="TRequest">请求类型。</typeparam>
|
||||
/// <typeparam name="TResponse">响应类型。</typeparam>
|
||||
/// <param name="message">当前请求消息。</param>
|
||||
|
||||
@ -2,7 +2,6 @@ using GFramework.Core.Abstractions.Architectures;
|
||||
using GFramework.Core.Abstractions.Utility;
|
||||
using GFramework.Core.Architectures;
|
||||
using GFramework.Core.Logging;
|
||||
using GFramework.Core.Tests;
|
||||
using GfCqrs = GFramework.Core.Abstractions.Cqrs;
|
||||
|
||||
namespace GFramework.Core.Tests.Architectures;
|
||||
@ -67,9 +66,7 @@ public class ArchitectureModulesBehaviorTests
|
||||
|
||||
await architecture.InitializeAsync();
|
||||
|
||||
var response = await CqrsTestRuntime.ExecutePipelineAsync<ModuleBehaviorRequest, string>(
|
||||
architecture.Context,
|
||||
new ModuleBehaviorRequest());
|
||||
var response = await architecture.Context.SendRequestAsync(new ModuleBehaviorRequest());
|
||||
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
@ -174,8 +171,7 @@ public sealed class TrackingPipelineBehavior<TRequest, TResponse> : GfCqrs.IPipe
|
||||
/// <param name="cancellationToken">取消令牌。</param>
|
||||
/// <returns>下游处理器的响应结果。</returns>
|
||||
public async ValueTask<TResponse> Handle(
|
||||
TRequest message,
|
||||
GfCqrs.MessageHandlerDelegate<TRequest, TResponse> next,
|
||||
TRequest message, GfCqrs.MessageHandlerDelegate<TRequest, TResponse> next,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
InvocationCount++;
|
||||
|
||||
165
GFramework.Core.Tests/Cqrs/CqrsHandlerRegistrarTests.cs
Normal file
165
GFramework.Core.Tests/Cqrs/CqrsHandlerRegistrarTests.cs
Normal file
@ -0,0 +1,165 @@
|
||||
using System.Reflection;
|
||||
using GFramework.Core.Abstractions.Cqrs;
|
||||
using GFramework.Core.Abstractions.Logging;
|
||||
using GFramework.Core.Architectures;
|
||||
using GFramework.Core.Ioc;
|
||||
using GFramework.Core.Logging;
|
||||
using GFramework.Core.Tests.Logging;
|
||||
|
||||
namespace GFramework.Core.Tests.Cqrs;
|
||||
|
||||
/// <summary>
|
||||
/// 验证 CQRS 处理器自动注册在顺序与容错层面的可观察行为。
|
||||
/// </summary>
|
||||
[TestFixture]
|
||||
internal sealed class CqrsHandlerRegistrarTests
|
||||
{
|
||||
private static readonly MethodInfo RecoverLoadableTypesMethod = typeof(ArchitectureContext).Assembly
|
||||
.GetType(
|
||||
"GFramework.Core.Cqrs.Internal.CqrsHandlerRegistrar",
|
||||
throwOnError: true)!
|
||||
.GetMethod("RecoverLoadableTypes",
|
||||
BindingFlags.NonPublic |
|
||||
BindingFlags.Static)!
|
||||
?? throw new InvalidOperationException(
|
||||
"Failed to locate CqrsHandlerRegistrar.RecoverLoadableTypes.");
|
||||
|
||||
private MicrosoftDiContainer? _container;
|
||||
|
||||
private ArchitectureContext? _context;
|
||||
|
||||
/// <summary>
|
||||
/// 初始化测试容器并重置共享状态。
|
||||
/// </summary>
|
||||
[SetUp]
|
||||
public void SetUp()
|
||||
{
|
||||
LoggerFactoryResolver.Provider = new ConsoleLoggerFactoryProvider();
|
||||
DeterministicNotificationHandlerState.Reset();
|
||||
|
||||
_container = new MicrosoftDiContainer();
|
||||
CqrsTestRuntime.RegisterHandlers(
|
||||
_container,
|
||||
typeof(CqrsHandlerRegistrarTests).Assembly,
|
||||
typeof(ArchitectureContext).Assembly);
|
||||
|
||||
_container.Freeze();
|
||||
_context = new ArchitectureContext(_container);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 清理测试过程中创建的上下文与共享状态。
|
||||
/// </summary>
|
||||
[TearDown]
|
||||
public void TearDown()
|
||||
{
|
||||
_context = null;
|
||||
_container = null;
|
||||
DeterministicNotificationHandlerState.Reset();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 验证自动扫描到的通知处理器会按稳定名称顺序执行，而不是依赖反射枚举顺序。
|
||||
/// </summary>
|
||||
[Test]
|
||||
public async Task PublishAsync_Should_Run_Notification_Handlers_In_Deterministic_Name_Order()
|
||||
{
|
||||
await _context!.PublishAsync(new DeterministicOrderNotification());
|
||||
|
||||
Assert.That(
|
||||
DeterministicNotificationHandlerState.InvocationOrder,
|
||||
Is.EqualTo(
|
||||
[
|
||||
nameof(AlphaDeterministicNotificationHandler),
|
||||
nameof(ZetaDeterministicNotificationHandler)
|
||||
]));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 验证部分类型加载失败时仍能保留可加载类型，并记录诊断日志。
|
||||
/// </summary>
|
||||
[Test]
|
||||
public void RecoverLoadableTypes_Should_Return_Loadable_Types_And_Log_Warnings()
|
||||
{
|
||||
var logger = new TestLogger(nameof(CqrsHandlerRegistrarTests), LogLevel.Warning);
|
||||
var reflectionTypeLoadException = new ReflectionTypeLoadException(
|
||||
[typeof(AlphaDeterministicNotificationHandler), null],
|
||||
[new TypeLoadException("Missing optional dependency for registrar test.")]);
|
||||
|
||||
var recoveredTypes = (IReadOnlyList<Type>)RecoverLoadableTypesMethod.Invoke(
|
||||
null,
|
||||
[typeof(CqrsHandlerRegistrarTests).Assembly, reflectionTypeLoadException, logger])!;
|
||||
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(recoveredTypes, Is.EqualTo([typeof(AlphaDeterministicNotificationHandler)]));
|
||||
Assert.That(logger.Logs.Count(log => log.Level == LogLevel.Warning), Is.GreaterThanOrEqualTo(2));
|
||||
Assert.That(
|
||||
logger.Logs.Any(log => log.Message.Contains("partially failed", StringComparison.Ordinal)),
|
||||
Is.True);
|
||||
Assert.That(
|
||||
logger.Logs.Any(log => log.Message.Contains("Missing optional dependency", StringComparison.Ordinal)),
|
||||
Is.True);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 记录确定性通知处理器的实际执行顺序。
|
||||
/// </summary>
|
||||
internal static class DeterministicNotificationHandlerState
|
||||
{
|
||||
/// <summary>
|
||||
/// 获取当前测试中的通知处理器执行顺序。
|
||||
/// </summary>
|
||||
public static List<string> InvocationOrder { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// 重置共享的执行顺序状态。
|
||||
/// </summary>
|
||||
public static void Reset()
|
||||
{
|
||||
InvocationOrder.Clear();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 用于验证同一通知的多个处理器是否按稳定顺序执行。
|
||||
/// </summary>
|
||||
internal sealed record DeterministicOrderNotification : INotification;
|
||||
|
||||
/// <summary>
|
||||
/// 故意放在 Alpha 之前声明，用于验证注册器不会依赖源码声明顺序。
|
||||
/// </summary>
|
||||
internal sealed class ZetaDeterministicNotificationHandler : INotificationHandler<DeterministicOrderNotification>
|
||||
{
|
||||
/// <summary>
|
||||
/// 记录当前处理器已执行。
|
||||
/// </summary>
|
||||
/// <param name="notification">通知实例。</param>
|
||||
/// <param name="cancellationToken">取消令牌。</param>
|
||||
/// <returns>已完成任务。</returns>
|
||||
public ValueTask Handle(DeterministicOrderNotification notification, CancellationToken cancellationToken)
|
||||
{
|
||||
DeterministicNotificationHandlerState.InvocationOrder.Add(nameof(ZetaDeterministicNotificationHandler));
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 名称排序上应先于 Zeta 处理器执行的通知处理器。
|
||||
/// </summary>
|
||||
internal sealed class AlphaDeterministicNotificationHandler : INotificationHandler<DeterministicOrderNotification>
|
||||
{
|
||||
/// <summary>
|
||||
/// 记录当前处理器已执行。
|
||||
/// </summary>
|
||||
/// <param name="notification">通知实例。</param>
|
||||
/// <param name="cancellationToken">取消令牌。</param>
|
||||
/// <returns>已完成任务。</returns>
|
||||
public ValueTask Handle(DeterministicOrderNotification notification, CancellationToken cancellationToken)
|
||||
{
|
||||
DeterministicNotificationHandlerState.InvocationOrder.Add(nameof(AlphaDeterministicNotificationHandler));
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
}
|
||||
@ -1,22 +1,22 @@
|
||||
using System.Reflection;
|
||||
using GFramework.Core.Abstractions.Architectures;
|
||||
using GFramework.Core.Abstractions.Logging;
|
||||
using GFramework.Core.Abstractions.Rule;
|
||||
using GFramework.Core.Architectures;
|
||||
using GFramework.Core.Ioc;
|
||||
using GFramework.Core.Logging;
|
||||
using GfCqrs = GFramework.Core.Abstractions.Cqrs;
|
||||
|
||||
namespace GFramework.Core.Tests;
|
||||
|
||||
internal static class CqrsTestRuntime
|
||||
{
|
||||
private static readonly MethodInfo RegisterHandlersMethod = typeof(ArchitectureContext).Assembly
|
||||
.GetType("GFramework.Core.Cqrs.Internal.CqrsHandlerRegistrar", throwOnError: true)!
|
||||
.GetMethod(
|
||||
"RegisterHandlers",
|
||||
BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Static)!
|
||||
?? throw new InvalidOperationException("Failed to locate CqrsHandlerRegistrar.RegisterHandlers.");
|
||||
.GetType(
|
||||
"GFramework.Core.Cqrs.Internal.CqrsHandlerRegistrar",
|
||||
throwOnError: true)!
|
||||
.GetMethod(
|
||||
"RegisterHandlers",
|
||||
BindingFlags.Public | BindingFlags.NonPublic |
|
||||
BindingFlags.Static)!
|
||||
?? throw new InvalidOperationException(
|
||||
"Failed to locate CqrsHandlerRegistrar.RegisterHandlers.");
|
||||
|
||||
public static void RegisterHandlers(MicrosoftDiContainer container, params Assembly[] assemblies)
|
||||
{
|
||||
@ -28,46 +28,4 @@ internal static class CqrsTestRuntime
|
||||
null,
|
||||
[container, assemblies.Where(static assembly => assembly is not null).Distinct().ToArray(), logger]);
|
||||
}
|
||||
|
||||
public static ValueTask<TResponse> ExecutePipelineAsync<TRequest, TResponse>(
|
||||
IArchitectureContext context,
|
||||
TRequest request,
|
||||
CancellationToken cancellationToken = default)
|
||||
where TRequest : class, GfCqrs.IRequest<TResponse>
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(context);
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
|
||||
var handlers = context.GetServices<GfCqrs.IRequestHandler<TRequest, TResponse>>();
|
||||
if (handlers.Count == 0)
|
||||
throw new InvalidOperationException(
|
||||
$"No CQRS request handler registered for {typeof(TRequest).FullName}.");
|
||||
|
||||
if (handlers.Count > 1)
|
||||
throw new InvalidOperationException(
|
||||
$"Expected a single CQRS request handler for {typeof(TRequest).FullName}, but found {handlers.Count}.");
|
||||
|
||||
var handler = handlers[0];
|
||||
PrepareContext(handler, context);
|
||||
|
||||
GfCqrs.MessageHandlerDelegate<TRequest, TResponse> pipeline = handler.Handle;
|
||||
|
||||
var behaviors = context.GetServices<GfCqrs.IPipelineBehavior<TRequest, TResponse>>();
|
||||
for (var index = behaviors.Count - 1; index >= 0; index--)
|
||||
{
|
||||
var behavior = behaviors[index];
|
||||
PrepareContext(behavior, context);
|
||||
|
||||
var next = pipeline;
|
||||
pipeline = (message, token) => behavior.Handle(message, next, token);
|
||||
}
|
||||
|
||||
return pipeline(request, cancellationToken);
|
||||
}
|
||||
|
||||
private static void PrepareContext(object instance, IArchitectureContext context)
|
||||
{
|
||||
if (instance is IContextAware contextAware)
|
||||
contextAware.SetContext(context);
|
||||
}
|
||||
}
|
||||
|
||||
@ -4,7 +4,6 @@ using GFramework.Core.Abstractions.Cqrs;
|
||||
using GFramework.Core.Architectures;
|
||||
using GFramework.Core.Ioc;
|
||||
using GFramework.Core.Logging;
|
||||
using GFramework.Core.Tests;
|
||||
|
||||
namespace GFramework.Core.Tests.Mediator;
|
||||
|
||||
@ -15,11 +14,16 @@ namespace GFramework.Core.Tests.Mediator;
|
||||
[TestFixture]
|
||||
public class MediatorAdvancedFeaturesTests
|
||||
{
|
||||
private MicrosoftDiContainer? _container;
|
||||
|
||||
private ArchitectureContext? _context;
|
||||
|
||||
[SetUp]
|
||||
public void SetUp()
|
||||
{
|
||||
LoggerFactoryResolver.Provider = new ConsoleLoggerFactoryProvider();
|
||||
_container = new MicrosoftDiContainer();
|
||||
TestCircuitBreakerHandler.Reset();
|
||||
|
||||
var loggerField = typeof(MicrosoftDiContainer).GetField("_logger",
|
||||
BindingFlags.NonPublic | BindingFlags.Instance);
|
||||
@ -42,9 +46,6 @@ public class MediatorAdvancedFeaturesTests
|
||||
_container = null;
|
||||
}
|
||||
|
||||
private ArchitectureContext? _context;
|
||||
private MicrosoftDiContainer? _container;
|
||||
|
||||
|
||||
[Test]
|
||||
public async Task Request_With_Validation_Behavior_Should_Validate_Input()
|
||||
@ -135,9 +136,6 @@ public class MediatorAdvancedFeaturesTests
|
||||
[Test]
|
||||
public async Task Circuit_Breaker_Should_Prevent_Cascading_Failures()
|
||||
{
|
||||
TestCircuitBreakerHandler.FailureCount = 0;
|
||||
TestCircuitBreakerHandler.SuccessCount = 0;
|
||||
|
||||
// 先触发几次失败
|
||||
for (int i = 0; i < 5; i++)
|
||||
{
|
||||
@ -275,12 +273,10 @@ public sealed class TestTransientErrorRequestHandler : IRequestHandler<TestTrans
|
||||
|
||||
public sealed class TestCircuitBreakerRequestHandler : IRequestHandler<TestCircuitBreakerRequest, string>
|
||||
{
|
||||
private static bool _circuitOpen = false;
|
||||
|
||||
public ValueTask<string> Handle(TestCircuitBreakerRequest request, CancellationToken cancellationToken)
|
||||
{
|
||||
// 检查断路器状态
|
||||
if (_circuitOpen)
|
||||
if (TestCircuitBreakerHandler.CircuitOpen)
|
||||
{
|
||||
throw new InvalidOperationException("Circuit breaker is open");
|
||||
}
|
||||
@ -292,7 +288,7 @@ public sealed class TestCircuitBreakerRequestHandler : IRequestHandler<TestCircu
|
||||
// 达到阈值后打开断路器
|
||||
if (TestCircuitBreakerHandler.FailureCount >= 5)
|
||||
{
|
||||
_circuitOpen = true;
|
||||
TestCircuitBreakerHandler.CircuitOpen = true;
|
||||
}
|
||||
|
||||
throw new InvalidOperationException("Service unavailable");
|
||||
@ -451,6 +447,17 @@ public static class TestCircuitBreakerHandler
|
||||
{
|
||||
public static int FailureCount { get; set; }
|
||||
public static int SuccessCount { get; set; }
|
||||
public static bool CircuitOpen { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 重置断路器测试状态，避免静态字段在测试之间互相污染。
|
||||
/// </summary>
|
||||
public static void Reset()
|
||||
{
|
||||
FailureCount = 0;
|
||||
SuccessCount = 0;
|
||||
CircuitOpen = false;
|
||||
}
|
||||
}
|
||||
|
||||
public sealed record TestCircuitBreakerRequest : IRequest<string>
|
||||
|
||||
@ -6,7 +6,7 @@ using GFramework.Core.Architectures;
|
||||
using GFramework.Core.Command;
|
||||
using GFramework.Core.Ioc;
|
||||
using GFramework.Core.Logging;
|
||||
using GFramework.Core.Tests;
|
||||
using GFramework.Core.Rule;
|
||||
using ICommand = GFramework.Core.Abstractions.Command.ICommand;
|
||||
|
||||
namespace GFramework.Core.Tests.Mediator;
|
||||
@ -18,11 +18,17 @@ namespace GFramework.Core.Tests.Mediator;
|
||||
[TestFixture]
|
||||
public class MediatorArchitectureIntegrationTests
|
||||
{
|
||||
private CommandExecutor? _commandBus;
|
||||
private MicrosoftDiContainer? _container;
|
||||
|
||||
private ArchitectureContext? _context;
|
||||
|
||||
[SetUp]
|
||||
public void SetUp()
|
||||
{
|
||||
LoggerFactoryResolver.Provider = new ConsoleLoggerFactoryProvider();
|
||||
_container = new MicrosoftDiContainer();
|
||||
TestPerDispatchContextAwareHandler.Reset();
|
||||
|
||||
var loggerField = typeof(MicrosoftDiContainer).GetField("_logger",
|
||||
BindingFlags.NonPublic | BindingFlags.Instance);
|
||||
@ -50,10 +56,6 @@ public class MediatorArchitectureIntegrationTests
|
||||
_commandBus = null;
|
||||
}
|
||||
|
||||
private ArchitectureContext? _context;
|
||||
private MicrosoftDiContainer? _container;
|
||||
private CommandExecutor? _commandBus;
|
||||
|
||||
[Test]
|
||||
public async Task Handler_Can_Access_Architecture_Context()
|
||||
{
|
||||
@ -291,6 +293,20 @@ public class MediatorArchitectureIntegrationTests
|
||||
Assert.That(traditionalCommand.Executed, Is.True);
|
||||
Assert.That(result, Is.EqualTo(42));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task ContextAware_Handler_Should_Use_A_Fresh_Instance_Per_Request()
|
||||
{
|
||||
var firstResult = await _context!.SendRequestAsync(new TestPerDispatchContextAwareRequest());
|
||||
var secondResult = await _context.SendRequestAsync(new TestPerDispatchContextAwareRequest());
|
||||
|
||||
Assert.Multiple(() =>
|
||||
{
|
||||
Assert.That(firstResult, Is.Not.EqualTo(secondResult));
|
||||
Assert.That(TestPerDispatchContextAwareHandler.SeenInstanceIds, Is.EqualTo([firstResult, secondResult]));
|
||||
Assert.That(TestPerDispatchContextAwareHandler.Contexts, Has.All.SameAs(_context));
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
#region Integration Test Classes
|
||||
@ -444,6 +460,42 @@ public sealed class TestMediatorRequestHandler : IRequestHandler<TestMediatorReq
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 用于验证自动扫描到的上下文感知处理器会按请求创建新实例。
|
||||
/// </summary>
|
||||
public sealed class TestPerDispatchContextAwareHandler : ContextAwareBase,
|
||||
IRequestHandler<TestPerDispatchContextAwareRequest, int>
|
||||
{
|
||||
private static int _nextInstanceId;
|
||||
private readonly int _instanceId = Interlocked.Increment(ref _nextInstanceId);
|
||||
|
||||
public static List<IArchitectureContext?> Contexts { get; } = [];
|
||||
public static List<int> SeenInstanceIds { get; } = [];
|
||||
|
||||
/// <summary>
|
||||
/// 记录当前实例编号与收到的架构上下文。
|
||||
/// </summary>
|
||||
/// <param name="request">请求实例。</param>
|
||||
/// <param name="cancellationToken">取消令牌。</param>
|
||||
/// <returns>当前处理器实例编号。</returns>
|
||||
public ValueTask<int> Handle(TestPerDispatchContextAwareRequest request, CancellationToken cancellationToken)
|
||||
{
|
||||
Contexts.Add(Context);
|
||||
SeenInstanceIds.Add(_instanceId);
|
||||
return ValueTask.FromResult(_instanceId);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 重置跨测试共享的实例跟踪状态。
|
||||
/// </summary>
|
||||
public static void Reset()
|
||||
{
|
||||
Contexts.Clear();
|
||||
SeenInstanceIds.Clear();
|
||||
_nextInstanceId = 0;
|
||||
}
|
||||
}
|
||||
|
||||
public sealed record TestContextAwareRequest : IRequest<string>;
|
||||
|
||||
public static class TestContextAwareHandler
|
||||
@ -544,6 +596,11 @@ public sealed record TestMediatorRequest : IRequest<int>
|
||||
public int Value { get; init; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 用于验证每次请求分发都会获得新的上下文感知处理器实例。
|
||||
/// </summary>
|
||||
public sealed record TestPerDispatchContextAwareRequest : IRequest<int>;
|
||||
|
||||
// 传统命令用于混合测试
|
||||
public class TestTraditionalCommand : ICommand
|
||||
{
|
||||
|
||||
@ -11,7 +11,6 @@ using GFramework.Core.Events;
|
||||
using GFramework.Core.Ioc;
|
||||
using GFramework.Core.Logging;
|
||||
using GFramework.Core.Query;
|
||||
using GFramework.Core.Tests;
|
||||
using ICommand = GFramework.Core.Abstractions.Command.ICommand;
|
||||
using Unit = GFramework.Core.Abstractions.Cqrs.Unit;
|
||||
|
||||
@ -20,6 +19,15 @@ namespace GFramework.Core.Tests.Mediator;
|
||||
[TestFixture]
|
||||
public class MediatorComprehensiveTests
|
||||
{
|
||||
private AsyncQueryExecutor? _asyncQueryBus;
|
||||
private CommandExecutor? _commandBus;
|
||||
private MicrosoftDiContainer? _container;
|
||||
|
||||
private ArchitectureContext? _context;
|
||||
private DefaultEnvironment? _environment;
|
||||
private EventBus? _eventBus;
|
||||
private QueryExecutor? _queryBus;
|
||||
|
||||
/// <summary>
|
||||
/// 测试初始化方法，在每个测试方法执行前运行。
|
||||
/// 负责初始化日志工厂、依赖注入容器、自有 CQRS 处理器以及各种总线服务。
|
||||
@ -74,14 +82,6 @@ public class MediatorComprehensiveTests
|
||||
_environment = null;
|
||||
}
|
||||
|
||||
private ArchitectureContext? _context;
|
||||
private MicrosoftDiContainer? _container;
|
||||
private EventBus? _eventBus;
|
||||
private CommandExecutor? _commandBus;
|
||||
private QueryExecutor? _queryBus;
|
||||
private AsyncQueryExecutor? _asyncQueryBus;
|
||||
private DefaultEnvironment? _environment;
|
||||
|
||||
/// <summary>
|
||||
/// 测试SendRequestAsync方法在请求有效时返回结果
|
||||
/// </summary>
|
||||
@ -418,7 +418,7 @@ public class MediatorComprehensiveTests
|
||||
_context!.SendCommand(legacyCommand);
|
||||
Assert.That(legacyCommand.Executed, Is.True);
|
||||
|
||||
// 使用Mediator方式
|
||||
// 使用自有 CQRS 方式
|
||||
var mediatorCommand = new TestCommandWithResult { ResultValue = 999 };
|
||||
var result = await _context.SendAsync(mediatorCommand);
|
||||
Assert.That(result, Is.EqualTo(999));
|
||||
@ -429,7 +429,7 @@ public class MediatorComprehensiveTests
|
||||
}
|
||||
}
|
||||
|
||||
#region Advanced Test Classes for Mediator Features
|
||||
#region Advanced Test Classes for CQRS Features
|
||||
|
||||
public sealed record TestLongRunningRequest : IRequest<string>
|
||||
{
|
||||
@ -623,9 +623,9 @@ public class TestLegacyCommand : ICommand
|
||||
|
||||
#endregion
|
||||
|
||||
#region Test Classes - Mediator (新实现)
|
||||
#region Test Classes - CQRS Runtime
|
||||
|
||||
// ✅ 这些类使用 Mediator.IRequest
|
||||
// ✅ 这些类使用自有 CQRS IRequest
|
||||
public sealed record TestRequest : IRequest<int>
|
||||
{
|
||||
public int Value { get; init; }
|
||||
@ -657,7 +657,7 @@ public sealed record TestStreamRequest : IStreamRequest<int>
|
||||
public int[] Values { get; init; } = [];
|
||||
}
|
||||
|
||||
// ✅ 这些 Handler 使用 Mediator.IRequestHandler
|
||||
// ✅ 这些 Handler 使用自有 CQRS IRequestHandler
|
||||
public sealed class TestRequestHandler : IRequestHandler<TestRequest, int>
|
||||
{
|
||||
public ValueTask<int> Handle(TestRequest request, CancellationToken cancellationToken)
|
||||
|
||||
@ -11,15 +11,16 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
using GFramework.Core.Rule;
|
||||
using GFramework.Core.Abstractions.Cqrs;
|
||||
using GFramework.Core.Abstractions.Cqrs.Command;
|
||||
using GFramework.Core.Rule;
|
||||
|
||||
namespace GFramework.Core.Cqrs.Command;
|
||||
|
||||
/// <summary>
|
||||
/// 抽象命令处理器基类
|
||||
/// 继承自ContextAwareBase并实现ICommandHandler接口，为具体的命令处理器提供基础功能
|
||||
/// 继承自 ContextAwareBase 并实现 IRequestHandler 接口，为具体的命令处理器提供基础功能。
|
||||
/// 框架会在每次分发前注入当前架构上下文，因此派生类可以通过 Context 访问架构级服务。
|
||||
/// </summary>
|
||||
/// <typeparam name="TCommand">命令类型</typeparam>
|
||||
public abstract class AbstractCommandHandler<TCommand> : ContextAwareBase, IRequestHandler<TCommand, Unit>
|
||||
@ -37,8 +38,8 @@ public abstract class AbstractCommandHandler<TCommand> : ContextAwareBase, IRequ
|
||||
|
||||
/// <summary>
|
||||
/// 抽象命令处理器基类（带返回值版本）
|
||||
/// 继承自ContextAwareBase并实现ICommandHandler接口，为具体的命令处理器提供基础功能
|
||||
/// 支持泛型命令和结果类型，实现CQRS模式中的命令处理
|
||||
/// 继承自 ContextAwareBase 并实现 IRequestHandler 接口，为具体的命令处理器提供基础功能。
|
||||
/// 支持泛型命令和结果类型，框架会在每次分发前注入当前架构上下文。
|
||||
/// </summary>
|
||||
/// <typeparam name="TCommand">命令类型，必须实现ICommand接口</typeparam>
|
||||
/// <typeparam name="TResult">命令执行结果类型</typeparam>
|
||||
|
||||
@ -11,16 +11,16 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
using GFramework.Core.Rule;
|
||||
using GFramework.Core.Abstractions.Cqrs;
|
||||
using GFramework.Core.Abstractions.Cqrs.Command;
|
||||
using GFramework.Core.Rule;
|
||||
|
||||
namespace GFramework.Core.Cqrs.Command;
|
||||
|
||||
/// <summary>
|
||||
/// 抽象流式命令处理器基类
|
||||
/// 继承自ContextAwareBase并实现IStreamCommandHandler接口，为具体的流式命令处理器提供基础功能
|
||||
/// 支持流式处理命令并产生异步可枚举的响应序列
|
||||
/// 继承自 ContextAwareBase 并实现 IStreamRequestHandler 接口，为具体的流式命令处理器提供基础功能。
|
||||
/// 支持流式处理命令并产生异步可枚举的响应序列，框架会在每次创建流前注入当前架构上下文。
|
||||
/// </summary>
|
||||
/// <typeparam name="TCommand">流式命令类型，必须实现IStreamCommand接口</typeparam>
|
||||
/// <typeparam name="TResponse">流式命令响应元素类型</typeparam>
|
||||
|
||||
@ -2,7 +2,6 @@ using System.Reflection;
|
||||
using GFramework.Core.Abstractions.Cqrs;
|
||||
using GFramework.Core.Abstractions.Ioc;
|
||||
using GFramework.Core.Abstractions.Logging;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
|
||||
namespace GFramework.Core.Cqrs.Internal;
|
||||
|
||||
@ -27,7 +26,10 @@ internal static class CqrsHandlerRegistrar
|
||||
ArgumentNullException.ThrowIfNull(assemblies);
|
||||
ArgumentNullException.ThrowIfNull(logger);
|
||||
|
||||
foreach (var assembly in assemblies.Distinct())
|
||||
foreach (var assembly in assemblies
|
||||
.Where(static assembly => assembly is not null)
|
||||
.Distinct()
|
||||
.OrderBy(GetAssemblySortKey, StringComparer.Ordinal))
|
||||
{
|
||||
RegisterAssemblyHandlers(container.GetServicesUnsafe, assembly, logger);
|
||||
}
|
||||
@ -38,11 +40,12 @@ internal static class CqrsHandlerRegistrar
|
||||
/// </summary>
|
||||
private static void RegisterAssemblyHandlers(IServiceCollection services, Assembly assembly, ILogger logger)
|
||||
{
|
||||
foreach (var implementationType in assembly.GetTypes().Where(IsConcreteHandlerType))
|
||||
foreach (var implementationType in GetLoadableTypes(assembly, logger).Where(IsConcreteHandlerType))
|
||||
{
|
||||
var handlerInterfaces = implementationType
|
||||
.GetInterfaces()
|
||||
.Where(IsSupportedHandlerInterface)
|
||||
.OrderBy(GetTypeSortKey, StringComparer.Ordinal)
|
||||
.ToList();
|
||||
|
||||
if (handlerInterfaces.Count == 0)
|
||||
@ -50,13 +53,58 @@ internal static class CqrsHandlerRegistrar
|
||||
|
||||
foreach (var handlerInterface in handlerInterfaces)
|
||||
{
|
||||
services.AddSingleton(handlerInterface, implementationType);
|
||||
// Request/notification handlers receive context injection before every dispatch.
|
||||
// Transient registration avoids sharing mutable Context across concurrent requests.
|
||||
services.AddTransient(handlerInterface, implementationType);
|
||||
logger.Debug(
|
||||
$"Registered CQRS handler {implementationType.FullName} as {handlerInterface.FullName}.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 安全获取程序集中的可加载类型，并在部分类型加载失败时保留其余处理器注册能力。
|
||||
/// </summary>
|
||||
private static IReadOnlyList<Type> GetLoadableTypes(Assembly assembly, ILogger logger)
|
||||
{
|
||||
try
|
||||
{
|
||||
return assembly.GetTypes()
|
||||
.Where(static type => type is not null)
|
||||
.OrderBy(GetTypeSortKey, StringComparer.Ordinal)
|
||||
.ToList();
|
||||
}
|
||||
catch (ReflectionTypeLoadException exception)
|
||||
{
|
||||
return RecoverLoadableTypes(assembly, exception, logger);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 记录部分类型加载失败，并返回仍然可用的类型集合。
|
||||
/// </summary>
|
||||
private static IReadOnlyList<Type> RecoverLoadableTypes(
|
||||
Assembly assembly,
|
||||
ReflectionTypeLoadException exception,
|
||||
ILogger logger)
|
||||
{
|
||||
var assemblyName = GetAssemblySortKey(assembly);
|
||||
logger.Warn(
|
||||
$"CQRS handler scan partially failed for assembly {assemblyName}. Continuing with loadable types.");
|
||||
|
||||
foreach (var loaderException in exception.LoaderExceptions.Where(static ex => ex is not null))
|
||||
{
|
||||
logger.Warn(
|
||||
$"Failed to load one or more types while scanning {assemblyName}: {loaderException!.Message}");
|
||||
}
|
||||
|
||||
return exception.Types
|
||||
.Where(static type => type is not null)
|
||||
.Cast<Type>()
|
||||
.OrderBy(GetTypeSortKey, StringComparer.Ordinal)
|
||||
.ToList();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 判断指定类型是否可作为可实例化处理器。
|
||||
/// </summary>
|
||||
@ -78,4 +126,20 @@ internal static class CqrsHandlerRegistrar
|
||||
definition == typeof(INotificationHandler<>) ||
|
||||
definition == typeof(IStreamRequestHandler<,>);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 生成程序集排序键，保证跨运行环境的处理器注册顺序稳定。
|
||||
/// </summary>
|
||||
private static string GetAssemblySortKey(Assembly assembly)
|
||||
{
|
||||
return assembly.FullName ?? assembly.GetName().Name ?? assembly.ToString();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 生成类型排序键，保证同一程序集内的处理器与接口映射顺序稳定。
|
||||
/// </summary>
|
||||
private static string GetTypeSortKey(Type type)
|
||||
{
|
||||
return type.FullName ?? type.Name;
|
||||
}
|
||||
}
|
||||
|
||||
@ -11,16 +11,16 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
using GFramework.Core.Rule;
|
||||
using GFramework.Core.Abstractions.Cqrs;
|
||||
using GFramework.Core.Abstractions.Cqrs.Query;
|
||||
using GFramework.Core.Rule;
|
||||
|
||||
namespace GFramework.Core.Cqrs.Query;
|
||||
|
||||
/// <summary>
|
||||
/// 抽象查询处理器基类
|
||||
/// 继承自ContextAwareBase并实现IQueryHandler接口，为具体的查询处理器提供基础功能
|
||||
/// 支持泛型查询和结果类型，实现CQRS模式中的查询处理
|
||||
/// 继承自 ContextAwareBase 并实现 IRequestHandler 接口，为具体的查询处理器提供基础功能。
|
||||
/// 框架会在每次分发前注入当前架构上下文，因此派生类可以通过 Context 访问架构级服务。
|
||||
/// </summary>
|
||||
/// <typeparam name="TQuery">查询类型，必须实现IQuery接口</typeparam>
|
||||
/// <typeparam name="TResult">查询结果类型</typeparam>
|
||||
|
||||
@ -17,7 +17,7 @@ namespace GFramework.Core.Cqrs.Query;
|
||||
|
||||
/// <summary>
|
||||
/// 表示一个基础查询类，用于处理带有输入和响应的查询模式实现。
|
||||
/// 该类继承自 Mediator.IQuery<TResponse> 接口，提供了通用的查询结构。
|
||||
/// 该类实现 IQuery<TResponse> 接口，提供了通用的查询结构。
|
||||
/// </summary>
|
||||
/// <typeparam name="TInput">查询输入数据的类型，必须实现 IQueryInput 接口</typeparam>
|
||||
/// <typeparam name="TResponse">查询执行后返回结果的类型</typeparam>
|
||||
|
||||
@ -1,109 +0,0 @@
|
||||
# CQRS 重写迁移跟踪
|
||||
|
||||
## 目标
|
||||
|
||||
围绕 `GFramework` 当前的双轨 CQRS 现状，完成一轮以“去 Mediator 外部依赖”为目标的架构迁移：
|
||||
|
||||
- 将 `Mediator` 从 GFramework 公共 API 和运行时主路径中移除
|
||||
- 基于 GFramework 自有抽象重建正式 CQRS runtime、行为管道和注册机制
|
||||
- 保留 `EventBus` 作为框架级事件系统，不与 CQRS notification 混同
|
||||
- 让 `CoreGrid-Migration` 直连本地 `GFramework`，作为真实迁移验证工程
|
||||
- 为复杂迁移建立明确恢复点与进度追踪，避免上下文过长或中断后失去状态
|
||||
|
||||
## 当前恢复点
|
||||
|
||||
- 恢复点编号：`CQRS-REWRITE-RP-002`
|
||||
- 当前阶段：`Phase 4`
|
||||
- 当前焦点：
|
||||
- 清理剩余 `Mediator` 包依赖与文档残留
|
||||
- 评估是否继续把协程扩展和测试项目中的 `Mediator.Abstractions` 完全移除
|
||||
- 规划第二阶段优化：代码生成注册、性能收敛、行为 API 命名统一
|
||||
|
||||
## 本轮计划
|
||||
|
||||
### Phase 0：工作流基础
|
||||
|
||||
- [x] 在 `local-plan/todos/` 建立本任务跟踪文档
|
||||
- [x] 在 `local-plan/traces/` 建立本任务追踪文档
|
||||
- [x] 将恢复点 / trace / subagent 协作规范写入 `AGENTS.md`
|
||||
|
||||
### Phase 1：本地验证链路
|
||||
|
||||
- [x] 确认 `CoreGrid-Migration` 当前引用形态
|
||||
- [x] 将 `CoreGrid-Migration` 从 NuGet 包切到本地 `GFramework` 工程引用
|
||||
- [x] 让 `CoreGrid-Migration` 使用本地 Source Generator 而不是外部已发布版本
|
||||
- [x] 验证本地引用链路至少能完成 restore / build
|
||||
|
||||
### Phase 2：CQRS 基础重建
|
||||
|
||||
- [x] 在 `GFramework.Core.Abstractions` 定义自有 CQRS 契约
|
||||
- [x] 在 `GFramework.Core` 落地 dispatcher / handler registry / behavior pipeline
|
||||
- [x] 清理 `IArchitectureContext` 中对 `Mediator.*` 的公共签名依赖
|
||||
- [x] 设计 CQRS 模块启用方式，替代 `Configurator => AddMediator(...)`
|
||||
|
||||
### Phase 3：接入迁移
|
||||
|
||||
- [x] 迁移 `GFramework.Core.Cqrs.*` 基类到新契约
|
||||
- [x] 迁移 `ContextAwareMediator*Extensions` 与协程扩展
|
||||
- [x] 迁移 `CoreGrid-Migration/scripts/cqrs/**` 到新契约
|
||||
- [x] 删除 `GameArchitecture.Configurator` 中的 `AddMediator(...)`
|
||||
|
||||
### Phase 4：收尾
|
||||
|
||||
- [ ] 移除 `Mediator` 包依赖与相关测试/文档残留
|
||||
- [x] 运行目标构建与测试
|
||||
- [x] 记录剩余风险与下一恢复点
|
||||
|
||||
## 当前完成结果
|
||||
|
||||
- `CoreGrid-Migration` 已直连本地 `GFramework` 源码与本地 source generators。
|
||||
- `GameArchitecture` 已不再依赖 `collection.AddMediator(...)` 即可使用 CQRS。
|
||||
- `GFramework.Core.Abstractions` 新增自有 CQRS 契约：
|
||||
- `IRequest<TResponse>` / `INotification` / `IStreamRequest<TResponse>`
|
||||
- `IRequestHandler<,>` / `INotificationHandler<>` / `IStreamRequestHandler<,>`
|
||||
- `Unit`
|
||||
- `IPipelineBehavior<,>` / `MessageHandlerDelegate<,>`
|
||||
- `ArchitectureBootstrapper` 会在初始化阶段自动扫描并注册当前架构程序集与 `GFramework.Core` 程序集中的 CQRS handlers。
|
||||
- `CqrsDispatcher` 已支持：
|
||||
- request dispatch
|
||||
- notification publish
|
||||
- stream dispatch
|
||||
- context-aware handler 注入
|
||||
- request pipeline behavior 链式执行
|
||||
- `GFramework.Core.Tests` 中原依赖 `Mediator` 注册路径的测试已切换到框架内建 CQRS 注册路径。
|
||||
- 当前验证状态：
|
||||
- `dotnet build GFramework/GFramework.sln` 通过
|
||||
- `dotnet test GFramework/GFramework.Core.Tests/GFramework.Core.Tests.csproj --no-build` 通过，`1621` 个测试全部通过
|
||||
- `dotnet build CoreGrid-Migration/CoreGrid.sln` 通过
|
||||
|
||||
## 当前已知事实
|
||||
|
||||
- `GFramework` 当前仍同时维护：
|
||||
- 基于 `CommandExecutor` / `QueryExecutor` / `EventBus` 的轻量旧 CQRS
|
||||
- 基于 GFramework 自有抽象的新 CQRS runtime
|
||||
- 仍存在 `Mediator` 残留的区域主要集中在：
|
||||
- 文档中的历史说明
|
||||
- `MediatorCoroutineExtensions` 及对应测试
|
||||
- 测试项目对 `Mediator.Abstractions` 的少量残余依赖
|
||||
- `CoreGrid-Migration` 已切到本地源码引用，并在当前恢复点完成构建验证
|
||||
|
||||
## 当前风险
|
||||
|
||||
- `GFramework` 仓库存在与本任务无关的既有改动，提交时必须避免覆盖
|
||||
- `CoreGrid-Migration` 是 worktree，WSL 下原生 `git` 解析该 worktree 路径有兼容问题
|
||||
- 当前 `RegisterMediatorBehavior` 命名仍保留历史前缀，但底层已切换为框架自有 CQRS pipeline；若后续要彻底脱媒介命名，需要一次 API 命名迁移
|
||||
- 当前 handler 自动注册基于运行时反射扫描；若后续追求冷启动与 AOT 友好性，需要补 source-generator 注册路径
|
||||
|
||||
## 下次恢复建议
|
||||
|
||||
若本轮中断，优先从以下顺序恢复：
|
||||
|
||||
1. 查看 `local-plan/traces/cqrs-rewrite-migration-trace.md`
|
||||
2. 确认当前恢复点 `CQRS-REWRITE-RP-002` 已对应到最新提交
|
||||
3. 优先决定是否继续移除 `Mediator.Abstractions` 包与 `MediatorCoroutineExtensions` 历史兼容层
|
||||
4. 若继续演进，再处理 CQRS 注册的生成器化与 API 命名统一
|
||||
|
||||
## 备注
|
||||
|
||||
- 本文档是当前任务的主恢复点，后续每个关键阶段完成后都要更新
|
||||
- 发生方向调整时，不覆盖旧结论，直接追加阶段记录与新的恢复点编号
|
||||
@ -1,68 +0,0 @@
|
||||
# CQRS 重写迁移追踪
|
||||
|
||||
## 2026-04-14
|
||||
|
||||
### 阶段：初始化
|
||||
|
||||
- 建立 `CQRS-REWRITE-RP-001` 恢复点
|
||||
- 已确认本次迁移目标：
|
||||
- 彻底参考 `Mediator` 思路重写 GFramework 正式 CQRS
|
||||
- 不保留对 `Mediator` 的兼容层
|
||||
- 使用 `abstractions + runtime 可选模块` 边界
|
||||
- 保留 `EventBus`，不与 CQRS notification 合并
|
||||
|
||||
### 已确认的实现前提
|
||||
|
||||
- `CoreGrid-Migration` 当前仍依赖 NuGet 版 `GeWuYou.GFramework*`
|
||||
- `CoreGrid/scripts/core/GameArchitecture.cs` 与 `CoreGrid-Migration/scripts/core/GameArchitecture.cs` 通过 `AddMediator(...)` 启用基于生成器的 runtime
|
||||
- `GFramework` 当前 `IArchitectureContext` 与一批 CQRS 基类直接引用 `Mediator.*`
|
||||
- `CoreGrid/scripts/cqrs/**` 的 handler 很薄，主要迁移成本在框架 runtime 和注册机制，不在业务逻辑本身
|
||||
|
||||
### 当前动作
|
||||
|
||||
- 准备更新 `AGENTS.md`，补充恢复点 / trace / subagent 协作规范
|
||||
- 准备将 `CoreGrid-Migration` 切换为本地项目引用，建立真实验证链路
|
||||
|
||||
### 下一步
|
||||
|
||||
1. 完成 `AGENTS.md` 规则补充
|
||||
2. 改造 `CoreGrid-Migration/CoreGrid.csproj` 为本地项目与本地生成器引用
|
||||
3. 进行第一次构建验证，确认本地链路可用
|
||||
|
||||
### 阶段：CQRS 主路径迁移完成
|
||||
|
||||
- `CoreGrid-Migration/CoreGrid.csproj` 已切到本地 `ProjectReference` + 本地 source generators
|
||||
- `CoreGrid-Migration/scripts/core/GameArchitecture.cs` 已删除 `AddMediator(...)` 配置钩子
|
||||
- `GFramework.Core.Abstractions` 新增 GFramework 自有 CQRS 契约与 `Unit`
|
||||
- `IArchitectureContext` / `ArchitectureContext` 已切到自有 CQRS 签名
|
||||
- `ArchitectureBootstrapper` 已内建 handler 扫描注册，使用方无需再显式调用 `AddMediator(...)`
|
||||
- `CqrsDispatcher` 已补齐 request/notification/stream dispatch 与 pipeline behavior 执行
|
||||
- `GFramework.Core.Cqrs.*` 基类、`ContextAwareMediator*Extensions`、Godot 协程上下文扩展均已迁到新契约
|
||||
- `GFramework.Core.Tests` 中原依赖旧 `Mediator` 注册入口的测试已迁移到 `CqrsTestRuntime` 反射注册路径
|
||||
|
||||
### 阶段：验证
|
||||
|
||||
- `dotnet build /mnt/f/gewuyou/System/Documents/WorkSpace/GameDev/GFramework/GFramework.Core/GFramework.Core.csproj`
|
||||
- 结果：通过
|
||||
- `dotnet build /mnt/f/gewuyou/System/Documents/WorkSpace/GameDev/GFramework/GFramework.Core.Tests/GFramework.Core.Tests.csproj`
|
||||
- 结果：通过
|
||||
- `dotnet test /mnt/f/gewuyou/System/Documents/WorkSpace/GameDev/GFramework/GFramework.Core.Tests/GFramework.Core.Tests.csproj --no-build`
|
||||
- 结果：通过
|
||||
- 明细：`1621` 个测试全部通过
|
||||
- `dotnet build /mnt/f/gewuyou/System/Documents/WorkSpace/GameDev/GFramework/GFramework.sln`
|
||||
- 结果：通过
|
||||
- `dotnet build /mnt/f/gewuyou/System/Documents/WorkSpace/GameDev/CoreGrid-Migration/CoreGrid.sln`
|
||||
- 结果：通过
|
||||
- 备注：仅存在既有 analyzer warnings，无新增构建错误
|
||||
|
||||
### 当前残留
|
||||
|
||||
- 文档与少量历史 API 命名仍保留 `Mediator` 前缀
|
||||
- `MediatorCoroutineExtensions` 与少量测试仍依赖 `Mediator.Abstractions`
|
||||
- handler 自动注册当前使用运行时反射扫描，尚未切回生成器注册
|
||||
|
||||
### 下一步建议
|
||||
|
||||
1. 决定是否继续做“完全移除 `Mediator.Abstractions` 包”的第二阶段清理
|
||||
2. 若继续，优先迁移协程扩展与相关测试
|
||||
3. 评估是否将 `RegisterMediatorBehavior`、`ContextAwareMediator*Extensions` 等历史命名升级为 CQRS 中性命名
|
||||
Loading…
x
Reference in New Issue
Block a user