From 2cccde7d1b0497356f534d8702b31e12d3308a4d Mon Sep 17 00:00:00 2001
From: Mahmoud <mahmoudsabzalii@gmail.com>
Date: Tue, 1 Jul 2025 11:56:26 +0330
Subject: [PATCH] feat: support multiple SubscribeFilter

---
 src/DotNetCore.CAP/CAP.Builder.cs             |  2 +-
 .../Internal/ISubscribeInvoker.Default.cs     | 38 +++++++++++--------
 2 files changed, 23 insertions(+), 17 deletions(-)

diff --git a/src/DotNetCore.CAP/CAP.Builder.cs b/src/DotNetCore.CAP/CAP.Builder.cs
index 31e6f9777..bdcb8954a 100644
--- a/src/DotNetCore.CAP/CAP.Builder.cs
+++ b/src/DotNetCore.CAP/CAP.Builder.cs
@@ -101,7 +101,7 @@ public CapBuilder(IServiceCollection services)
     /// <typeparam name="T">Type of filter</typeparam>
     public CapBuilder AddSubscribeFilter<T>() where T : class, ISubscribeFilter
     {
-        Services.TryAddScoped<ISubscribeFilter, T>();
+        Services.AddScoped<ISubscribeFilter, T>();
         return this;
     }
 
diff --git a/src/DotNetCore.CAP/Internal/ISubscribeInvoker.Default.cs b/src/DotNetCore.CAP/Internal/ISubscribeInvoker.Default.cs
index 81ec004f6..313f1fc47 100644
--- a/src/DotNetCore.CAP/Internal/ISubscribeInvoker.Default.cs
+++ b/src/DotNetCore.CAP/Internal/ISubscribeInvoker.Default.cs
@@ -3,6 +3,7 @@
 
 using System;
 using System.Collections.Concurrent;
+using System.Collections.Generic;
 using System.ComponentModel;
 using System.Linq;
 using System.Threading;
@@ -87,40 +88,45 @@ public async Task<ConsumerExecutedResult> InvokeAsync(ConsumerContext context,
             }
         }
 
-        var filter = provider.GetService<ISubscribeFilter>();
+        var filters = provider.GetServices<ISubscribeFilter>().ToList();
+        Stack<ISubscribeFilter> executedFilters = new Stack<ISubscribeFilter>();
+
         object? resultObj = null;
         try
         {
-            if (filter != null)
+            foreach (var filter in filters)
             {
-                var etContext = new ExecutingContext(context, executeParameters);
-                await filter.OnSubscribeExecutingAsync(etContext).ConfigureAwait(false);
-                executeParameters = etContext.Arguments;
+                var ctx = new ExecutingContext(context, executeParameters);
+                await filter.OnSubscribeExecutingAsync(ctx).ConfigureAwait(false);
+                executeParameters = ctx.Arguments;
+                executedFilters.Push(filter);
             }
 
             resultObj = await ExecuteWithParameterAsync(executor, obj, executeParameters).ConfigureAwait(false);
 
-            if (filter != null)
+            while (executedFilters.Count > 0)
             {
-                var edContext = new ExecutedContext(context, resultObj);
-                await filter.OnSubscribeExecutedAsync(edContext).ConfigureAwait(false);
-                resultObj = edContext.Result;
+                var filter = executedFilters.Peek();
+                var ctx = new ExecutedContext(context, resultObj);
+                await filter.OnSubscribeExecutedAsync(ctx).ConfigureAwait(false);
+                resultObj = ctx.Result;
+                executedFilters.Pop();
             }
         }
-        catch (Exception e)
+        catch (Exception ex)
         {
-            if (filter != null)
+            if (executedFilters.Count == 0)
+                throw;
+            while (executedFilters.Count > 0)
             {
-                var exContext = new ExceptionContext(context, e);
+                var exContext = new ExceptionContext(context, ex);
+                var filter = executedFilters.Pop();
                 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();