diff --git a/aspnet-core/Directory.Build.targets b/aspnet-core/Directory.Build.targets
index a0540f2e..7b6f751f 100644
--- a/aspnet-core/Directory.Build.targets
+++ b/aspnet-core/Directory.Build.targets
@@ -46,12 +46,12 @@
-
-
-
-
-
-
+
+
+
+
+
+
diff --git a/aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/EfCoreAbpProCapTransactionApiFactory.cs b/aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/EfCoreAbpProCapTransactionApiFactory.cs
index 71482e3a..695fa474 100644
--- a/aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/EfCoreAbpProCapTransactionApiFactory.cs
+++ b/aspnet-core/frameworks/src/Lion.AbpPro.CAP.EntityFrameworkCore/Lion/AbpPro/CAP/EntityFrameworkCore/EfCoreAbpProCapTransactionApiFactory.cs
@@ -57,7 +57,7 @@ public class EfCoreAbpProCapTransactionApiFactory : IAbpProCapTransactionApiFact
capTransaction.DbTransaction = originalApi.DbContextTransaction;
capTransaction.AutoCommit = false;
- Publisher.Transaction.Value = capTransaction;
+ Publisher.Transaction = capTransaction;
return (IDbContextTransaction)Activator.CreateInstance(dbProviderInfo.CapEfDbTransactionType, capTransaction);
}
diff --git a/aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/AbpProCAPMessageExtensions.cs b/aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/AbpProCAPMessageExtensions.cs
new file mode 100644
index 00000000..fc407abb
--- /dev/null
+++ b/aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/AbpProCAPMessageExtensions.cs
@@ -0,0 +1,35 @@
+namespace Lion.AbpPro.CAP;
+
+public static class AbpProCAPMessageExtensions
+{
+ ///
+ /// 尝试获取消息标头中的租户标识
+ ///
+ public static bool TryGetTenantId(this Message message, out Guid? tenantId)
+ {
+ if (message.Headers.TryGetValue(AbpProCapConsts.Tenant, out string tenantStr))
+ {
+ if (Guid.TryParse(tenantStr, out Guid id))
+ {
+ tenantId = id;
+ return true;
+ }
+ }
+
+ tenantId = null;
+ return false;
+ }
+
+ ///
+ /// 获取消息标头中的租户标识
+ ///
+ public static Guid? GetTenantIdOrNull(this Message message)
+ {
+ if (message.TryGetTenantId(out Guid? tenantId))
+ {
+ return tenantId;
+ }
+
+ return null;
+ }
+}
\ No newline at end of file
diff --git a/aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/AbpProCAPSubscribeInvoker.cs b/aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/AbpProCAPSubscribeInvoker.cs
new file mode 100644
index 00000000..c05487fa
--- /dev/null
+++ b/aspnet-core/frameworks/src/Lion.AbpPro.CAP/Lion/AbpPro/CAP/AbpProCAPSubscribeInvoker.cs
@@ -0,0 +1,165 @@
+using System.ComponentModel;
+using DotNetCore.CAP.Filter;
+using Lion.AbpPro.CAP.Internal;
+using Microsoft.Extensions.Internal;
+
+namespace Lion.AbpPro.CAP;
+
+public class AbpProCAPSubscribeInvoker
+{
+ private readonly ConcurrentDictionary _executors;
+ private readonly ISerializer _serializer;
+ private readonly IServiceProvider _serviceProvider;
+ private readonly ICurrentTenant _currentTenant;
+ public AbpProCAPSubscribeInvoker(IServiceProvider serviceProvider, ISerializer serializer, ICurrentTenant currentTenant)
+ {
+ _serviceProvider = serviceProvider;
+ _serializer = serializer;
+ _currentTenant = currentTenant;
+ _executors = new ConcurrentDictionary();
+ }
+
+ public async Task InvokeAsync(ConsumerContext context,
+ CancellationToken cancellationToken = default)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+
+ var methodInfo = context.ConsumerDescriptor.MethodInfo;
+ var reflectedTypeHandle = methodInfo.ReflectedType!.TypeHandle.Value;
+ var methodHandle = methodInfo.MethodHandle.Value;
+ var key = $"{reflectedTypeHandle}_{methodHandle}";
+
+ var executor = _executors.GetOrAdd(key,
+ _ => ObjectMethodExecutor.Create(methodInfo, context.ConsumerDescriptor.ImplTypeInfo));
+
+ await using var scope = _serviceProvider.CreateAsyncScope();
+
+ var provider = scope.ServiceProvider;
+
+ var obj = GetInstance(provider, context);
+
+ var message = context.DeliverMessage;
+ // 租户数据可能在消息标头中
+ var tenantId = message.GetTenantIdOrNull();
+ var parameterDescriptors = context.ConsumerDescriptor.Parameters;
+ var executeParameters = new object?[parameterDescriptors.Count];
+ for (var i = 0; i < parameterDescriptors.Count; i++)
+ {
+ var parameterDescriptor = parameterDescriptors[i];
+ if (parameterDescriptor.IsFromCap)
+ {
+ executeParameters[i] = GetCapProvidedParameter(parameterDescriptor, message, cancellationToken);
+ }
+ else
+ {
+ if (message.Value != null)
+ {
+ // use ISerializer when reading from storage, skip other objects if not Json
+ if (_serializer.IsJsonType(message.Value))
+ {
+ executeParameters[i] =
+ _serializer.Deserialize(message.Value, parameterDescriptor.ParameterType);
+ }
+ else
+ {
+ var converter = TypeDescriptor.GetConverter(parameterDescriptor.ParameterType);
+ if (converter.CanConvertFrom(message.Value.GetType()))
+ {
+ executeParameters[i] = converter.ConvertFrom(message.Value);
+ }
+ else
+ {
+ if (parameterDescriptor.ParameterType.IsInstanceOfType(message.Value))
+ executeParameters[i] = message.Value;
+ else
+ executeParameters[i] =
+ Convert.ChangeType(message.Value, parameterDescriptor.ParameterType);
+ }
+ }
+ }
+ }
+ }
+
+ var filter = provider.GetService();
+ object? resultObj = null;
+ try
+ {
+ if (filter != null)
+ {
+ var etContext = new ExecutingContext(context, executeParameters);
+ await filter.OnSubscribeExecutingAsync(etContext).ConfigureAwait(false);
+ executeParameters = etContext.Arguments;
+ }
+
+ using (_currentTenant.Change(tenantId))
+ {
+ resultObj = await ExecuteWithParameterAsync(executor, obj, executeParameters).ConfigureAwait(false);
+ }
+
+ if (filter != null)
+ {
+ var edContext = new ExecutedContext(context, resultObj);
+ await filter.OnSubscribeExecutedAsync(edContext).ConfigureAwait(false);
+ resultObj = edContext.Result;
+ }
+ }
+ catch (Exception e)
+ {
+ if (filter != null)
+ {
+ var exContext = new ExceptionContext(context, e);
+ await filter.OnSubscribeExceptionAsync(exContext).ConfigureAwait(false);
+ if (!exContext.ExceptionHandled) exContext.Exception.ReThrow();
+
+ if (exContext.Result != null) resultObj = exContext.Result;
+ }
+ else
+ {
+ throw;
+ }
+ }
+
+ var callbackName = message.GetCallbackName();
+ if (string.IsNullOrEmpty(callbackName))
+ {
+ return new ConsumerExecutedResult(resultObj, message.GetId(), null, null);
+ }
+ else
+ {
+ var capHeader = executeParameters.FirstOrDefault(x => x is CapHeader) as AbpProCapHeader;
+ return new ConsumerExecutedResult(resultObj, message.GetId(), callbackName, capHeader?.ResponseHeader);
+ }
+ }
+
+ private static object GetCapProvidedParameter(ParameterDescriptor parameterDescriptor, Message message,
+ CancellationToken cancellationToken)
+ {
+ if (typeof(CancellationToken).IsAssignableFrom(parameterDescriptor.ParameterType)) return cancellationToken;
+
+ if (parameterDescriptor.ParameterType.IsAssignableFrom(typeof(CapHeader)))
+ return new CapHeader(message.Headers);
+
+ throw new ArgumentException(parameterDescriptor.Name);
+ }
+
+ protected virtual object GetInstance(IServiceProvider provider, ConsumerContext context)
+ {
+ var srvType = context.ConsumerDescriptor.ServiceTypeInfo?.AsType();
+ var implType = context.ConsumerDescriptor.ImplTypeInfo.AsType();
+
+ object? obj = null;
+ if (srvType != null) obj = provider.GetServices(srvType).FirstOrDefault(o => o?.GetType() == implType);
+
+ if (obj == null) obj = ActivatorUtilities.GetServiceOrCreateInstance(provider, implType);
+
+ return obj;
+ }
+
+ private async Task