gateway proxy middleware function maturation

src/DotNetCore.CAP/Dashboard/GatewayProxy/GatewayProxyMiddleware.cs

@@ -1,18 +1,31 @@
 using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Net;
+using System.Net.Http;
 using System.Threading.Tasks;
 using DotNetCore.CAP.Dashboard.GatewayProxy.Requester;
+using DotNetCore.CAP.NodeDiscovery;
 using Microsoft.AspNetCore.Http;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Primitives;
 
 namespace DotNetCore.CAP.Dashboard.GatewayProxy
 {
-    public class GatewayProxyMiddleware : GatewayProxyMiddlewareBase
+    public class GatewayProxyMiddleware
     {
+        private const string NODE_COOKIE_NAME = "cap.node";
+
         private readonly RequestDelegate _next;
         private readonly ILogger _logger;
         private readonly IRequestMapper _requestMapper;
         private readonly IHttpRequester _requester;
 
+        private INodeDiscoveryProvider _discoveryProvider;
+
+        protected HttpRequestMessage DownstreamRequest { get; set; }
+
         public GatewayProxyMiddleware(RequestDelegate next,
            ILoggerFactory loggerFactory,
            IRequestMapper requestMapper,
@@ -24,39 +37,108 @@ namespace DotNetCore.CAP.Dashboard.GatewayProxy
             _requester = requester;
         }
 
-        public async Task Invoke(HttpContext context, IRequestScopedDataRepository requestScopedDataRepository)
+        public async Task Invoke(HttpContext context,
+            DiscoveryOptions discoveryOptions,
+            INodeDiscoveryProvider discoveryProvider)
         {
-            _requestScopedDataRepository = requestScopedDataRepository;
+            _discoveryProvider = discoveryProvider;
 
-            _logger.LogDebug("started calling gateway proxy middleware");
+            var request = context.Request;
+            var pathMatch = discoveryOptions.MatchPath;
+            var isCapRequest = request.Path.StartsWithSegments(
+                new PathString(pathMatch),
+                out PathString matchedPath,
+                out PathString remainingPath);
 
-            var downstreamRequest = await _requestMapper.Map(context.Request);
+            var isSwitchNode = request.Cookies.TryGetValue(NODE_COOKIE_NAME, out string requestNodeId);
+            var isCurrentNode = discoveryOptions.NodeId.ToString() == requestNodeId;
 
-            _logger.LogDebug("setting downstream request");
-
-            SetDownstreamRequest(downstreamRequest);
+            if (!isCapRequest || !isSwitchNode || isCurrentNode)
+            {
+                await _next.Invoke(context);
+            }
+            else
+            {
+                _logger.LogDebug("started calling gateway proxy middleware");
+
+                if (TryGetRemoteNode(requestNodeId, out Node node))
+                {
+                    try
+                    {
+                        DownstreamRequest = await _requestMapper.Map(request);
+
+                        SetDownStreamRequestUri(node, request.Path.Value);
+
+                        var response = await _requester.GetResponse(DownstreamRequest);
+
+                        await SetResponseOnHttpContext(context, response);
+                    }
+                    catch(Exception ex)
+                    {
+                        _logger.LogError(ex.Message);
+                    }
+                }
+                else
+                {
+                    context.Response.Cookies.Delete(NODE_COOKIE_NAME);
+                    await _next.Invoke(context);
+                }
+            }
+        }
 
-            _logger.LogDebug("setting upstream request");
+        private bool TryGetRemoteNode(string requestNodeId, out Node node)
+        {
+            var nodes = _discoveryProvider.GetNodes().GetAwaiter().GetResult();
+            node = nodes.FirstOrDefault(x => x.Id == requestNodeId);
+            return node != null;
+        }
 
-            SetUpstreamRequestForThisRequest(DownstreamRequest);
+        private void SetDownStreamRequestUri(Node node, string requestPath)
+        {
+            var uriBuilder = new UriBuilder("http://", node.Address, node.Port, requestPath);
+            DownstreamRequest.RequestUri = uriBuilder.Uri;
+        }
 
-            var uriBuilder = new UriBuilder(DownstreamRequest.RequestUri)
+        public async Task SetResponseOnHttpContext(HttpContext context, HttpResponseMessage response)
+        {
+            foreach (var httpResponseHeader in response.Content.Headers)
             {
-                //Path = dsPath.Data.Value,
-                //Scheme = DownstreamRoute.ReRoute.DownstreamScheme
-            };
+                AddHeaderIfDoesntExist(context, httpResponseHeader);
+            }
 
-            DownstreamRequest.RequestUri = uriBuilder.Uri;
+            var stringContent = await response.Content.ReadAsStringAsync();
+            var content = await response.Content.ReadAsByteArrayAsync();
+
+            AddHeaderIfDoesntExist(context, 
+                new KeyValuePair<string, IEnumerable<string>>("Content-Length", new[] { content.Length.ToString() }));
+
+            context.Response.OnStarting(state =>
+            {
+                var httpContext = (HttpContext)state;
 
-            _logger.LogDebug("started calling request");
+                httpContext.Response.StatusCode = (int)response.StatusCode;
 
-            var response = await _requester.GetResponse(Request);
+                return Task.CompletedTask;
 
-            _logger.LogDebug("setting http response message");
+            }, context);
 
-            SetHttpResponseMessageThisRequest(response);
+            using (Stream stream = new MemoryStream(content))
+            {
+                if (response.StatusCode != HttpStatusCode.NotModified)
+                {
+                    await stream.CopyToAsync(context.Response.Body);
+                }
+            }
+        }
 
-            _logger.LogDebug("returning to calling middleware");
+        private static void AddHeaderIfDoesntExist(HttpContext context, 
+            KeyValuePair<string, IEnumerable<string>> httpResponseHeader)
+        {
+            if (!context.Response.Headers.ContainsKey(httpResponseHeader.Key))
+            {
+                context.Response.Headers.Add(httpResponseHeader.Key,
+                    new StringValues(httpResponseHeader.Value.ToArray()));
+            }
         }
     }
 }

src/DotNetCore.CAP/Dashboard/GatewayProxy/IRequestMapper.Default.cs

@@ -12,7 +12,7 @@ namespace DotNetCore.CAP.Dashboard.GatewayProxy
 {
     public class RequestMapper : IRequestMapper
     {
-        private readonly string[] _unsupportedHeaders = { "host" };
+        private readonly string[] _unsupportedHeaders = { "host", "cookie" };
         private const string SchemeDelimiter = "://";
 
         public async Task<HttpRequestMessage> Map(HttpRequest request)