FISLib_Old/FISSDK/JsonRpcLite/Network/JsonRpcHttpServerEngineBase.cs

using JsonRpcLite.Log;
using JsonRpcLite.Rpc;
using JsonRpcLite.Services;
using JsonRpcLite.Utilities;
using System;
using System.Buffers;
using System.Collections.Generic;
using System.IO;
using System.IO.Compression;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace JsonRpcLite.Network
{
    public abstract class JsonRpcHttpServerEngineBase : IJsonRpcServerEngine
    {
        /// <summary>
        /// Gets the registered plugins.
        /// </summary>
        protected List<IJsonRpcHttpServerEnginePlugin> Plugins { get; } = new List<IJsonRpcHttpServerEnginePlugin>();

        /// <summary>
        /// Gets the engine name.
        /// </summary>
        public string Name { get; protected set; }

        /// <summary>
        /// Start the engine and use given router to handle request.
        /// </summary>
        /// <param name="router">The router which will handle the request.</param>
        public abstract void Start(IJsonRpcRouter router);

        /// <summary>
        /// Stop the engine.
        /// </summary>
        public abstract void Stop();

        /// <summary>
        /// Register the plugin into the engine.
        /// </summary>
        /// <param name="plugin">The plugin to register.</param>
        public void RegisterPlugin(IJsonRpcHttpServerEnginePlugin plugin)
        {
            Plugins.Add(plugin);
        }

        /// <summary>
        /// Read the request data from the input stream.
        /// </summary>
        /// <param name="inputStream">The stream to handle.</param>
        /// <param name="requestData">The request data to fill.</param>
        /// <param name="dataLength">The data length to read.</param>
        /// <param name="cancellationToken"></param>
        private async Task ReadRequestDataAsync(Stream inputStream, byte[] requestData, int dataLength, CancellationToken cancellationToken = default)
        {
            var length = dataLength;
            var offset = 0;
            while (length > 0)
            {
                cancellationToken.ThrowIfCancellationRequested();
                var readLength = await inputStream.ReadAsync(requestData, offset, length, cancellationToken).ConfigureAwait(false);
                length -= readLength;
                offset += readLength;
            }
        }

        /// <summary>
        /// Dispatch request to specified service.
        /// </summary>
        /// <param name="context">The HttpListenerContext</param>
        /// <param name="router">The router to handle the rpc request</param>
        /// <param name="serviceName">The name of the service</param>
        /// <param name="cancellationToken">The cancellation token which can cancel this method</param>
        /// <returns>Void</returns>
        private async Task DispatchAsync(IJsonRpcHttpContext context, IJsonRpcRouter router, string serviceName, CancellationToken cancellationToken = default)
        {
            var dataLength = (int)context.GetRequestContentLength();
            var requestData = ArrayPool<byte>.Shared.Rent(dataLength);
            JsonRpcRequest[] requests;
            try
            {
                var inputStream = context.GetInputStream();
                await ReadRequestDataAsync(inputStream, requestData, dataLength, cancellationToken).ConfigureAwait(false);
                if (Logger.DebugMode)
                {
                    var requestString = Encoding.UTF8.GetString(requestData);
                    Logger.WriteDebug($"Receive request data:{requestString}");
                }

                var handled = false;
                //let the plugin pre process the request data.
                foreach (var plugin in Plugins)
                {
                    var result = plugin.PreProcess(context, requestData);
                    requestData = result.Data;
                    handled = result.Handled;
                    if (handled)
                    {
                        break;
                    }
                }
                if (!handled)
                {
                    requests = await JsonRpcCodec.DecodeRequestsAsync(requestData, cancellationToken, dataLength).ConfigureAwait(false);
                    var responses = await router.DispatchRequestsAsync(serviceName, requests, cancellationToken).ConfigureAwait(false);
                    var responseData = await JsonRpcCodec.EncodeResponsesAsync(responses, cancellationToken).ConfigureAwait(false);
                    //let the plugin post process the response data.
                    foreach (var plugin in Plugins)
                    {
                        var result = plugin.PostProcess(context, responseData);
                        responseData = result.Data;
                        handled = result.Handled;
                        if (handled)
                        {
                            break;
                        }
                    }
                    if (!handled)
                    {
                        await WriteRpcResponsesAsync(context, responseData, cancellationToken).ConfigureAwait(false);
                    }
                    else
                    {
                        //The response data has been post-processed by the plugin, so just return the data to the remote side.
                        await WriteHttpResultAsync(context, (int)HttpStatusCode.OK, Encoding.UTF8.GetString(responseData), cancellationToken).ConfigureAwait(false);
                    }
                }
                else
                {
                    //The request has been pre-processed by the plugin and no need dispatch, so just return the data to the remote side.
                    await WriteHttpResultAsync(context, (int)HttpStatusCode.OK, Encoding.UTF8.GetString(requestData), cancellationToken).ConfigureAwait(false);
                }
            }
            finally
            {
                ArrayPool<byte>.Shared.Return(requestData);
            }
        }

        /// <summary>
        /// Parser the request url, get the calling information.
        /// </summary>
        /// <param name="requestPath">The path requested by the caller.</param>
        /// <returns>The service name parsed from the uri.</returns>
        private string GetRpcServiceName(string requestPath)
        {
            var url = $"{requestPath.Trim('/')}";
            var urlParts = url.Split('/');
            if (urlParts.Length != 1) return null;
            return urlParts[0];
        }

        /// <summary>
        /// Write smd data back to the client.
        /// </summary>
        /// <param name="context">The http context</param>
        /// <param name="smdData">The smd data to write back.</param>
        /// <param name="cancellationToken">The cancellation token which will cancel this method.</param>
        /// <returns>Void</returns>
        private async Task WriteSmdDataAsync(IJsonRpcHttpContext context, byte[] smdData, CancellationToken cancellationToken = default)
        {
            try
            {
                await WriteRpcResultAsync(context, smdData, cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                Logger.WriteWarning($"Write smd data back to client error:{ex}");
            }
        }

        /// <summary>
        /// Write http exception back to the client.
        /// </summary>
        /// <param name="context">The http context</param>
        /// <param name="exception">The exception to write back.</param>
        /// <param name="cancellationToken">The cancel token which will cancel this method.</param>
        /// <returns>Void</returns>
        private async Task WriteHttpExceptionAsync(IJsonRpcHttpContext context, HttpException exception, CancellationToken cancellationToken = default)
        {
            try
            {
                await WriteHttpResultAsync(context, exception.ErrorCode, exception.Message, cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                Logger.WriteWarning($"Write http exception back to client error:{ex}");
            }
        }

        /// <summary>
        /// Write rpc responses back to the client.
        /// </summary>
        /// <param name="context">The http context</param>
        /// <param name="responses">The responses to write back.</param>
        /// <param name="cancellationToken">The cancel token which will cancel this method.</param>
        /// <returns>Void</returns>
        private async Task WriteRpcResponsesAsync(IJsonRpcHttpContext context, byte[] responseData, CancellationToken cancellationToken = default)
        {
            try
            {
                await WriteRpcResultAsync(context, responseData, cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                Logger.WriteWarning($"Write rpc response back to client error:{ex}");
            }
        }

        /// <summary>
        /// Get the compressed(or not) output data according to request.
        /// </summary>
        /// <param name="context">The http context.</param>
        /// <param name="data">The data to output.</param>
        /// <param name="cancellationToken">The cancellation token which will cancel this method.</param>
        /// <returns>The compressed or not output data.</returns>
        private async Task<byte[]> GetOutputDataAsync(IJsonRpcHttpContext context, byte[] data, CancellationToken cancellationToken = default)
        {
            var outputData = data;
            var acceptEncoding = context.GetRequestHeader("Accept-Encoding");
            if (acceptEncoding != null && acceptEncoding.Contains("gzip"))
            {
                context.SetResponseHeader("Content-Encoding", "gzip");
                using var memoryStream = new MemoryStream();
                using var outputStream = new GZipStream(memoryStream, CompressionMode.Compress);
                await outputStream.WriteAsync(outputData, 0, outputData.Length, cancellationToken).ConfigureAwait(false);
                await outputStream.FlushAsync(cancellationToken).ConfigureAwait(false);
                outputData = memoryStream.ToArray();
            }
            else if (acceptEncoding != null && acceptEncoding.Contains("deflate"))
            {
                context.SetResponseHeader("Content-Encoding", "deflate");
                using var memoryStream = new MemoryStream();
                using var outputStream = new DeflateStream(memoryStream, CompressionMode.Compress);
                await outputStream.WriteAsync(outputData, 0, outputData.Length, cancellationToken).ConfigureAwait(false);
                await outputStream.FlushAsync(cancellationToken).ConfigureAwait(false);
                outputData = memoryStream.ToArray();
            }
            return outputData;
        }

        /// <summary>
        /// Write http message to remote side.
        /// </summary>
        /// <param name="context">The  http context.</param>
        /// <param name="statusCode">The status code to return</param>
        /// <param name="message">The message to write back.</param>
        /// <param name="cancellationToken">The cancellation token which will cancel this method.</param>
        /// <returns>Void</returns>
        private async Task WriteHttpResultAsync(IJsonRpcHttpContext context, int statusCode, string message, CancellationToken cancellationToken = default)
        {
            context.SetResponseHeader("Server", "JsonRpcLite");
            context.SetResponseHeader("Access-Control-Allow-Origin", "*");
            context.SetResponseHeader("Access-Control-Allow-Methods", "*");
            context.SetResponseHeader("Access-Control-Allow-Headers", "*");
            context.SetResponseStatusCode(statusCode);
            context.SetResponseContentType("text/html");
            var outputData = await GetOutputDataAsync(context, Encoding.UTF8.GetBytes(message), cancellationToken).ConfigureAwait(false);
            context.SetResponseContentLength(outputData.Length);
            var outputStream = context.GetOutputStream();
            await outputStream.WriteAsync(outputData, 0, outputData.Length, cancellationToken).ConfigureAwait(false);
            await outputStream.FlushAsync(cancellationToken).ConfigureAwait(false);
            if (Logger.DebugMode)
            {
                Logger.WriteDebug($"Response data sent:{message}");
            }
        }

        /// <summary>
        /// Write rpc result struct data to remote side.
        /// </summary>
        /// <param name="context">The context of the http.</param>
        /// <param name="result">The result data to write</param>
        /// <param name="cancellationToken">The cancellation which can cancel this method</param>
        /// <returns>Void</returns>
        private async Task WriteRpcResultAsync(IJsonRpcHttpContext context, byte[] result, CancellationToken cancellationToken = default)
        {
            context.SetResponseHeader("Server", "JsonRpcLite");
            context.SetResponseHeader("Access-Control-Allow-Origin", "*");
            context.SetResponseHeader("Access-Control-Allow-Methods", "*");
            context.SetResponseHeader("Access-Control-Allow-Headers", "*");
            context.SetResponseStatusCode((int)HttpStatusCode.OK);
            context.SetResponseContentType("application/json");
            if (result != null)
            {
                var outputData = await GetOutputDataAsync(context, result, cancellationToken).ConfigureAwait(false);
                context.SetResponseContentLength(outputData.Length);
                var outputStream = context.GetOutputStream();
                await outputStream.WriteAsync(outputData, 0, outputData.Length, cancellationToken).ConfigureAwait(false);
                await outputStream.FlushAsync(cancellationToken).ConfigureAwait(false);
            }
            if (Logger.DebugMode)
            {
                if (result != null)
                {
                    var resultString = Encoding.UTF8.GetString(result);
                    Logger.WriteDebug($"Response data sent:{resultString}");
                }
            }
        }

        /// <summary>
        /// Handle connected request and return result.
        /// </summary>
        /// <param name="context">The http context to handle.</param>
        /// <param name="router">The router to dispatch the request data.</param>
        /// <param name="cancellationToken">The cancellation token which can cancel this method.</param>
        /// <returns>Void</returns>
        protected async Task HandleContextAsync(IJsonRpcHttpContext context, IJsonRpcRouter router, CancellationToken cancellationToken = default)
        {
            var httpMethod = context.GetRequestHttpMethod();
            var requestPath = context.GetRequestPath();
            Logger.WriteVerbose($"Handle request [{httpMethod}]: {requestPath}");
            try
            {
                var serviceName = GetRpcServiceName(requestPath);
                if (string.IsNullOrEmpty(serviceName))
                {
                    Logger.WriteWarning($"Service for request: {requestPath} not found.");
                    throw new HttpException((int)HttpStatusCode.ServiceUnavailable, "Service does not exist.");
                }
                if (httpMethod == "options")
                {
                    try
                    {
                        await WriteHttpResultAsync(context, (int)HttpStatusCode.OK, string.Empty, cancellationToken).ConfigureAwait(false);
                    }
                    catch (Exception ex)
                    {
                        throw new HttpException((int)HttpStatusCode.InternalServerError, ex.Message);
                    }
                }
                else if (httpMethod == "get")
                {
                    var smdRequest = false;
                    var smdIndex = serviceName.LastIndexOf(".smd", StringComparison.InvariantCultureIgnoreCase);
                    if (smdIndex != -1)
                    {
                        serviceName = serviceName.Substring(0, smdIndex);
                        smdRequest = true;
                    }

                    if (!router.ServiceExists(serviceName))
                    {
                        Logger.WriteWarning($"Service for request: {requestPath} not found.");
                        throw new HttpException((int)HttpStatusCode.ServiceUnavailable, $"Service [{serviceName}] does not exist.");
                    }

                    if (smdRequest)
                    {
                        try
                        {
                            var smdData = await router.GetServiceSmdData(serviceName).ConfigureAwait(false);
                            if (smdData != null)
                            {
                                await WriteSmdDataAsync(context, smdData, cancellationToken).ConfigureAwait(false);
                            }
                            else
                            {
                                throw new HttpException((int)HttpStatusCode.NotFound, $"Resource for {requestPath} does not exist.");
                            }
                        }
                        catch (Exception ex)
                        {
                            throw new HttpException((int)HttpStatusCode.InternalServerError, ex.Message);
                        }
                    }
                    else
                    {
                        throw new HttpException((int)HttpStatusCode.NotFound, $"Resource for {requestPath} does not exist.");
                    }
                }
                else if (httpMethod == "post")
                {
                    if (!router.ServiceExists(serviceName))
                    {
                        Logger.WriteWarning($"Service for request: {requestPath} not found.");
                        throw new ServerErrorException("Service does not exist.", $"Service [{serviceName}] does not exist.");
                    }

                    try
                    {
                        await DispatchAsync(context, router, serviceName, cancellationToken).ConfigureAwait(false);
                    }
                    catch (RpcException)
                    {
                        throw;
                    }
                    catch (Exception ex)
                    {
                        throw new ServerErrorException("Internal server error.", ex.Message);
                    }
                }
                else
                {
                    throw new HttpException((int)HttpStatusCode.MethodNotAllowed, $"Invalid http-method:{httpMethod}");
                }
            }
            catch (RpcException rpcException)
            {
                var response = new JsonRpcResponse();
                response.WriteResult(rpcException);
                var responseData = await JsonRpcCodec.EncodeResponsesAsync(new[] { response }, cancellationToken).ConfigureAwait(false);
                await WriteRpcResponsesAsync(context, responseData, cancellationToken).ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                Logger.WriteError($"Handle request {requestPath} error: {ex.Message}");
                if (ex is HttpException httpException)
                {
                    await WriteHttpExceptionAsync(context, httpException, cancellationToken).ConfigureAwait(false);
                }
                else
                {
                    var response = new JsonRpcResponse();
                    var serverError = new InternalErrorException($"Handle request {requestPath} error: {ex.Message}");
                    response.WriteResult(serverError);
                    var responseData = await JsonRpcCodec.EncodeResponsesAsync(new[] { response }, cancellationToken).ConfigureAwait(false);
                    await WriteRpcResponsesAsync(context, responseData, cancellationToken).ConfigureAwait(false);
                }
            }
            finally
            {
                context.Close();
            }
        }
    }
}