diff --git a/GFramework.Cqrs.Benchmarks/Messaging/NotificationFanOutBenchmarks.cs b/GFramework.Cqrs.Benchmarks/Messaging/NotificationFanOutBenchmarks.cs
index 0e3efb1a..29ee33ea 100644
--- a/GFramework.Cqrs.Benchmarks/Messaging/NotificationFanOutBenchmarks.cs
+++ b/GFramework.Cqrs.Benchmarks/Messaging/NotificationFanOutBenchmarks.cs
@@ -22,8 +22,8 @@
namespace GFramework.Cqrs.Benchmarks.Messaging;
///
-/// 对比固定 4 个处理器的 notification fan-out publish 在 baseline、GFramework.CQRS、NuGet `Mediator`
-/// 与 MediatR 之间的开销。
+/// 对比固定 4 个处理器的 notification fan-out publish 在 baseline、GFramework.CQRS 默认顺序发布器、
+/// GFramework.CQRS 内置 TaskWhenAllNotificationPublisher、NuGet `Mediator` 与 MediatR 之间的开销。
///
[Config(typeof(Config))]
public class NotificationFanOutBenchmarks
diff --git a/GFramework.Cqrs.Benchmarks/Messaging/RequestBenchmarks.cs b/GFramework.Cqrs.Benchmarks/Messaging/RequestBenchmarks.cs
index ff5ca2c4..f0dbf4b9 100644
--- a/GFramework.Cqrs.Benchmarks/Messaging/RequestBenchmarks.cs
+++ b/GFramework.Cqrs.Benchmarks/Messaging/RequestBenchmarks.cs
@@ -132,7 +132,7 @@ public Task SendRequest_MediatR()
}
///
- /// 通过 `ai-libs/Mediator` 的 source-generated concrete mediator 发送 request，作为高性能对照组。
+ /// 通过 NuGet `Mediator` 的 source-generated concrete mediator 发送 request，作为高性能对照组。
///
/// 代表当前 `Mediator` request dispatch 完成的值任务。
[Benchmark]
diff --git a/GFramework.Cqrs.Benchmarks/Messaging/RequestLifetimeBenchmarks.cs b/GFramework.Cqrs.Benchmarks/Messaging/RequestLifetimeBenchmarks.cs
index f12f83e6..6869b0ab 100644
--- a/GFramework.Cqrs.Benchmarks/Messaging/RequestLifetimeBenchmarks.cs
+++ b/GFramework.Cqrs.Benchmarks/Messaging/RequestLifetimeBenchmarks.cs
@@ -201,7 +201,7 @@ public Task SendRequest_MediatR()
}
///
- /// 按生命周期把 benchmark request handler 注册到 GFramework 容器。
+ /// 按生命周期把 benchmark request handler 注册到 GFramework 容器。
///
/// 当前 benchmark 拥有并负责释放的容器。
/// 待比较的 handler 生命周期。
@@ -248,7 +248,7 @@ private static ServiceLifetime ResolveMediatRLifetime(HandlerLifetime lifetime)
}
///
- /// Benchmark request。
+ /// Benchmark request。
///
/// 请求标识。
public sealed record BenchmarkRequest(Guid Id) :
diff --git a/GFramework.Cqrs.Benchmarks/Messaging/StreamingBenchmarks.cs b/GFramework.Cqrs.Benchmarks/Messaging/StreamingBenchmarks.cs
index 449c1bba..a4dff680 100644
--- a/GFramework.Cqrs.Benchmarks/Messaging/StreamingBenchmarks.cs
+++ b/GFramework.Cqrs.Benchmarks/Messaging/StreamingBenchmarks.cs
@@ -168,7 +168,7 @@ public ValueTask Stream_MediatR()
}
///
- /// 通过 `ai-libs/Mediator` 的 source-generated concrete mediator 创建 stream，并按当前观测模式消费。
+ /// 通过 NuGet `Mediator` 的 source-generated concrete mediator 创建 stream，并按当前观测模式消费。
///
/// 按当前观测模式完成 stream 消费后的等待句柄。
[Benchmark]
diff --git a/GFramework.Cqrs.Tests/Cqrs/CqrsDispatcherContextValidationTests.cs b/GFramework.Cqrs.Tests/Cqrs/CqrsDispatcherContextValidationTests.cs
index e3e0d337..c189eb53 100644
--- a/GFramework.Cqrs.Tests/Cqrs/CqrsDispatcherContextValidationTests.cs
+++ b/GFramework.Cqrs.Tests/Cqrs/CqrsDispatcherContextValidationTests.cs
@@ -147,6 +147,29 @@ public void CreateStream_Should_Throw_When_Context_Does_Not_Implement_IArchitect
Throws.InvalidOperationException.With.Message.Contains("does not implement IArchitectureContext"));
}
+ ///
+ /// 验证 stream handler 缺失时，dispatcher 会在建流调用点同步抛出异常，
+ /// 而不是返回一个延迟到枚举阶段才失败的异步流。
+ ///
+ [Test]
+ public void CreateStream_Should_Throw_When_Handler_Is_Missing()
+ {
+ var runtime = CreateRuntime(
+ container =>
+ {
+ container
+ .Setup(currentContainer => currentContainer.Get(typeof(IStreamRequestHandler)))
+ .Returns((object?)null);
+ container
+ .Setup(currentContainer => currentContainer.HasRegistration(typeof(IStreamPipelineBehavior)))
+ .Returns(false);
+ });
+
+ Assert.That(
+ () => runtime.CreateStream(new FakeCqrsContext(), new ContextAwareStreamRequest()),
+ Throws.InvalidOperationException.With.Message.Contains("No CQRS stream handler registered"));
+ }
+
///
/// 验证当 stream pipeline behavior 需要上下文注入、但当前 CQRS 上下文不实现
/// 时，
diff --git a/GFramework.Cqrs.Tests/Cqrs/CqrsGeneratedRequestInvokerProviderTests.cs b/GFramework.Cqrs.Tests/Cqrs/CqrsGeneratedRequestInvokerProviderTests.cs
index 227f101d..6c16b20e 100644
--- a/GFramework.Cqrs.Tests/Cqrs/CqrsGeneratedRequestInvokerProviderTests.cs
+++ b/GFramework.Cqrs.Tests/Cqrs/CqrsGeneratedRequestInvokerProviderTests.cs
@@ -2,6 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
using System.Reflection;
+using System.Threading;
using GFramework.Core.Abstractions.Logging;
using GFramework.Core.Architectures;
using GFramework.Core.Ioc;
@@ -28,6 +29,7 @@ public void SetUp()
{
_previousLoggerFactoryProvider = LoggerFactoryResolver.Provider;
LoggerFactoryResolver.Provider = new ConsoleLoggerFactoryProvider();
+ GeneratedRequestPipelineTrackingBehavior.InvocationCount = 0;
GeneratedStreamPipelineTrackingBehavior.InvocationCount = 0;
ClearRegistrarCaches();
ClearDispatcherCaches();
@@ -40,6 +42,7 @@ public void SetUp()
public void TearDown()
{
LoggerFactoryResolver.Provider = _previousLoggerFactoryProvider ?? new ConsoleLoggerFactoryProvider();
+ GeneratedRequestPipelineTrackingBehavior.InvocationCount = 0;
GeneratedStreamPipelineTrackingBehavior.InvocationCount = 0;
ClearRegistrarCaches();
ClearDispatcherCaches();
@@ -181,6 +184,30 @@ public async Task SendAsync_Should_Use_Generated_Request_Invoker_For_Hidden_Impl
Assert.That(response, Is.EqualTo("generated-hidden:payload"));
}
+ ///
+ /// 验证 generated request invoker 与 request pipeline 行为同时存在时，
+ /// dispatcher 仍会保持 generated invoker 优先，并正确复用现有 request 执行链。
+ ///
+ [Test]
+ public async Task SendAsync_Should_Use_Generated_Request_Invoker_Inside_Request_Pipeline()
+ {
+ var generatedAssembly = CreateGeneratedRequestInvokerAssembly();
+ var container = new MicrosoftDiContainer();
+ container.RegisterCqrsPipelineBehavior();
+
+ CqrsTestRuntime.RegisterHandlers(container, generatedAssembly.Object);
+ container.Freeze();
+
+ var context = new ArchitectureContext(container);
+ var response = await context.SendRequestAsync(new GeneratedRequestInvokerRequest("payload")).ConfigureAwait(false);
+
+ Assert.Multiple(() =>
+ {
+ Assert.That(response, Is.EqualTo("generated:payload"));
+ Assert.That(GeneratedRequestPipelineTrackingBehavior.InvocationCount, Is.EqualTo(1));
+ });
+ }
+
///
/// 验证 dispatcher 在首次创建 stream binding 时，会优先消费 generated stream invoker provider。
///
@@ -608,6 +635,40 @@ public async Task CreateStream_Should_Use_Later_Valid_Generated_Stream_Descripto
Assert.That(results, Is.EqualTo([30, 31]));
}
+ ///
+ /// 记录 generated request invoker 与 request pipeline 行为组合时的命中次数。
+ ///
+ private sealed class GeneratedRequestPipelineTrackingBehavior
+ : IPipelineBehavior
+ {
+ private static int _invocationCount;
+
+ ///
+ /// 获取或重置当前测试进程中的行为触发次数。
+ ///
+ public static int InvocationCount
+ {
+ get => Volatile.Read(ref _invocationCount);
+ set => Volatile.Write(ref _invocationCount, value);
+ }
+
+ ///
+ /// 记录一次行为执行，然后继续执行 generated request invoker。
+ ///
+ /// 当前请求消息。
+ /// 下一个处理阶段。
+ /// 取消令牌。
+ /// 下游处理阶段返回的响应。
+ public ValueTask Handle(
+ GeneratedRequestInvokerRequest message,
+ MessageHandlerDelegate next,
+ CancellationToken cancellationToken)
+ {
+ Interlocked.Increment(ref _invocationCount);
+ return next(message, cancellationToken);
+ }
+ }
+
///
/// 模拟返回实例 request invoker 方法的 generated registry。
///
diff --git a/GFramework.Cqrs/Internal/CqrsDispatcher.cs b/GFramework.Cqrs/Internal/CqrsDispatcher.cs
index b9bab1db..3e039e3e 100644
--- a/GFramework.Cqrs/Internal/CqrsDispatcher.cs
+++ b/GFramework.Cqrs/Internal/CqrsDispatcher.cs
@@ -150,7 +150,7 @@ public ValueTask SendAsync(
}
return dispatchBinding.GetPipelineExecutor(behaviors.Count)
- .Invoke(handler, behaviors, request, cancellationToken);
+ .Invoke(handler, behaviors, dispatchBinding.RequestInvoker, request, cancellationToken);
}
catch (Exception exception)
{
@@ -604,12 +604,14 @@ private static ValueTask InvokeRequestHandlerAsync InvokeRequestPipelineExecutorAsync(
object handler,
IReadOnlyList
private sealed class RequestPipelineInvocation(
- IRequestHandler handler,
+ object handler,
+ RequestInvoker requestInvoker,
IReadOnlyList