using System;
using System.Collections.Generic;
using System.Net.WebSockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace WsSubDemoNetCore
{
    public delegate void EventMapCallback(string msg);


	/// <summary>
	/// WebSocket订阅
	/// </summary>
	public class WSClient : IDisposable
    {
        event EventMapCallback _callback = null;

        bool _stopFlag = false;
        static object _locker = new object();
        ClientWebSocket _webSocket = null;

        QueueObj _queue = null;
        ClientWebSocket Clienter
        {
            get
            {
                if (_webSocket == null)
                {
                    lock (_locker)
                    {
                        if (_webSocket == null)
                        {
                            _webSocket = new ClientWebSocket();
                        }
                    }
                }

                return _webSocket;
            }
        }

        string _token;
        string _connectaddress;
        string _fields;
        string _symbols;
        string _routs;
		string _markets;
		public void Init(int queuelen, string token,string hostip,int port,string fields,string symbols, string routs,string markets)
        {
            if (queuelen < 100 || queuelen > 100000)
                queuelen = 2000;

            _queue = new QueueObj(queuelen);

            _token = token;
            _connectaddress = string.Format("ws://{0}:{1}", hostip, port);
            _fields = fields;
			_symbols = symbols;
            _routs = routs;
			_markets = markets;
		}

        public void Start(EventMapCallback callback, bool statistics)
        {
            _callback = callback;
            _statistics = statistics;

            new Task(dataReceive).Start();

            new Task(checkConnect).Start();

            new Task(queueProcess).Start();

            if(_statistics)
            {
                _timer = new System.Timers.Timer();
                _timer.Interval = 60000;
                _timer.Elapsed += _timer_Statistics;
                _timer.Start();
            }
        }

        private void _timer_Statistics(object sender, System.Timers.ElapsedEventArgs e)
        {
            int receivecount = _receivecount;
            int dealcount = _dealcount;
            int misscount = _misscount;

            Log4net.InfoFormat("Receive:{0},Deal:{1},Miss:{2}", receivecount, dealcount, misscount);

            _receivecount = 0;
            _dealcount = 0;
            _misscount = 0;

			//同步各个周期历史数据
			ReqKData("1M");
			int hour = DateTime.Now.Hour;
			int min = DateTime.Now.Minute;
            if(min%3==0) ReqKData("3M");
			if(min%5==0) ReqKData("5M");
			if(min % 10== 0) ReqKData("10M");
			if (min % 15 == 0) ReqKData("15M");
			if (min % 30 == 0) ReqKData("30M");
            if (min == 0 || min == 5)
            {
                ReqKData("1H");
                ReqKData("2H");
                ReqKData("4H");
            }
            if(hour==0 && (min==1 || min == 10)) ReqKData("D");
			//Synchronize end 
		}

		System.Timers.Timer _timer;
        bool _statistics = false;

		/// <summary>
		/// 连续重连多次时，关闭程序
		/// </summary>
		int _reconnectTimes = 0;

        int _receivecount=0;
        int _dealcount = 0;
        int _misscount = 0;

        void reconnect()
        {
			//多次重连，肯定哪里出错了，关闭程序
			if (_reconnectTimes > 30)
            {
                Release();
                return;
            }

			Log4net.WarnFormat("ws reconnect");
            _reconnectTimes++;

			if (Clienter.State != WebSocketState.Open)
            {
                Release(false);

                _stopFlag = false;
                Clienter.ConnectAsync(new Uri(string.Format("{0}/connect/json/{1}", _connectaddress, _token)), CancellationToken.None).Wait();
            }
        }

		/// <summary>
		/// 最后一次正常心跳时间
		/// </summary>
		long _lstheartbeat = 0;


        void SubSymbols()
        {
            if (!string.IsNullOrEmpty(_symbols))
                sendCmd("/sub/" + _symbols);
        }

        void SubRouts()
        {
            if (!string.IsNullOrEmpty(_routs))
                sendCmd("/subrout/" + _routs);
        }

		void SubMarkets()
		{
			if (!string.IsNullOrEmpty(_markets))
				sendCmd("/submkt/" + _markets);
		}

		public void UnSub()
        {
            if (!string.IsNullOrEmpty(_routs))
                sendCmd("/unsub/" + _routs);

            if (!string.IsNullOrEmpty(_symbols))
                sendCmd("/unsub/" + _symbols);

			if (!string.IsNullOrEmpty(_markets))
				sendCmd("/unsub/" + _markets);
		}

        void SubFields()
        {
            if (!string.IsNullOrEmpty(_fields))
                sendCmd("/fields/" + _fields);
        }

		/// <summary>
		/// 请求一次数据
		/// </summary>
		void ReqData()
        {
			if (!string.IsNullOrEmpty(_symbols))
				sendCmd("/req/" + _symbols);
		}

		/// <summary>
		/// 请求历史数据
		/// </summary>
		/// <param name="period"></param>
		void ReqKData(string period)
		{
            if (!string.IsNullOrEmpty(_symbols))
            {
                string[] syms = _symbols.Split(new char[] { ',' });
                foreach (string sym in syms)
                    sendCmd(string.Format("/reqk/{0},{1},10", sym, period));
            }
		}

		void sendCmd(string cmd)
        {
            try
            {
                if (Clienter.State == WebSocketState.Open)
                {
                    Clienter.SendAsync(new ArraySegment<byte>(Encoding.UTF8.GetBytes(cmd)), WebSocketMessageType.Text, true, CancellationToken.None).Wait();
                }
                else
                {
                    Log4net.WarnFormat("WebSocket Not Open:{0}", Clienter.State);
                }
            }
            catch (Exception ex)
            {
                Log4net.ErrorFormat("WebSocket cmd error:{0}", ex.Message);
            }
        }

		/// <summary>
		/// 检查心跳，确认连接正常
		/// </summary>
		void checkConnect()
        {
            while (!_stopFlag)
            {
                try
                {
                    if (Clienter.State == WebSocketState.Open)
                    {
                        sendCmd("/ping/ok");

                        System.Threading.Thread.Sleep(15000);
                    }
                    else
                    {
                        reconnect();

                        System.Threading.Thread.Sleep(10000);

                        continue;
                    }

                    if (_lstheartbeat != 0 && _lstheartbeat < (CurrentTimeTicks() - 20))
                        reconnect();
                }
                catch (System.Exception ex)
                {
                    Log4net.ErrorFormat("cmd err:{0}", ex.Message);
                    System.Threading.Thread.Sleep(10000);
                }
            }
        }

		/// <summary>
		/// 从程序缓存队列中依次取出数据进行处理
		/// </summary>
		void queueProcess()
        {
            string msg = "";
            while (!_stopFlag)
            {
                try
                {
                    if(_queue.Count>0)
                    {
                        _misscount+=_queue.CheckMaxMiss();

                        if (_queue.TryDequeue(out msg))
                        {
                            if(_callback!=null && !string.IsNullOrEmpty(msg))
                                _callback(msg);

                            _dealcount++;
                        }
                    }
                    else System.Threading.Thread.Sleep(50);
                }
                catch (System.Exception ex)
                {
                    Log4net.ErrorFormat("queue err:{0},{1}", ex.Message,ex.StackTrace);
                    System.Threading.Thread.Sleep(50);
                }
            }
        }

		/// <summary>
		/// 接收到信息分类处理
		/// </summary>
		async void dataReceive()
        {
            while (!_stopFlag)
            {
                if (Clienter.State == WebSocketState.Open)
                {
                    string msg = "";
                    try
                    {
                        var array = new byte[4096];
                        WebSocketReceiveResult result = null;
                        do
                        {
                            result = await Clienter.ReceiveAsync(new ArraySegment<byte>(array), CancellationToken.None);
                            if (!result.CloseStatus.HasValue)
                            {
                                if (result.MessageType != WebSocketMessageType.Text)
                                    break;

                                msg += Encoding.UTF8.GetString(array, 0, result.Count);
                            }
                            else
                            {
                                Log4net.WarnFormat("receive fail status:{0}", result.CloseStatusDescription);
                                System.Threading.Thread.Sleep(3000);
                                break;
                            }
                        } while (!result.EndOfMessage);

                        if (string.IsNullOrEmpty(msg))
                            continue;

                        Ret ret = ServiceStack.Text.JsonSerializer.DeserializeFromString<Ret>(msg);
						if (ret.IsValid())
                        {
                            switch (ret.Cmd)
                            {
                                case "connect":
                                    {
                                        if (ret.Code == 0)
                                        {
                                            Log4net.WarnFormat("ws subs");
                                            _reconnectTimes = 0;

                                            SubFields();

                                            SubSymbols();
                                            SubRouts();
                                            SubMarkets();
                                        }
                                        else
                                        {
                                            Log4net.WarnFormat("connect fail:{0}", msg);
                                        }

                                        break;
                                    }
                                case "heartbeat":
                                    {
                                        if (string.Compare(ret.Msg, "ping") == 0)
                                            sendCmd("/ping/ok");

                                        break;
                                    }
                                case "ping":
                                    {
                                        if (ret.Code == 0)
                                            _lstheartbeat = CurrentTimeTicks();

                                        break;
                                    }
                                case "rm":
								case "reqk":
								default:
                                    {
                                        _receivecount++;
                                        _misscount += _queue.Enqueue(msg);

                                        break;
                                    }
							}
                        }
                    }
                    catch (System.Exception ex)
                    {
                        Log4net.ErrorFormat("rev msg error:{0},{1},{2},{3}", Clienter.CloseStatus, Clienter.CloseStatusDescription, ex.Message, ex.StackTrace, msg);
                        System.Threading.Thread.Sleep(3000);
                    }
                }
                else System.Threading.Thread.Sleep(20);
            }
        }

        void Release(bool stop = true)
        {
            if (stop)
            {
                _stopFlag = stop;

				Console.WriteLine("wsclient已关闭");

				Log4net.Info("ws client Stop!");
			}

            if (_webSocket != null)
            {
                if (_webSocket.State == WebSocketState.Open || _webSocket.State == WebSocketState.Connecting)
                    _webSocket.CloseOutputAsync(WebSocketCloseStatus.NormalClosure, "clientclose", CancellationToken.None).Wait();

                _webSocket.Dispose();
                _webSocket = null;
            }
        }

        long CurrentTimeTicks()
        {
            return (DateTime.Now.Ticks - (new System.DateTime(1970, 1, 1, 0, 0, 0, 0)).ToLocalTime().Ticks) / 10000000;
        }
        public void Dispose()
        {
            Release();

            if (_timer != null)
            {
                _timer.Dispose();
                _timer = null;
            }
        }
    }
}
