SOMS/Yunda.SOMS.DataMonitoringServer.Service/WebSocket/DotNetty/Server/WebSocketServerHandler.cs

using DotNetty.Buffers;
using DotNetty.Codecs.Http;
using DotNetty.Codecs.Http.WebSockets;
using DotNetty.Common.Utilities;
using DotNetty.Transport.Channels;
using DotNetty.Transport.Channels.Groups;
using DotNettyHelper;
using Newtonsoft.Json;
using System;
using System.Diagnostics;
using System.Text;
using System.Threading.Tasks;
using ToolLibrary.LogHelper;
using Yunda.ISAS.DataMonitoringServer.WebSocket.Model;
using static DotNetty.Codecs.Http.HttpResponseStatus;
using static DotNetty.Codecs.Http.HttpVersion;
using HttpMethod = DotNetty.Codecs.Http.HttpMethod;

namespace Yunda.ISAS.DataMonitoringServer.WebSocket.DotNetty.Server
{
    public delegate void ExceptionCaughtDelegate(ExceptionCaughtEventArgs exceptionCaughtEventArgs);

    public delegate void HandlerDelegate(HandlerEventArgs hanlerEventArgs);

    public delegate void ReceiveMessageDelegate(ReceiveMessageEventArgs<DataMonitorMessageModel> recMsgEventArgs);

    public sealed class WebSocketServerHandler : SimpleChannelInboundHandler<object>
    {
        private string _websocketPath;
        public WebSocketServerHandler(string path)
        {
            _websocketPath = path;
        }
        private WebSocketServerHandshaker _handshaker;

        public event HandlerDelegate HandlerRemovedEvent;

        public event HandlerDelegate HandlerAddedEvent;

        public event ExceptionCaughtDelegate ExceptionCaughtEvent;

        public event ReceiveMessageDelegate ReceiveMessageEvent;

        protected override void ChannelRead0(IChannelHandlerContext ctx, object msg)
        {
            if (msg is IFullHttpRequest request)
            {
                this.HandleHttpRequest(ctx, request);
            }
            else if (msg is WebSocketFrame frame)
            {
                this.HandleWebSocketFrame(ctx, frame);
            }
        }

        private void HandleHttpRequest(IChannelHandlerContext ctx, IFullHttpRequest req)
        {
            // Handle a bad request.
            if (!req.Result.IsSuccess)
            {
                try
                {
                    SendHttpResponse(ctx, req, new DefaultFullHttpResponse(Http11, BadRequest));
                }
                catch (JsonReaderException ex)
                {
                    // return;
                    throw new JsonReaderException(ex.ToString());
                }
                return;
            }

            // Allow only GET methods.
            if (!Equals(req.Method, HttpMethod.Get))
            {
                SendHttpResponse(ctx, req, new DefaultFullHttpResponse(Http11, Forbidden));
                return;
            }
            // Send the demo page and favicon.ico

            if ("/favicon.ico".Equals(req.Uri))
            {
                SendHttpResponse(ctx, req, new DefaultFullHttpResponse(Http11, NotFound));
                return;
            }
            string webSocketPath = @"/" + _websocketPath;
            string[] requestStrs = req.Uri.Split("?");
            if ("/".Equals(requestStrs[0]) || !webSocketPath.Equals(requestStrs[0]))
            {
                SendHttpResponse(ctx, req, new DefaultFullHttpResponse(Http11, NotFound));
                return;
            }
            // Handshake
            string location = GetWebSocketLocation(req);
            var wsFactory = new WebSocketServerHandshakerFactory(location, null, true, 5 * 1024 * 1024);
            this._handshaker = wsFactory.NewHandshaker(req);
            if (this._handshaker == null)
            {
                WebSocketServerHandshakerFactory.SendUnsupportedVersionResponse(ctx.Channel);
            }
            else
            {
                this._handshaker.HandshakeAsync(ctx.Channel, req);
            }
        }

        private void HandleWebSocketFrame(IChannelHandlerContext ctx, WebSocketFrame frame)
        {
            // Check for closing frame
            if (frame is CloseWebSocketFrame)
            {
                this._handshaker.CloseAsync(ctx.Channel, (CloseWebSocketFrame)frame.Retain());
                return;
            }

            if (frame is PingWebSocketFrame)
            {
                ctx.WriteAndFlushAsync(new PongWebSocketFrame((IByteBuffer)frame.Content.Retain()));
                return;
            }

            if (frame is TextWebSocketFrame)
            {
                try
                {
                    #region 接收客户端消息并群发

                    var frameCopy = frame.Copy();
                    ////群发消息
                    //DotNettyWebSocketServer
                    //    .GetWebSocketServerSingleton()
                    //    .GetChannelGroup()
                    //    .WriteAndFlushAsync(frameCopy, new EveryOneBut(ctx.Channel.Id));
                    //// Echo the frame
                    //ctx.WriteAndFlushAsync(frame.Retain());

                    #endregion 接收客户端消息并群发
                    string msg = frameCopy.Content.ToString(Encoding.UTF8);
                    DataMonitorMessageModel msgModel = JsonConvert.DeserializeObject<DataMonitorMessageModel>(msg);
                    if (msgModel == null || msgModel.GroupType == GroupTypeEnum.Error)
                        return;
                    ReceiveMessageEvent?.Invoke(new ReceiveMessageEventArgs<DataMonitorMessageModel>(ctx, msgModel));
                    return;
                }
                
                catch (Exception ex)
                {
                    //throw new Exception(ex.ToString());
                    Log4Helper.Error(this.GetType(), "接收客户端消息并群发", ex);
                }
            }

            if (frame is BinaryWebSocketFrame)
            {
                // Echo the frame
                ctx.WriteAndFlushAsync(frame.Retain());
            }
        }

        private static void SendHttpResponse(IChannelHandlerContext ctx, IFullHttpRequest req, IFullHttpResponse res)
        {
            // Generate an error page if response getStatus code is not OK (200).
            if (res.Status.Code != 200)
            {
                IByteBuffer buf = Unpooled.CopiedBuffer(Encoding.UTF8.GetBytes(res.Status.ToString()));
                res.Content.WriteBytes(buf);
                buf.Release();
                HttpUtil.SetContentLength(res, res.Content.ReadableBytes);
            }

            // Send the response and close the connection if necessary.
            Task task = ctx.Channel.WriteAndFlushAsync(res);
            if (!HttpUtil.IsKeepAlive(req) || res.Status.Code != 200)
            {
                task.ContinueWith((t, c) => ((IChannelHandlerContext)c).CloseAsync(),
                    ctx, TaskContinuationOptions.ExecuteSynchronously);
            }
        }

        private static string GetWebSocketLocation(IFullHttpRequest req)
        {
            bool result = req.Headers.TryGet(HttpHeaderNames.Host, out ICharSequence value);
            Debug.Assert(result, "Host header does not exist.");
            string location = value.ToString() + req.Uri;
            return "ws://" + location;
        }

        // 输出到客户端，也可以在上面的方法中直接调用WriteAndFlushAsync方法直接输出
        //public override void ChannelReadComplete(IChannelHandlerContext context) => context.Flush();

        //捕获 异常，并输出到控制台后断开链接，提示：客户端意外断开链接，也会触发
        public override void ExceptionCaught(IChannelHandlerContext context, Exception exception)
        {

            Log4Helper.Error(this.GetType(), "客户端意外断开链接", exception);
            context.CloseAsync();
            ExceptionCaughtEvent?.Invoke(new ExceptionCaughtEventArgs(context, exception));
        }

        //客户端连接进来时
        public override void HandlerAdded(IChannelHandlerContext context)
        {
            base.HandlerAdded(context);
            HandlerAddedEvent?.Invoke(new HandlerEventArgs(context));
        }

        //客户端下线断线时
        public override void HandlerRemoved(IChannelHandlerContext context)
        {
            base.HandlerRemoved(context);
            HandlerRemovedEvent?.Invoke(new HandlerEventArgs(context));
        }

        //服务器监听到客户端活动时
        public override void ChannelActive(IChannelHandlerContext context)
        {
            base.ChannelActive(context);
        }

        //服务器监听到客户端不活动时
        public override void ChannelInactive(IChannelHandlerContext context)
        {
            base.ChannelInactive(context);
        }
    }

    internal class EveryOneBut : IChannelMatcher
    {
        private readonly IChannelId id;

        public EveryOneBut(IChannelId id)
        {
            this.id = id;
        }

        public bool Matches(IChannel channel) => channel.Id != this.id;
    }
}