Merge branch 'refs/heads/dev'

src/OfX.EntityFrameworkCore/OfX.EntityFrameworkCore.csproj

@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <TargetFrameworks>net9.0;net8.0</TargetFrameworks>
         <LangVersion>default</LangVersion>
-        <Version>3.2.0</Version>
+        <Version>3.2.1</Version>
         <Authors>Quy Vu</Authors>
         <PackageId>OfX-EFCore</PackageId>
         <Description>OfX extension. Use EntityFramework as Data Querying</Description>

src/OfX.Grpc/OfX.Grpc.csproj

@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <TargetFrameworks>net9.0;net8.0</TargetFrameworks>
         <LangVersion>default</LangVersion>
-        <Version>3.2.0</Version>
+        <Version>3.2.1</Version>
         <Authors>Quy Vu</Authors>
         <PackageId>OfX-gRPC</PackageId>
         <Description>OfX extension. Use gRPC as Data transporting</Description>

src/OfX.Kafka/OfX.Kafka.csproj

@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <TargetFrameworks>net9.0;net8.0</TargetFrameworks>
         <LangVersion>default</LangVersion>
-        <Version>3.2.0</Version>
+        <Version>3.2.1</Version>
         <Authors>Quy Vu</Authors>
         <PackageId>OfX-Kafka</PackageId>
         <Description>OfX-Kafka extension. Use Kafka as Data transporting</Description>

src/OfX.Nats/OfX.Nats.csproj

@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <TargetFrameworks>net9.0;net8.0</TargetFrameworks>
         <LangVersion>default</LangVersion>
-        <Version>3.2.0</Version>
+        <Version>3.2.1</Version>
         <Authors>Quy Vu</Authors>
         <PackageId>OfX-Nats</PackageId>
         <Description>Nats.io extension. Use Nats as Data transporting</Description>

src/OfX.Nats/Servers/NatsServersListening.cs

@@ -1,10 +1,10 @@
 using Microsoft.Extensions.DependencyInjection;
 using OfX.Abstractions;
+using OfX.ApplicationModels;
 using OfX.Cached;
 using OfX.Exceptions;
 using OfX.Implementations;
 using OfX.Nats.Extensions;
-using OfX.Nats.Messages;
 using OfX.Nats.Wrappers;
 using OfX.Responses;
 
@@ -22,8 +22,8 @@ internal static void StartAsync(IServiceProvider serviceProvider)
             .Select(a => a.GetGenericArguments()[1]).ToList();
         attributeTypes.ForEach(attributeType => Task.Factory.StartNew(async () =>
         {
-            var natsScribeAsync =
-                natsClient.NatsClient.SubscribeAsync<MessageRequestOf>(attributeType.GetNatsSubject());
+            var natsScribeAsync = natsClient.NatsClient
+                .SubscribeAsync<MessageDeserializable>(attributeType.GetNatsSubject());
             await foreach (var message in natsScribeAsync)
             {
                 if (message.Data is null) continue;

src/OfX.RabbitMq/OfX.RabbitMq.csproj

@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <TargetFrameworks>net9.0;net8.0</TargetFrameworks>
         <LangVersion>default</LangVersion>
-        <Version>3.2.0</Version>
+        <Version>3.2.1</Version>
         <Authors>Quy Vu</Authors>
         <PackageId>OfX-RabbitMq</PackageId>
         <Description>OfX-RabbitMq extension. Use RabbitMq as Data transporting</Description>

src/OfX/Abstractions/IReceivedPipelinesBase.cs

@@ -0,0 +1,9 @@
+using OfX.Attributes;
+using OfX.Responses;
+
+namespace OfX.Abstractions;
+
+public interface IReceivedPipelinesBase<TAttribute> where TAttribute : OfXAttribute
+{
+    Task<ItemsResponse<OfXDataResponse>> ExecuteAsync(RequestContext<TAttribute> requestContext);
+}

src/OfX/Extensions/OfXExtensions.cs

@@ -67,6 +67,8 @@ public static OfXRegister AddOfX(this IServiceCollection serviceCollection, Acti
 
         serviceCollection.AddTransient(typeof(SendPipelinesImpl<>));
 
+        serviceCollection.AddTransient(typeof(ISendPipelineBehavior<>), typeof(SendPipelineRoutingBehavior<>));
+
         return newOfRegister;
     }
 

src/OfX/Implementations/ReceivedPipelinesImpl.cs

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

src/OfX/Implementations/SendPipelineRoutingBehavior.cs

@@ -0,0 +1,26 @@
+using OfX.Abstractions;
+using OfX.Attributes;
+using OfX.Cached;
+using OfX.Responses;
+
+namespace OfX.Implementations;
+
+internal sealed class SendPipelineRoutingBehavior<TAttribute>(
+    IServiceProvider serviceProvider) :
+    ISendPipelineBehavior<TAttribute> where TAttribute : OfXAttribute
+{
+    public async Task<ItemsResponse<OfXDataResponse>> HandleAsync(RequestContext<TAttribute> requestContext,
+        Func<Task<ItemsResponse<OfXDataResponse>>> next)
+    {
+        // Check if we have the inner handler for `TAttribute` or not. If have, we will call the ReceivedPipelinesImpl<,> instead of sending via message!
+        var existedHandler = OfXCached.AttributeMapHandler;
+        if (!existedHandler.TryGetValue(typeof(TAttribute), out var handlerType)) return await next.Invoke();
+        if (!handlerType.IsGenericType) return await next.Invoke();
+        var args = handlerType.GetGenericArguments();
+        var receivedPipelineBehavior = serviceProvider
+            .GetService(typeof(ReceivedPipelinesImpl<,>).MakeGenericType(args));
+        if (receivedPipelineBehavior is not IReceivedPipelinesBase<TAttribute> receivedPipelinesBase)
+            return await next.Invoke();
+        return await receivedPipelinesBase.ExecuteAsync(requestContext);
+    }
+}

src/OfX/OfX.csproj

@@ -4,7 +4,7 @@
         <ImplicitUsings>enable</ImplicitUsings>
         <TargetFrameworks>net9.0;net8.0</TargetFrameworks>
         <LangVersion>default</LangVersion>
-        <Version>3.2.0</Version>
+        <Version>3.2.1</Version>
         <Authors>Quy Vu</Authors>
         <PackageId>OfX</PackageId>
         <Description>The high performance and easiest way to play with microservices for .NET</Description>