old_flsystem/类库/Api.Framework/NoticeSocketClient.cs

using System;
using System.Linq;
using System.Text;
using System.Threading;
using System.Xml;
using Api.Framework.Utils;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using WebSocket = WebSocketSharp.WebSocket;
using CsharpHttpHelper;
using WebSocketSharp;

namespace Api.Framework
{
    /// <summary>
    /// 云通知数据
    /// </summary>
    public class CloudNoticeEventArgs : EventArgs
    {
        /// <summary>
        /// 通知数据ID,通过这个排重
        /// </summary>
        public string Id { get; set; }
        /// <summary>
        /// 通知内容
        /// </summary>
        public string Body { get; set; }
        /// <summary>
        /// 命令ID
        /// </summary>
        public string CommonId { get; set; }
        /// <summary>
        /// 发布时间
        /// </summary>
        public DateTime CreateDateTime { get; set; }
        /// <summary>
        /// 信号通道
        /// </summary>
        public string Channel { get; set; }
    }
    /// <summary>
    /// 发送Ack模式通知数据
    /// </summary>
    public class AckNoticeData : EventArgs
    {
        /// <summary>
        /// ID
        /// </summary>
        public string Id { get; set; }
        /// <summary>
        /// 命令ID
        /// </summary>
        public string CommonId { get; set; }
        /// <summary>
        /// 信号道
        /// </summary>
        public string Channel { get; set; }
        /// <summary>
        /// 数据
        /// </summary>
        public string Data { get; set; }
        /// <summary>
        /// 生成时间
        /// </summary>
        public DateTime CreateDateTime { get; set; }
        /// <summary>
        /// 标记被删除
        /// </summary>
        public bool IsRemove { get; set; }
        /// <summary>
        /// 重试
        /// </summary>
        public bool IsRetry { get; set; }
    }
    /// <summary>
    /// 通知客户端
    /// </summary>
    public class NoticeSocketClient
    {
       
        /// <summary>
        /// 获取数据
        /// </summary>
        /// <param name="key"></param>
        /// <returns></returns>
        public string GetData(string key)
        {
            var http = new HttpHelper();
            var html = http.GetHtml($"http://event.api.52cmg.cn/api/TransferData/Get?key={key}&rd={DateTime.Now.Ticks}").Html;
            var rest = JObject.Parse(html);
            //if (!(rest["Ok"] ?? false).Value<bool>() throw new Exception(html);
            //return (rest["Data"] ?? string.Empty).Value<string>();

            //EventClient.OnEvent($"获取淘客域名数据", $"{html}");

            if (!(rest["Ok"] ?? false).Value<bool>())
                throw new Exception(html);
            return (rest["Data"] ?? string.Empty).Value<string>();

        }

    }


    /// <summary>
    /// 远程调度服务连接
    /// </summary>
    public class RemoteCloudConnectService
    {




        /// <summary>
        /// 初始化
        /// </summary>
        public void Initialize()
        {
            this.Url = "ws://event.api.52cmg.cn";
            ThreadPool.QueueUserWorkItem(o => { DisputeConnect(); });
        }

        /// <summary>
        /// 争抢连接
        /// </summary>
        private void DisputeConnect()
        {
            try
            {
                Start();
            }
            finally
            {
            }
        }

        /// <summary>
        /// 地址
        /// </summary>
        public string Url { get; set; }
        /// <summary>
        /// 启动 连接
        /// </summary>
        public void Start()
        {
            while (true)
            {
                if (Connect())
                {
                    return;
                }

                Thread.Sleep(5000);
            }
        }
        /// <summary>
        /// 是否正常连接
        /// </summary>
        public bool IsConnect { get; private set; }
        /// <summary>
        /// 收到消息
        /// </summary>
        public event Action<MessageEventArgs> OnMessage;
        /// <summary>
        /// 收到云端通知事件
        /// </summary>
        public event EventHandler<CloudNoticeEventArgs> CloudNoticeEvent;
        /// <summary>
        /// 连接打开
        /// </summary>
        public event Action OnOpen;

        /// <summary>
        /// websocket
        /// </summary>
        private WebSocket _ws;
        /// <summary>
        /// ping
        /// </summary>
        private string _pingThradId;
        /// <summary>
        /// ping失败次数
        /// </summary>
        public int PingFailureCount { get; set; }
        /// <summary>
        /// 重新连接次数
        /// </summary>
        public int ReconnectionCount { get; set; }
        /// <summary>
        /// 连接ID
        /// </summary>
        public string ConnectId { get; set; } = Guid.NewGuid().ToString("N");

        /// <summary>
        /// 连接服务器
        /// </summary>
        private bool Connect()
        {
            try
            {
                if (IsConnect)
                {
                    return true;
                }
                if (string.IsNullOrWhiteSpace(Url))
                {
                    return false;
                }
                _ws = new WebSocket(this.Url + "/api/push?id=" + this.ConnectId);
                _ws.OnOpen += _ws_OnOpen;
                _ws.OnClose += _ws_OnClose;
                _ws.OnMessage += _ws_OnMessage;
                _ws.OnError += _ws_OnError;
                _ws.Connect();
                Thread.Sleep(1000);
                if (!IsConnect)
                {
                    return false;
                }
                //连接成功
                StartPing();
            }
            catch (Exception e)
            {
               // _log.Error(e);
            }
            return true;
        }
        /// <summary>
        /// 错误信息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        /// <exception cref="NotImplementedException"></exception>
        private void _ws_OnError(object sender, ErrorEventArgs e)
        {
            //_log.Warn(e.Exception, "知易云服务连接断开" + e.Message);
            this.IsConnect = false;
            OnReconnection();
        }
        /// <summary>
        /// 连接断开
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void _ws_OnClose(object sender, CloseEventArgs e)
        {
           // _log.Warn("知易云服务连接断开");
            this.IsConnect = false;
            OnReconnection();
        }

        /// <summary>
        /// 连接打开
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void _ws_OnOpen(object sender, EventArgs e)
        {
           // _log.Info("知易云服务连接成功");
            this.IsConnect = true;
            OnOpen?.Invoke();
        }
        /// <summary>
        /// 启动Ping
        /// </summary>
        public void StartPing()
        {
           // _log.Info("启动心跳检测");
            var tr = new Thread(() =>
            {
                var id = _pingThradId = Guid.NewGuid().ToString();
                while (id == _pingThradId)
                {
                    Thread.Sleep(1000 * 10);
                    if (_ws == null)
                    {
                        return;
                    }
                    try
                    {
                        if (_ws.ReadyState == WebSocketState.Open)
                        {
                            _ws.Send("\u0001\u0002");
                            this.PingFailureCount = 0;
                        }
                        else
                        {
                            this.PingFailureCount = this.PingFailureCount + 1;
                        }
                    }
                    catch
                    {
                        // ignored
                        this.PingFailureCount = this.PingFailureCount + 1;
                    }
                    finally
                    {
                        if (this.PingFailureCount > 5)
                        {
                            this.IsConnect = false;
                            //_log.Error("心跳异常连接断开");
                            _ws?.Close();
                        }
                    }
                }
            });
            tr.IsBackground = true;
            tr.Start();
        }

        /// <summary>
        /// 是否允许重新连接
        /// </summary>
        public bool AllowReconnected { get; set; } = true;
        private bool _isBeginRconnection;
        private readonly object _lockReconnection = new object();
        private void OnReconnection()
        {
            if (_isBeginRconnection || !AllowReconnected) return;
            lock (_lockReconnection)
            {
                if (_isBeginRconnection || !AllowReconnected) return;
                _onReconnection();
            }
        }
        private void _onReconnection()
        {
            if (_isBeginRconnection || !AllowReconnected) return;
            try
            {
                _isBeginRconnection = true;
                //_log.Debug("重连中...");
                while (true)
                {
                    Thread.Sleep(1000 * 5);
                    if (AllowReconnected && !IsConnect)
                    {
                        this.ReconnectionCount++;
                        if (Connect())
                        {
                           // _log.Debug("重连成功");
                            break;
                        }
                    }
                }
            }
            finally
            {
                _isBeginRconnection = false;
            }
        }
        /// <summary>
        /// 收到消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        /// <exception cref="NotImplementedException"></exception>
        private void _ws_OnMessage(object sender, MessageEventArgs e)
        {
            if (e.IsText)
            {
                var index = e.Data.IndexOf("\u0000", StringComparison.Ordinal);
                if (index <= -1)
                {
                    return;
                }

                LogHelper.GetSingleObj().Info("接收到通知",$"{e.Data}");

                var commonId = e.Data.Substring(0, index);
                var body = e.Data.Substring(index + 1);
                if (commonId == "notice")//收到通知
                {
                    var ackNoticeData = Newtonsoft.Json.JsonConvert.DeserializeObject<AckNoticeData>(body);
                    if (ackNoticeData == null)
                    {
                        return;
                    }
                    if (ackNoticeData.IsRemove)
                    {
                        return;
                    }
                    this.CloudNoticeEvent?.Invoke(this, new CloudNoticeEventArgs()
                    {
                        Id = ackNoticeData.Id,
                        Body = ackNoticeData.Data,
                        CommonId = ackNoticeData.CommonId,
                        CreateDateTime = ackNoticeData.CreateDateTime,
                        Channel = ackNoticeData.Channel
                    });
                    _ws.Send("notice_ack\u0000" + ackNoticeData.Id);//回复确认标示
                    return;
                }
            }

            this.OnMessage?.Invoke(e);
        }
        /// <summary>
        /// 发送字节
        /// </summary>
        /// <param name="bytes"></param>
        public void SendBytes(byte[] bytes)
        {
            if (IsConnect)
            {
                _ws.Send(bytes);
            }
        }
        /// <summary>
        /// 发送字符串
        /// </summary>
        /// <param name="context"></param>
        public void SendString(string context)
        {
            if (IsConnect)
            {
                _ws.Send(context);
            }
        }

    }
}