fix issue #661 for Advanced aggregations (#704)

* Add Advanced Aggregation Feature

* fix overwrite error

* distinct data for better performance

* remove constructor parameter

* fix tests issue

* fix tests

* fix tests issue

* Add UnitTest and AcceptanceTest

* fix responseKeys typo

* Update SimpleJsonResponseAggregator.cs

* change port

src/Ocelot/Configuration/Builder/ReRouteBuilder.cs

@@ -4,6 +4,7 @@
     using System.Net.Http;
     using Ocelot.Values;
     using System.Linq;
+    using Ocelot.Configuration.File;
 
     public class ReRouteBuilder
     {
@@ -11,11 +12,13 @@
         private List<HttpMethod> _upstreamHttpMethod;
         private string _upstreamHost;
         private List<DownstreamReRoute> _downstreamReRoutes;
+        private List<AggregateReRouteConfig> _downstreamReRoutesConfig;
         private string _aggregator;
 
         public ReRouteBuilder()
         {
             _downstreamReRoutes = new List<DownstreamReRoute>();
+            _downstreamReRoutesConfig = new List<AggregateReRouteConfig>();
         }
 
         public ReRouteBuilder WithDownstreamReRoute(DownstreamReRoute value)
@@ -48,6 +51,12 @@
             return this;
         }
 
+        public ReRouteBuilder WithAggregateReRouteConfig(List<AggregateReRouteConfig> aggregateReRouteConfigs)
+        {
+            _downstreamReRoutesConfig = aggregateReRouteConfigs;
+            return this;
+        }
+
         public ReRouteBuilder WithAggregator(string aggregator)
         {
             _aggregator = aggregator;
@@ -58,6 +67,7 @@
         {
             return new ReRoute(
                 _downstreamReRoutes,
+                _downstreamReRoutesConfig,
                 _upstreamHttpMethod, 
                 _upstreamTemplatePattern, 
                 _upstreamHost,

src/Ocelot/Configuration/Creator/AggregatesCreator.cs

@@ -24,22 +24,27 @@ namespace Ocelot.Configuration.Creator
 
         private ReRoute SetUpAggregateReRoute(IEnumerable<ReRoute> reRoutes, FileAggregateReRoute aggregateReRoute, FileGlobalConfiguration globalConfiguration)
         {
-            var applicableReRoutes = reRoutes
+            var applicableReRoutes = new List<DownstreamReRoute>();
-                .SelectMany(x => x.DownstreamReRoute)
+            var allReRoutes = reRoutes.SelectMany(x => x.DownstreamReRoute);
-                .Where(r => aggregateReRoute.ReRouteKeys.Contains(r.Key))
-                .ToList();
 
-            if (applicableReRoutes.Count != aggregateReRoute.ReRouteKeys.Count)
+            foreach (var reRouteKey in aggregateReRoute.ReRouteKeys)
+            {
+                var selec = allReRoutes.FirstOrDefault(q => q.Key == reRouteKey);
+                if (selec == null)
                 {
                     return null;
                 }
 
+                applicableReRoutes.Add(selec);
+            }
+
             var upstreamTemplatePattern = _creator.Create(aggregateReRoute);
 
             var reRoute = new ReRouteBuilder()
                 .WithUpstreamHttpMethod(aggregateReRoute.UpstreamHttpMethod)
                 .WithUpstreamPathTemplate(upstreamTemplatePattern)
                 .WithDownstreamReRoutes(applicableReRoutes)
+                .WithAggregateReRouteConfig(aggregateReRoute.ReRouteKeysConfig)
                 .WithUpstreamHost(aggregateReRoute.UpstreamHost)
                 .WithAggregator(aggregateReRoute.Aggregator)
                 .Build();

src/Ocelot/Configuration/File/AggregateReRouteConfig.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Text;
+
+namespace Ocelot.Configuration.File
+{
+    public class AggregateReRouteConfig
+    {
+        public string ReRouteKey { get; set; }
+        public string Parameter { get; set; }
+        public string JsonPath { get; set; }
+    }
+}

src/Ocelot/Configuration/File/FileAggregateReRoute.cs

@@ -5,6 +5,7 @@ namespace Ocelot.Configuration.File
     public class FileAggregateReRoute : IReRoute
     {
         public List<string> ReRouteKeys { get;set; }
+        public List<AggregateReRouteConfig> ReRouteKeysConfig { get;set; }
         public string UpstreamPathTemplate { get;set; }
         public string UpstreamHost { get; set; }
         public bool ReRouteIsCaseSensitive { get; set; }

src/Ocelot/Configuration/ReRoute.cs

@@ -2,11 +2,13 @@
 {
     using System.Collections.Generic;
     using System.Net.Http;
+    using Ocelot.Configuration.File;
     using Ocelot.Values;
 
     public class ReRoute
     {
         public ReRoute(List<DownstreamReRoute> downstreamReRoute,
+            List<AggregateReRouteConfig> downstreamReRouteConfig,
             List<HttpMethod> upstreamHttpMethod, 
             UpstreamPathTemplate upstreamTemplatePattern, 
             string upstreamHost,
@@ -14,6 +16,7 @@
         {
             UpstreamHost = upstreamHost;
             DownstreamReRoute = downstreamReRoute;
+            DownstreamReRouteConfig = downstreamReRouteConfig;
             UpstreamHttpMethod = upstreamHttpMethod;
             UpstreamTemplatePattern = upstreamTemplatePattern;
             Aggregator = aggregator;
@@ -23,6 +26,7 @@
         public List<HttpMethod> UpstreamHttpMethod { get; private set; }
         public string UpstreamHost { get; private set; }
         public List<DownstreamReRoute> DownstreamReRoute { get; private set; }
+        public List<AggregateReRouteConfig> DownstreamReRouteConfig { get; private set; }
         public string Aggregator {get; private set;}
     }
 }

src/Ocelot/Middleware/Multiplexer/InMemoryResponseAggregatorFactory.cs

@@ -5,12 +5,12 @@ namespace Ocelot.Middleware.Multiplexer
     public class InMemoryResponseAggregatorFactory : IResponseAggregatorFactory
     {
         private readonly UserDefinedResponseAggregator _userDefined;
-        private readonly SimpleJsonResponseAggregator _simple;
+        private readonly IResponseAggregator _simple;
 
-        public InMemoryResponseAggregatorFactory(IDefinedAggregatorProvider provider)
+        public InMemoryResponseAggregatorFactory(IDefinedAggregatorProvider provider, IResponseAggregator responseAggregator)
         {
             _userDefined = new UserDefinedResponseAggregator(provider);
-            _simple = new SimpleJsonResponseAggregator();
+            _simple = responseAggregator;
         }
 
         public IResponseAggregator Get(ReRoute reRoute)

src/Ocelot/Middleware/Multiplexer/Multiplexer.cs

@@ -1,7 +1,9 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
+using Newtonsoft.Json.Linq;
 using Ocelot.Configuration;
+using Ocelot.DownstreamRouteFinder.UrlMatcher;
 
 namespace Ocelot.Middleware.Multiplexer
 {
@@ -15,6 +17,9 @@ namespace Ocelot.Middleware.Multiplexer
         }
 
         public async Task Multiplex(DownstreamContext context, ReRoute reRoute, OcelotRequestDelegate next)
+        {
+            var reRouteKeysConfigs = reRoute.DownstreamReRouteConfig;
+            if (reRouteKeysConfigs == null || !reRouteKeysConfigs.Any())
             {
                 var tasks = new Task<DownstreamContext>[reRoute.DownstreamReRoute.Count];
 
@@ -42,6 +47,77 @@ namespace Ocelot.Middleware.Multiplexer
 
                 await Map(reRoute, context, contexts);
             }
+            else
+            {
+                var downstreamContextMain = new DownstreamContext(context.HttpContext)
+                {
+                    TemplatePlaceholderNameAndValues = context.TemplatePlaceholderNameAndValues,
+                    Configuration = context.Configuration,
+                    DownstreamReRoute = reRoute.DownstreamReRoute[0],
+                };
+                var mainResponse = await Fire(downstreamContextMain, next);
+
+                if (reRoute.DownstreamReRoute.Count == 1)
+                {
+                    MapNotAggregate(context, new List<DownstreamContext>() { mainResponse });
+                    return;
+                }
+
+                var tasks = new List<Task<DownstreamContext>>();
+                if (mainResponse.DownstreamResponse == null)
+                {
+                    return;
+                }
+
+                var content = await mainResponse.DownstreamResponse.Content.ReadAsStringAsync();
+                var jObject = Newtonsoft.Json.Linq.JToken.Parse(content);
+
+                for (var i = 1; i < reRoute.DownstreamReRoute.Count; i++)
+                {
+                    var templatePlaceholderNameAndValues = context.TemplatePlaceholderNameAndValues;
+                    var downstreamReRoute = reRoute.DownstreamReRoute[i];
+                    var matchAdvancedAgg = reRouteKeysConfigs.FirstOrDefault(q => q.ReRouteKey == downstreamReRoute.Key);
+                    if (matchAdvancedAgg != null)
+                    {
+                        var values = jObject.SelectTokens(matchAdvancedAgg.JsonPath).Select(s => s.ToString()).Distinct().ToList();
+
+                        foreach (var value in values)
+                        {
+                            var downstreamContext = new DownstreamContext(context.HttpContext)
+                            {
+                                TemplatePlaceholderNameAndValues = new List<PlaceholderNameAndValue>(templatePlaceholderNameAndValues),
+                                Configuration = context.Configuration,
+                                DownstreamReRoute = downstreamReRoute,
+                            };
+                            downstreamContext.TemplatePlaceholderNameAndValues.Add(new PlaceholderNameAndValue("{" + matchAdvancedAgg.Parameter + "}", value.ToString()));
+                            tasks.Add(Fire(downstreamContext, next));
+                        }
+                    }
+                    else
+                    {
+                        var downstreamContext = new DownstreamContext(context.HttpContext)
+                        {
+                            TemplatePlaceholderNameAndValues = new List<PlaceholderNameAndValue>(templatePlaceholderNameAndValues),
+                            Configuration = context.Configuration,
+                            DownstreamReRoute = downstreamReRoute,
+                        };
+                        tasks.Add(Fire(downstreamContext, next));
+                    }
+                }
+
+                await Task.WhenAll(tasks);
+
+                var contexts = new List<DownstreamContext>() { mainResponse };
+
+                foreach (var task in tasks)
+                {
+                    var finished = await task;
+                    contexts.Add(finished);
+                }
+
+                await Map(reRoute, context, contexts);
+            }
+        }
 
         private async Task Map(ReRoute reRoute, DownstreamContext context, List<DownstreamContext> contexts)
         {

src/Ocelot/Middleware/Multiplexer/SimpleJsonResponseAggregator.cs

@@ -1,4 +1,5 @@
 using System.Collections.Generic;
+using System.Linq;
 using System.Net;
 using System.Net.Http;
 using System.Net.Http.Headers;
@@ -21,19 +22,54 @@ namespace Ocelot.Middleware.Multiplexer
 
             contentBuilder.Append("{");
 
-            for (var i = 0; i < downstreamContexts.Count; i++)
+            var responseKeys = downstreamContexts.Select(s => s.DownstreamReRoute.Key).Distinct().ToList();
+
+            for (var k = 0; k < responseKeys.Count; k++)
             {
-                if (downstreamContexts[i].IsError)
+                var contexts = downstreamContexts.Where(w => w.DownstreamReRoute.Key == responseKeys[k]).ToList();
+                if (contexts.Count == 1)
                 {
-                    MapAggregateError(originalContext, downstreamContexts, i);
+                    if (contexts[0].IsError)
+                    {
+                        MapAggregateError(originalContext, contexts[0]);
                         return;
                     }
 
-                var content = await downstreamContexts[i].DownstreamResponse.Content.ReadAsStringAsync();
+                    var content = await contexts[0].DownstreamResponse.Content.ReadAsStringAsync();
+                    contentBuilder.Append($"\"{responseKeys[k]}\":{content}");
 
-                contentBuilder.Append($"\"{downstreamContexts[i].DownstreamReRoute.Key}\":{content}");
+                }
+                else
+                {
+                    contentBuilder.Append($"\"{responseKeys[k]}\":");
+                    contentBuilder.Append("[");
 
-                if (i + 1 < downstreamContexts.Count)
+                    for (var i = 0; i < contexts.Count; i++)
+                    {
+                        if (contexts[i].IsError)
+                        {
+                            MapAggregateError(originalContext, contexts[i]);
+                            return;
+                        }
+
+                        var content = await contexts[i].DownstreamResponse.Content.ReadAsStringAsync();
+                        if (string.IsNullOrWhiteSpace(content))
+                        {
+                            continue;
+                        }
+
+                        contentBuilder.Append($"{content}");
+
+                        if (i + 1 < contexts.Count)
+                        {
+                            contentBuilder.Append(",");
+                        }
+                    }
+
+                    contentBuilder.Append("]");
+                }
+
+                if (k + 1 < responseKeys.Count)
                 {
                     contentBuilder.Append(",");
                 }
@@ -43,16 +79,16 @@ namespace Ocelot.Middleware.Multiplexer
 
             var stringContent = new StringContent(contentBuilder.ToString())
             {
-                Headers = {ContentType = new MediaTypeHeaderValue("application/json")}
+                Headers = { ContentType = new MediaTypeHeaderValue("application/json") }
             };
 
             originalContext.DownstreamResponse = new DownstreamResponse(stringContent, HttpStatusCode.OK, new List<KeyValuePair<string, IEnumerable<string>>>(), "cannot return from aggregate..which reason phrase would you use?");
         }
 
-        private static void MapAggregateError(DownstreamContext originalContext, List<DownstreamContext> downstreamContexts, int i)
+        private static void MapAggregateError(DownstreamContext originalContext, DownstreamContext downstreamContext)
         {
-            originalContext.Errors.AddRange(downstreamContexts[i].Errors);
+            originalContext.Errors.AddRange(downstreamContext.Errors);
-            originalContext.DownstreamResponse = downstreamContexts[i].DownstreamResponse;
+            originalContext.DownstreamResponse = downstreamContext.DownstreamResponse;
         }
     }
 }

test/Ocelot.AcceptanceTests/AggregateTests.cs

@@ -141,6 +141,100 @@ namespace Ocelot.AcceptanceTests
                 .BDDfy();
         }
 
+        [Fact]
+        public void should_return_response_200_with_advanced_aggregate_configs()
+        {
+            var configuration = new FileConfiguration
+            {
+                ReRoutes = new List<FileReRoute>
+                    {
+                        new FileReRoute
+                        {
+                            DownstreamPathTemplate = "/",
+                            DownstreamScheme = "http",
+                            DownstreamHostAndPorts = new List<FileHostAndPort>
+                            {
+                                new FileHostAndPort
+                                {
+                                    Host = "localhost",
+                                    Port = 51889,
+                                }
+                            },
+                            UpstreamPathTemplate = "/Comments",
+                            UpstreamHttpMethod = new List<string> { "Get" },
+                            Key = "Comments"
+                        },
+                        new FileReRoute
+                        {
+                            DownstreamPathTemplate = "/users/{userId}",
+                            DownstreamScheme = "http",
+                            DownstreamHostAndPorts = new List<FileHostAndPort>
+                            {
+                                new FileHostAndPort
+                                {
+                                    Host = "localhost",
+                                    Port = 51890,
+                                }
+                            },
+                            UpstreamPathTemplate = "/UserDetails",
+                            UpstreamHttpMethod = new List<string> { "Get" },
+                            Key = "UserDetails"
+                        },
+                        new FileReRoute
+                        {
+                            DownstreamPathTemplate = "/posts/{postId}",
+                            DownstreamScheme = "http",
+                            DownstreamHostAndPorts = new List<FileHostAndPort>
+                            {
+                                new FileHostAndPort
+                                {
+                                    Host = "localhost",
+                                    Port = 51887,
+                                }
+                            },
+                            UpstreamPathTemplate = "/PostDetails",
+                            UpstreamHttpMethod = new List<string> { "Get" },
+                            Key = "PostDetails"
+                        }
+                    },
+                Aggregates = new List<FileAggregateReRoute>
+                    {
+                        new FileAggregateReRoute
+                        {
+                            UpstreamPathTemplate = "/",
+                            UpstreamHost = "localhost",
+                            ReRouteKeys = new List<string>
+                            {
+                                "Comments",
+                                "UserDetails",
+                                "PostDetails"
+                            },
+                            ReRouteKeysConfig = new List<AggregateReRouteConfig>()
+                            {
+                                new AggregateReRouteConfig(){ReRouteKey = "UserDetails",JsonPath = "$[*].writerId",Parameter = "userId"},
+                                new AggregateReRouteConfig(){ReRouteKey = "PostDetails",JsonPath = "$[*].postId",Parameter = "postId"}
+                            },
+                        }
+                    }
+            };
+
+            var userDetailsResponseContent = @"{""id"":1,""firstName"":""abolfazl"",""lastName"":""rajabpour""}";
+            var postDetailsResponseContent = @"{""id"":1,""title"":""post1""}";
+            var commentsResponseContent = @"[{""id"":1,""writerId"":1,""postId"":2,""text"":""text1""},{""id"":2,""writerId"":1,""postId"":2,""text"":""text2""}]";
+
+            var expected = "{\"Comments\":" + commentsResponseContent + ",\"UserDetails\":" + userDetailsResponseContent + ",\"PostDetails\":" + postDetailsResponseContent + "}";
+
+            this.Given(x => x.GivenServiceOneIsRunning("http://localhost:51889", "/", 200, commentsResponseContent))
+                .Given(x => x.GivenServiceTwoIsRunning("http://localhost:51890", "/users/1", 200, userDetailsResponseContent))
+                .Given(x => x.GivenServiceTwoIsRunning("http://localhost:51887", "/posts/2", 200, postDetailsResponseContent))
+                .And(x => _steps.GivenThereIsAConfiguration(configuration))
+                .And(x => _steps.GivenOcelotIsRunning())
+                .When(x => _steps.WhenIGetUrlOnTheApiGateway("/"))
+                .Then(x => _steps.ThenTheStatusCodeShouldBe(HttpStatusCode.OK))
+                .And(x => _steps.ThenTheResponseBodyShouldBe(expected))
+                .BDDfy();
+        }
+
         [Fact]
         public void should_return_response_200_with_simple_url_user_defined_aggregate()
         {
@@ -189,8 +283,8 @@ namespace Ocelot.AcceptanceTests
                             UpstreamHost = "localhost",
                             ReRouteKeys = new List<string>
                             {
-                                "Tom",
+                                "Laura",
-                                "Laura"
+                                "Tom"
                             },
                             Aggregator = "FakeDefinedAggregator"
                         }
@@ -258,8 +352,8 @@ namespace Ocelot.AcceptanceTests
                             UpstreamHost = "localhost",
                             ReRouteKeys = new List<string>
                             {
-                                "Tom",
+                                "Laura",
-                                "Laura"
+                                "Tom"
                             }
                         }
                     }
@@ -326,8 +420,9 @@ namespace Ocelot.AcceptanceTests
                             UpstreamHost = "localhost",
                             ReRouteKeys = new List<string>
                             {
-                                "Tom",
+                                "Laura",
-                                "Laura"
+                                "Tom"
+
                             }
                         }
                     }
@@ -394,8 +489,8 @@ namespace Ocelot.AcceptanceTests
                             UpstreamHost = "localhost",
                             ReRouteKeys = new List<string>
                             {
-                                "Tom",
+                                "Laura",
-                                "Laura"
+                                "Tom"
                             }
                         }
                     }
@@ -462,8 +557,8 @@ namespace Ocelot.AcceptanceTests
                             UpstreamHost = "localhost",
                             ReRouteKeys = new List<string>
                             {
-                                "Tom",
+                                "Laura",
-                                "Laura"
+                                "Tom"
                             }
                         }
                     }

test/Ocelot.UnitTests/Middleware/ResponseAggregatorFactoryTests.cs

@@ -18,7 +18,8 @@ namespace Ocelot.UnitTests.Middleware
         public ResponseAggregatorFactoryTests()
         {
             _provider = new Mock<IDefinedAggregatorProvider>();
-            _factory = new InMemoryResponseAggregatorFactory(_provider.Object);
+            _aggregator = new SimpleJsonResponseAggregator();
+            _factory = new InMemoryResponseAggregatorFactory(_provider.Object, _aggregator);
         }
         
         [Fact]

test/Ocelot.UnitTests/Middleware/SimpleJsonResponseAggregatorTests.cs

@@ -7,11 +7,11 @@ using Castle.Components.DictionaryAdapter;
 using Microsoft.AspNetCore.Http;
 using Ocelot.Configuration;
 using Ocelot.Configuration.Builder;
-using Ocelot.Errors;
+using Ocelot.Configuration.File;
 using Ocelot.Middleware;
 using Ocelot.Middleware.Multiplexer;
-using Ocelot.Request.Middleware;
 using Ocelot.UnitTests.Responder;
+using Ocelot.Values;
 using Shouldly;
 using TestStack.BDDfy;
 using Xunit;
@@ -30,6 +30,58 @@ namespace Ocelot.UnitTests.Middleware
             _aggregator = new SimpleJsonResponseAggregator();
         }
 
+        [Fact]
+        public void should_aggregate_n_responses_and_set_response_content_on_upstream_context_withConfig()
+        {
+            var commentsDownstreamReRoute = new DownstreamReRouteBuilder().WithKey("Comments").Build();
+
+            var userDetailsDownstreamReRoute = new DownstreamReRouteBuilder().WithKey("UserDetails")
+                .WithUpstreamPathTemplate(new UpstreamPathTemplate("", 0, false, "/v1/users/{userId}"))
+                .Build();
+
+            var downstreamReRoutes = new List<DownstreamReRoute>
+            {
+                commentsDownstreamReRoute,
+                userDetailsDownstreamReRoute
+            };
+
+            var reRoute = new ReRouteBuilder()
+                .WithDownstreamReRoutes(downstreamReRoutes)
+                .WithAggregateReRouteConfig(new List<AggregateReRouteConfig>()
+                {
+                    new AggregateReRouteConfig(){ReRouteKey = "UserDetails",JsonPath = "$[*].writerId",Parameter = "userId"}
+                })
+                .Build();
+
+            var commentsResponseContent = @"[{""id"":1,""writerId"":1,""postId"":1,""text"":""text1""},{""id"":2,""writerId"":2,""postId"":2,""text"":""text2""},{""id"":3,""writerId"":2,""postId"":1,""text"":""text21""}]";
+            var commentsDownstreamContext = new DownstreamContext(new DefaultHttpContext())
+            {
+                DownstreamResponse = new DownstreamResponse(new StringContent(commentsResponseContent, Encoding.UTF8, "application/json"), HttpStatusCode.OK, new EditableList<KeyValuePair<string, IEnumerable<string>>>(), "some reason"),
+                DownstreamReRoute = commentsDownstreamReRoute
+            };
+
+            var userDetailsResponseContent = @"[{""id"":1,""firstName"":""abolfazl"",""lastName"":""rajabpour""},{""id"":2,""firstName"":""reza"",""lastName"":""rezaei""}]";
+            var userDetailsDownstreamContext = new DownstreamContext(new DefaultHttpContext())
+            {
+                DownstreamResponse = new DownstreamResponse(new StringContent(userDetailsResponseContent, Encoding.UTF8, "application/json"), HttpStatusCode.OK, new List<KeyValuePair<string, IEnumerable<string>>>(), "some reason"),
+                DownstreamReRoute = userDetailsDownstreamReRoute
+            };
+
+            var downstreamContexts = new List<DownstreamContext> { commentsDownstreamContext, userDetailsDownstreamContext };
+
+            var expected = "{\"Comments\":" + commentsResponseContent + ",\"UserDetails\":" + userDetailsResponseContent + "}";
+
+            this.Given(x => GivenTheUpstreamContext(new DownstreamContext(new DefaultHttpContext())))
+                .And(x => GivenTheReRoute(reRoute))
+                .And(x => GivenTheDownstreamContext(downstreamContexts))
+                .When(x => WhenIAggregate())
+                .Then(x => ThenTheContentIs(expected))
+                .And(x => ThenTheContentTypeIs("application/json"))
+                .And(x => ThenTheReasonPhraseIs("cannot return from aggregate..which reason phrase would you use?"))
+                .BDDfy();
+        }
+
+
         [Fact]
         public void should_aggregate_n_responses_and_set_response_content_on_upstream_context()
         {