Add Pipeline to received context, this mean when you got a new request to fetch data, you can add the pipeline to handle what you want!

Author: quyvu01

OfX.sln

@@ -31,4 +31,6 @@ Global
 		{36F9049C-6A77-4EF7-9CE0-86A22CA36DEA}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{36F9049C-6A77-4EF7-9CE0-86A22CA36DEA}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
+	GlobalSection(NestedProjects) = preSolution
+	EndGlobalSection
 EndGlobal

src/OfX.EntityFrameworkCore/EfQueryOfHandler.cs

@@ -6,7 +6,6 @@
 using OfX.Abstractions;
 using OfX.Attributes;
 using OfX.EntityFrameworkCore.Delegates;
-using OfX.Exceptions;
 using OfX.Responses;
 
 namespace OfX.EntityFrameworkCore;

src/OfX.Grpc/Servers/OfXGrpcServer.cs

@@ -13,7 +13,7 @@ namespace OfX.Grpc.Servers;
 
 public sealed class OfXGrpcServer(IServiceProvider serviceProvider) : OfXTransportService.OfXTransportServiceBase
 {
-    private const string GetDataAsync = nameof(GetDataAsync);
+    private const string ExecuteAsync = nameof(ExecuteAsync);
 
     private static readonly Lazy<ConcurrentDictionary<Type, MethodInfo>> MethodInfoStorage =
         new(() => new ConcurrentDictionary<Type, MethodInfo>());
@@ -30,12 +30,16 @@ public override async Task<OfXItemsGrpcResponse> GetItems(GetOfXGrpcQuery reques
             if (!OfXCached.QueryMapHandler.TryGetValue(attributeType, out var handlerType))
                 throw new OfXGrpcExceptions.CannotFindHandlerForOfAttribute(attributeType);
 
-            var handler = serviceProvider.GetRequiredService(handlerType);
+            var modelArg = handlerType.GetGenericArguments()[0];
 
-            var genericMethod = MethodInfoStorage.Value.GetOrAdd(attributeType, q => handler.GetType().GetMethods()
+            var pipeline = serviceProvider
+                .GetRequiredService(typeof(ReceivedPipelinesImpl<,>).MakeGenericType(modelArg, attributeType));
+
+            var pipelineMethod = MethodInfoStorage.Value.GetOrAdd(attributeType, q => pipeline.GetType().GetMethods()
                 .FirstOrDefault(m =>
-                    m.Name == GetDataAsync && m.GetParameters() is { Length: 1 } parameters &&
+                    m.Name == ExecuteAsync && m.GetParameters() is { Length: 1 } parameters &&
                     parameters[0].ParameterType == typeof(RequestContext<>).MakeGenericType(q)));
+
             var requestContextType = typeof(RequestContextImpl<>).MakeGenericType(attributeType);
 
             var queryType = typeof(RequestOf<>).MakeGenericType(attributeType);
@@ -47,9 +51,8 @@ public override async Task<OfXItemsGrpcResponse> GetItems(GetOfXGrpcQuery reques
                 .CreateInstance(requestContextType, query, headers, context.CancellationToken);
             object[] arguments = [requestContext];
             // Invoke the method and get the result
-            var requestTask = ((Task<ItemsResponse<OfXDataResponse>>)genericMethod
-                .Invoke(handler, arguments))!;
-            var response = await requestTask;
+            var response = await ((Task<ItemsResponse<OfXDataResponse>>)pipelineMethod!
+                .Invoke(pipeline, arguments))!;
             var res = new OfXItemsGrpcResponse();
             response.Items.ForEach(a => res.Items.Add(new ItemGrpc { Id = a.Id, Value = a.Value }));
             return res;

src/OfX.Tests/OfXCoreTests.cs

@@ -7,6 +7,7 @@
 using OfX.Tests.Attributes;
 using OfX.Tests.Contexts;
 using OfX.Tests.Models;
+using OfX.Tests.Pipelines;
 using Xunit;
 
 namespace OfX.Tests;
@@ -26,9 +27,8 @@ public OfXCoreTests()
                         options.AddAttributesContainNamespaces(assembly);
                         options.AddHandlersFromNamespaceContaining<ITestAssemblyMarker>();
                         options.AddGrpcClients(c =>
-                        {
-                            c.AddGrpcHostWithOfXAttributes("localhost:5001", [typeof(UserOfAttribute)]);
-                        });
+                            c.AddGrpcHostWithOfXAttributes("localhost:5001", [typeof(UserOfAttribute)]));
+                        options.AddReceivedPipelines(c => c.OfType(typeof(TestReceivedPipelinesImpl<>)));
                     })
                     .AddOfXEFCore(options =>
                     {

src/OfX.Tests/Pipelines/TestReceivedPipelinesImpl.cs

@@ -0,0 +1,17 @@
+using OfX.Abstractions;
+using OfX.Attributes;
+using OfX.Responses;
+
+namespace OfX.Tests.Pipelines;
+
+public sealed class TestReceivedPipelinesImpl<TAttribute> : IReceivedPipelineBehavior<TAttribute>
+    where TAttribute : OfXAttribute
+{
+    public async Task<ItemsResponse<OfXDataResponse>> HandleAsync(RequestContext<TAttribute> requestContext,
+        Func<Task<ItemsResponse<OfXDataResponse>>> next)
+    {
+        var result = await next.Invoke();
+        Console.WriteLine("Here is TestReceivedPipelinesImpl<>");
+        return result;
+    }
+}

src/OfX/Abstractions/IReceivedPipelineBehavior.cs

@@ -0,0 +1,10 @@
+using OfX.Attributes;
+using OfX.Responses;
+
+namespace OfX.Abstractions;
+
+public interface IReceivedPipelineBehavior<TTAttribute> where TTAttribute : OfXAttribute
+{
+    Task<ItemsResponse<OfXDataResponse>> HandleAsync(RequestContext<TTAttribute> requestContext,
+        Func<Task<ItemsResponse<OfXDataResponse>>> next);
+}

src/OfX/ApplicationModels/ReceivedPipeline.cs

@@ -0,0 +1,36 @@
+using Microsoft.Extensions.DependencyInjection;
+using OfX.Abstractions;
+using OfX.Exceptions;
+
+namespace OfX.ApplicationModels;
+
+public sealed class ReceivedPipeline(IServiceCollection serviceCollection)
+{
+    private static readonly Type interfaceReceivedPipeline = typeof(IReceivedPipelineBehavior<>);
+
+    public ReceivedPipeline OfType<TReceivedPipeline>()
+    {
+        OfType(typeof(TReceivedPipeline));
+        return this;
+    }
+
+    // Hmmm, this one is temporary!. I think should test more case!
+    public ReceivedPipeline OfType(Type pipelineType)
+    {
+        var signatureInterfaceType = pipelineType.GetInterfaces()
+            .FirstOrDefault(a => a.IsGenericType && a.GetGenericTypeDefinition() == interfaceReceivedPipeline);
+        if (signatureInterfaceType is null)
+            throw new OfXException.PipelineIsNotReceivedPipelineBehavior(pipelineType);
+        if (pipelineType.IsGenericType)
+        {
+            var isContainsGenericParameters = pipelineType.ContainsGenericParameters;
+            if (isContainsGenericParameters)
+            {
+                serviceCollection.AddScoped(interfaceReceivedPipeline, pipelineType);
+                return this;
+            }
+        }
+        serviceCollection.AddScoped(signatureInterfaceType, pipelineType);
+        return this;
+    }
+}

src/OfX/Exceptions/OfXException.cs

@@ -10,4 +10,7 @@ public sealed class AttributesFromNamespaceShouldBeAdded() :
 
     public sealed class CurrentIdTypeWasNotSupported() :
         Exception("Current Id type was not supported. Please create a join us to contribute more!");
+    
+    public sealed class PipelineIsNotReceivedPipelineBehavior(Type type) :
+        Exception($"The input pipeline: {type.Name} is not matched with ReceivedPipelineBehavior. Please check again!");
 }

src/OfX/Extensions/OfXExtensions.cs

@@ -46,10 +46,6 @@ public static OfXRegister AddOfX(this IServiceCollection serviceCollection, Acti
                 });
         }
 
-
-        serviceCollection.AddScoped<IDataMappableService>(sp =>
-            new DataMappableService(sp, newOfRegister.AttributesRegister));
-
         var defaultImplementedInterface = typeof(DefaultMappableRequestHandler<>);
         newOfRegister.AttributesRegister.SelectMany(a => a.ExportedTypes)
             .Where(t => t is { IsClass: true, IsAbstract: false } && typeof(OfXAttribute).IsAssignableFrom(t))
@@ -63,12 +59,17 @@ public static OfXRegister AddOfX(this IServiceCollection serviceCollection, Acti
                 // So we have to replace the default service if existed -> Good!
                 serviceCollection.TryAddScoped(parentType, defaultImplementedService);
             });
+        
+        serviceCollection.AddScoped<IDataMappableService>(sp =>
+            new DataMappableService(sp, newOfRegister.AttributesRegister));
+        
+        serviceCollection.AddScoped(typeof(ReceivedPipelinesImpl<,>));
+
         return newOfRegister;
     }
 
     public static void AddExtensionHandler(this IExtensionHandlersInstaller extensionHandlersInstaller,
-        Type serviceType,
-        Type implementationType, Type attributeType)
+        Type serviceType, Type implementationType, Type attributeType)
     {
         if (!OfXCached.InternalQueryMapHandler.TryAdd(attributeType, serviceType))
             throw new OfXException.RequestMustNotBeAddMoreThanOneTimes();

src/OfX/Extensions/PipelineExtensions.cs

@@ -0,0 +1,13 @@
+using OfX.ApplicationModels;
+using OfX.Registries;
+
+namespace OfX.Extensions;
+
+public static class PipelineExtensions
+{
+    public static void AddReceivedPipelines(this OfXRegister ofXRegister, Action<ReceivedPipeline> options)
+    {
+        var receivedPipeline = new ReceivedPipeline(ofXRegister.ServiceCollection);
+        options.Invoke(receivedPipeline);
+    }
+}

src/OfX/Implementations/ReceivedPipelinesImpl.cs

@@ -0,0 +1,24 @@
+using OfX.Abstractions;
+using OfX.Attributes;
+using OfX.Responses;
+
+namespace OfX.Implementations;
+
+public class ReceivedPipelinesImpl<TModel, TAttribute>(
+    IEnumerable<IReceivedPipelineBehavior<TAttribute>> behaviors,
+    IQueryOfHandler<TModel, TAttribute> handler)
+    where TAttribute : OfXAttribute where TModel : class
+{
+    public async Task<ItemsResponse<OfXDataResponse>> ExecuteAsync(RequestContext<TAttribute> request)
+    {
+        var next = new Func<Task<ItemsResponse<OfXDataResponse>>>(() => handler.GetDataAsync(request));
+
+        foreach (var behavior in behaviors.Reverse())
+        {
+            var current = next;
+            next = () => behavior.HandleAsync(request, current);
+        }
+
+        return await next();
+    }
+}