gRPC Client Middleware.

Posted 2020-12-30 kingreatwill

　　中间件想必大家不陌生，今天给大家介绍如何实现中间件以及实现gRPC的客户端中间件。

什么是中间件？

   https://docs.microsoft.com/zh-cn/aspnet/core/fundamentals/middleware/index?view=aspnetcore-2.1&tabs=aspnetcore2x

中间件管道

 先定义管道Pipeline

/// <summary>
    /// Built pipeline for gRPC
    /// </summary>
    public class Pipeline
    {
        private PipelineDelagate processChain;

        internal Pipeline(PipelineDelagate middlewareChain)
        {
            processChain = middlewareChain;
        }

        internal Task RunPipeline(MiddlewareContext context)
        {
            return processChain(context);
        }
    }

然后实现PipelineBuilder

/// <summary>
    /// PipelineBuilder
    /// </summary>
    public class PipelineBuilder
    {
        private List<Func<PipelineDelagate, PipelineDelagate>> middlewares = new List<Func<PipelineDelagate, PipelineDelagate>>();

        /// <summary>
        /// Add a middleware
        /// </summary>
        public PipelineBuilder Use(Func<PipelineDelagate, PipelineDelagate> middleware)
        {
            middlewares.Add(middleware);
            return this;
        }

        /// <summary>
        /// Use
        /// </summary>
        public PipelineBuilder Use<T>(params object[] args)
        {
            middlewares.Add(d => WrapClass<T>(d, args));
            return this;
        }

        /// <summary>
        /// UseWhen
        /// </summary>
        public PipelineBuilder UseWhen<T>(Func<MiddlewareContext, bool> condition, params object[] args)
        {
            middlewares.Add(d =>
            {
                return async ctx => { if (condition(ctx)) { await WrapClass<T>(d, args)(ctx); } };
            });
            return this;
        }

        /// <summary>
        /// Use
        /// </summary>
        public PipelineBuilder Use(Func<MiddlewareContext, PipelineDelagate, Task> middleware)
        {
            middlewares.Add(d =>
            {
                return ctx => { return middleware(ctx, d); };
            });
            return this;
        }

        private PipelineDelagate WrapClass<T>(PipelineDelagate next, params object[] args)
        {
            var ctorArgs = new object[args.Length + 1];
            ctorArgs[0] = next;
            Array.Copy(args, 0, ctorArgs, 1, args.Length);
            var type = typeof(T);
            var instance = Activator.CreateInstance(type, ctorArgs);
            MethodInfo method = type.GetMethod("Invoke");
            return (PipelineDelagate)method.CreateDelegate(typeof(PipelineDelagate), instance);
        }

        /// <summary>
        /// Build
        /// </summary>
        public Pipeline Build()
        {
            PipelineDelagate pipeline = ExecuteMainHandler;
            middlewares.Reverse();
            foreach (var middleware in middlewares)
            {
                pipeline = middleware(pipeline);
            }
            return new Pipeline(pipeline);
        }

        internal static Task ExecuteMainHandler(MiddlewareContext context)
        {
            return context.HandlerExecutor();
        }
    }

MiddlewareContext 和 委托定义

/// <summary>
    /// MiddlewareContext
    /// </summary>
    public class MiddlewareContext
    {
        public IMethod Method { get; set; }
        
        public object Request { get; set; }
       
        public object Response { get; set; }

        public string Host { get; set; }

        public CallOptions Options { get; set; }

        internal Func<Task> HandlerExecutor { get; set; }
    }


    public delegate Task PipelineDelagate(MiddlewareContext context);

到这里管道建设完成，那个如何在gRPC中使用呢？

首先实现自己的客户端拦截器MiddlewareCallInvoker

 public sealed class MiddlewareCallInvoker : DefaultCallInvoker
    {
        private readonly Channel grpcChannel;

        private Pipeline MiddlewarePipeline { get; set; }

        public MiddlewareCallInvoker(Channel channel) : base(channel)
        {
            this.grpcChannel = channel;
        }

        public MiddlewareCallInvoker(Channel channel, Pipeline pipeline) : this(channel)
        {
            this.MiddlewarePipeline = pipeline;
        }

        private TResponse Call<TResponse>(Func<MiddlewareContext, TResponse> call, MiddlewareContext context)
        {
         
            TResponse response = default(TResponse);
            if (MiddlewarePipeline != null)
            {
                context.HandlerExecutor = async () =>
                {
                    response = await Task.FromResult(call(context));
                    context.Response = response;
                };
                MiddlewarePipeline.RunPipeline(context).ConfigureAwait(false);
            }
            else
            {
                response = call(context);
            }
            return response;
       
        }

        public override TResponse BlockingUnaryCall<TRequest, TResponse>(Method<TRequest, TResponse> method,
            string host, CallOptions options, TRequest request)
        {
            return Call((context) => base.BlockingUnaryCall((Method<TRequest, TResponse>)context.Method, context.Host, context.Options, (TRequest)context.Request), new MiddlewareContext
            {
                Host = host,
                Method = method,
                Options = options,
                Request = request,
                Response = null
            });
        }

        public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
            Method<TRequest, TResponse> method, string host, CallOptions options, TRequest request)
        {
            return Call((context) => base.AsyncUnaryCall((Method<TRequest, TResponse>)context.Method, context.Host, context.Options, (TRequest)context.Request), new MiddlewareContext
            {
                Host = host,
                Method = method,
                Options = options,
                Request = request,
                Response = null
            });
        }

        public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options,
            TRequest request)
        {
            return Call((context) => base.AsyncServerStreamingCall((Method<TRequest, TResponse>)context.Method, context.Host, context.Options, (TRequest)context.Request), new MiddlewareContext
            {
                Host = host,
                Method = method,
                Options = options,
                Request = request,
                Response = null
            });
        }

        public override AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options)
        {
            return Call((context) => base.AsyncClientStreamingCall((Method<TRequest, TResponse>)context.Method, context.Host, context.Options), new MiddlewareContext
            {
                Host = host,
                Method = method,
                Options = options,
                Request = null,
                Response = null
            });
        }

        public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options)
        {
            return Call((context) => base.AsyncDuplexStreamingCall((Method<TRequest, TResponse>)context.Method, context.Host, context.Options), new MiddlewareContext
            {
                Host = host,
                Method = method,
                Options = options,
                Request = null,
                Response = null
            });
        }
    }

到这里基于管道的中间件基本完成。接下来就是在客户端使用了

 var pipeline = new PipelineBuilder()
              //.Use<ExceptionMiddleware>()
              //.Use<TimerMiddleware>()
              //.Use<LoggingMiddleware>()
              //.Use<TimeoutMiddleware>()
              .Use<PolicyMiddleware>(new PolicyMiddlewareOptions
              {
                  RetryTimes = 2,
                  TimoutMilliseconds = 100
              })
              ;
            //pipeline.Use<LoggingMiddleware>();// pipeline.UseWhen<LoggingMiddleware>(ctx => ctx.Context.Method.EndsWith("SayHello"));
            //pipeline.Use<TimeoutMiddleware>(new TimeoutMiddlewareOptions { TimoutMilliseconds = 1000 });
            //console logger
            pipeline.Use(async (ctx, next) =>
            {
                Console.WriteLine(ctx.Request.ToString());
                await next(ctx);
                Console.WriteLine(ctx.Response.ToString());
            });

            Channel channel = new Channel("127.0.0.1:50051", ChannelCredentials.Insecure);
            MiddlewareCallInvoker callInvoker = new MiddlewareCallInvoker(channel, pipeline.Build());

            var clientApi = new Get.GetClient(callInvoker);

            var replyApi = clientApi.SayHello(new ContractsSample1.HelloApiDemo.HelloRequest { Name = " APII" });
            Console.WriteLine("Greeting: " + replyApi.Message);

　　Middleware

 public class LoggingMiddleware
    {
        private PipelineDelagate _next;

        public LoggingMiddleware(PipelineDelagate next)
        {
            _next = next;
        }

        public async Task Invoke(MiddlewareContext context)
        {
           //Before
            await _next(context);
            //End
        }
    }

　　至此已完成所有工作。希望能够帮助到大家。

如果有机会将给大家介绍gRPC跨平台网关，该网关能够实现自动将下游的gRPC服务转化成http api，只需要你的proto文件即可。如果你使用consul，甚至无需你做任务配置。