OHV/Dev/Connection/OHVProtocolServer/OHVConnector/Manager.cs

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using GSG.NET.Concurrent;
using GSG.NET.LINQ;
using GSG.NET.Logging;
using GSG.NET.Quartz;
using GSG.NET.TCP;
using GSG.NET.Utils;
using GSG.NET.Extensions;

namespace OHVConnector
{
    public partial class Manager
    {
        const byte STX = 0x02;
        const byte ETX = 0x03;

        const long INIT_CTRL_SYSBYTE = 0x40000000;
        long sysbyte = 0;
        long ctrl_sysbyte = INIT_CTRL_SYSBYTE;

        static Logger logger = Logger.GetLogger();

        TsQueue<QueueObject> qQ = new TsQueue<QueueObject>();
        TsQueue<OCSMessage> qqW = new TsQueue<OCSMessage>();

        TimerTemplate<byte, OCSMessage> quzT3 = new TimerTemplate<byte, OCSMessage>();
        TsMap<long, SyncObject> ddReq = new TsMap<long, SyncObject>(); //sync 통신을 위함.

        TcpConnector h = new TcpConnector();

        Thread _TQ;//pumping queue thread
        Thread _TW;//write
        Thread _TR;//read
        Thread _TLINK;//Linktest

        bool ModeActive { get; set; }
        bool? inited;

        public Config Config { get; set; }

        #region Properties
        public bool Connected
        {
            get { return h.Connected; }
        }

        /// <summary>
        /// 연결 시도 가능여부 체크
        /// <para>Connect 호출 or 연결중: true</para>
        /// <para>최초 or Disconnect 호출: false</para>
        /// </summary>
        public bool Connecting
        {
            get { return inited.HasValue ? inited.Value : false; }
        }

        #endregion


        #region Constactor
        public Manager()
        {
            ModeActive = true;
            Config = new Config();
            quzT3.OnTimeout += _OnTimeout;
            h.OnTcpStateChanged += _OnLog;
        }
        #endregion

        #region Connection Method
        public void Connect(bool active = false)
        {
            ModeActive = active;
            if (!inited.HasValue)//최초 한번 실행.
            {
                _TQ = ThreadUtils.Invoke(_ThPullQueue);
                inited = false;
            }
            Assert.IsFalse(inited.Value, "Already connecting");

            inited = true;

            _TW = ThreadUtils.Invoke(_ThWriteTcp);
            _TR = ThreadUtils.Invoke(_ThReadTcp);
            _TLINK = ThreadUtils.Invoke(_ThLinkQuz);

            _OnLog("OHV CONNECT REQ " + Config);
        }

        public void Disconnect()
        {
            //if (inited.HasValue && inited.Value)
            if (Connecting)
            {
                _OnLog("HSMS DISCONNECT REQ " + Config);

                inited = false;
                quzT3.StopAll();

                ThreadUtils.Kill(_TLINK);
                ThreadUtils.Kill(_TW);
                h.StopListen();
                ThreadUtils.Kill(_TR);

                h.CloseSocket();//Kill을 사용할 경우 뒤에 존재해야 한다.
            }
        }

        void _OnDicontd(Exception e)
        {
            _OnLog("OHV DISCONNECTED");
            sysbyte = 0;
            ctrl_sysbyte = INIT_CTRL_SYSBYTE;
            quzT3.StopAll();
            ddReq.Clear();

            qQ.Enqueue(new QoNotComm { Arg0 = e });
        }

        void _OnContd()
        {
            _OnLog("OHV CONNECTED");
            qQ.Enqueue(new QoComm());
        }

        void TcpConnect()
        {
            h.Connect(new TcpComm
            {
                Active = ModeActive,
                RetryCnt = 1, //T5를 처리해야 함.
                Ip = Config.IpAddress,
                PortNo = Config.Port,
                T5 = Config.T5,
                T6 = Config.T6,//Config.TcpRecdTimeout,
            });

            if (!h.Connected)
            {
                if (ModeActive)
                    _OnLog("T5 TIMEOUT " + Config.ID);
                return;
            }
            _OnContd();

            ChgTcpTimeout(true);
            //if (ModeActive)
            //SendCtrlMsg(1);//무조건 HSMS Active
        }

        #endregion


        private void _OnLog(string obj)
        {
            qQ.Enqueue(new QoLog { Arg0 = obj });
        }

        private void _OnTimeout(byte id, OCSMessage msg)
        {
            if (null == msg)
            {
                logger.W("T3 [{0}] attachment is null", id);
                return;
            }

            qQ.Enqueue(new QoTimeout { Arg0 = msg });
            //_OnLog("T3 TIMEOUT {0}".format(msg.LogHeader));
        }

        #region Thread Method
        readonly object lockLink = new object();
        void _ThLinkQuz()
        {
            for (; ; )
            {
                try
                {
                    bool waked;

                    if (Config.LinkOn)
                        waked = LockUtils.Wait(Config.TLink, lockLink);
                    else
                        waked = LockUtils.Wait(lockLink);

                    if (waked)
                        continue;//notify: 패킷을 수신할때마다 reset함.

                    if (Connected)//연결여부와 상관없이 thread가 기동되므로 연결시에만.
                    {
                        //SendCtrlMsg(5);
                    }
                }
                catch (ThreadAbortException)
                {
                    break;
                }
                catch (Exception e)
                {
                    logger.E(e);
                }
            }
        }

        void _ThWriteTcp()
        {
            logger.I("Write {0}", ThreadUtils.GetCurrThreadID());
            for (; ; )
            {
                try
                {
                    var v = qqW.Dequeue();

                    this.TcpWriteMsg(v);
                    //v.IsRecd = false;

                    //if (v.AfterMillis > 0)
                    //    LockUtils.Wait(v.AfterMillis);

                    //if (v.CtrlMsg)
                    //    TcpWriteCtrlMsg(v);
                    //else
                    //    TcpWriteNormalMsg(v);
                }
                catch (ThreadAbortException)
                {
                    break;
                }
                catch (Exception e)
                {
                    logger.E(e);
                }
            }
        }

        void _ThReadTcp()
        {
            for (; ; )
            {
                try
                {
                    if (!h.Connected)
                    {
                        TcpConnect();
                        continue;
                    }
                    ReadSocket();
                }
                catch (ObjectDisposedException e)
                {
                    TcpError(e);
                }
                catch (IOException e)
                {
                    TcpError(e);
                }
                catch (ThreadAbortException)
                {
                    _OnLog("DISCONNECT REQUEST APPLIED " + Config);
                    TcpError(new IOException("DISCONNECT REQUEST"));
                    break;
                }
                catch (Exception e)
                {
                    logger.E(e);
                }
            }
        }

        void _ThPullQueue()
        {
            for (; ; )
            {
                try
                {
                    var qo = this.qQ.Dequeue();

                    if (qo is QoRecdUnk)
                    {
                        DelegateUtils.Invoke(OnRecdUnk, qo.Arg0, qo.Arg1);
                        //if (AutoS9Fy)
                        //{
                        //    var v = qo.Arg0 as SFMessage;
                        //    Send(v.S9Fy);
                        //}
                    }
                    else if (qo is QoComm)
                        DelegateUtils.Invoke(OnContd, Config.ID);
                    else if (qo is QoNotComm)
                        DelegateUtils.Invoke(OnDiscontd, Config.ID, qo.Arg0);
                    else if (qo is QoLog)
                        DelegateUtils.Invoke(OnLog, Config.ID, qo.Arg0);
                    else if (qo is QoRecd)
                        DelegateUtils.Invoke(OnRecd, qo.Arg0);
                    else if (qo is QoTimeout)
                    {
                        DelegateUtils.Invoke(OnT3Timeout, qo.Arg0);
                        //if (AutoS9Fy)
                        //{
                        //    var v = qo.Arg0 as SFMessage;
                        //    Send(MessageSupport.MakeS9FX(9, v));
                        //}
                    }
                    else if (qo is QoSent)
                        DelegateUtils.Invoke(OnSent, qo.Arg0);
                    else
                        Assert.Fail("Unk Object {0}", qo);
                }
                catch (ThreadAbortException)
                {
                    break;
                }
                catch (Exception e)
                {
                    logger.E(e);
                }
            }
        }

        #endregion

        #region Read Method
        void ReadSocket()
        {
            h.ReadByte(); //STX

            string revID = string.Empty;
            string sendID = string.Empty;

            if (ModeActive)
            {
                revID = h.ReadAscii(2);
                sendID = h.ReadAscii(5);
            }
            else
            {
                revID = h.ReadAscii(5);
                sendID = h.ReadAscii(2);
            }

            if (!this.Config.ID.Equals(revID))
                OnLog(this.Config.ID, $"RevID Not Equals");

            var ocsMeg = new OCSMessage();
            ocsMeg.RevID = revID;
            ocsMeg.SendID = sendID;
            ocsMeg.Kind = h.ReadAscii(1).ToEnum<eKind>(eKind.Unknown);
            ocsMeg.Tag = h.ReadAscii(4);
            ocsMeg.SubCode = h.ReadAscii(3);

            //CheckSum 을 해야 하나??
            ocsMeg.CheckSum = h.ReadByte();

            h.ReadUntil(ETX);

            //Todo: 응답으로 온건지 그냥 보낸건지 분류가 필요. = CheckSum 을 저장 했다가 이용하자.


            //if (!len.FwBtw(10, MAX_SIZE))
            //    throw new IOException("HSMS ABNORMAL LENGTH:" + len);

            //var head = h.ReadBytes(10);
            //var body = h.ReadBytes(len - 10);

            ChgTcpTimeout(true);//무언가 받으면

            LockUtils.NotifyAll(lockLink);//Linktest thread 변환의 notify

            //var v = new OCSMessage { Header = head, Body = body, IsRecd = true };
            //v.Decoding();

            _OnRecd(ocsMeg);
        }

        void _OnRecd(OCSMessage recd)
        {
            //Alive Check Reply
            if (recd.Kind == eKind.A && !ModeActive) //자동으로 응답을 보낸다. OCS 가 Active 상태
            {
                var reply = new OCSMessage()
                {
                    Id = this.Config.ID,
                    RevID = recd.SendID,
                    SendID = this.Config.ID,
                    Kind = eKind.A,
                    Tag = recd.Tag,
                    SubCode = recd.SubCode,
                };
                Reply(reply);
                return;
            }

            if ( recd.Kind == eKind.C )
            {
                this.qQ.Enqueue( new QoRecd { Arg0 = recd } );
                return;
            }

            //Send 한 Message 의 Reply 로 판단.
            if (this.quzT3.HasId(recd.CheckSum))
            {
                //Send 목록에서 삭제한다.
                this.quzT3.Stop(recd.CheckSum);
                this._OnLog($"[Received] - Reply - {recd.LogFormat()}");
                return;
            }

            this.qQ.Enqueue(new QoRecd { Arg0 = recd });
        }

        void TcpError(Exception e)
        {
            _OnLog(TcpUtils.GetTcpErrMsg(h.IPClient, e));

            h.CloseSocket();
            _OnDicontd(e);

            LockUtils.Wait(1000);//잠시대기.
        }

        #endregion

        #region Write Method
        void TcpWriteMsg(OCSMessage msg)
        {
            qQ.Enqueue(new QoSent { Arg0 = msg });

            this.h.WriteFlush(msg.ToMemoryBuffer().ToBytes);
        }

        void TcpWriteNormalMsg(OCSMessage nm)
        {
            //nm.Encoding();

            //if (nm.IsPrimary && nm.IsWbit)
            //    quzT3.StartOnce(Config.T3 * ConstUtils.ONE_SECOND, nm.Systembyte, nm);

            //int len = nm.Header.Length + nm.Body.Length + 10;
            //var mb = new MemoryBuffer(len);
            //mb.AppendBeInt(nm.Length);
            //mb.Append(nm.Header);
            //mb.Append(nm.Body);

            //qQ.Enqueue(new QoSent { Arg0 = nm });

            //h.WriteFlush(mb.ToBytes);
        }

        void TcpWriteCtrlMsg(OCSMessage ctrl)
        {
            //var mb = new MemoryBuffer(16);
            //mb.AppendBeInt(10);
            //mb.Append(ctrl.Header);

            //bool skip = ctrl.CtrlLinkTest && Config.HideLogLink;
            //if (!skip)
            //    _OnLog(ctrl.LogFormat());

            //h.WriteFlush(mb.ToBytes);
        }

        void SendCtrlMsg(int stype)
        {
            //Send(new OCSMessage { SType = stype });
        }

        public void Send(OCSMessage msg, int after)
        {
            if (after > 0)
                TimerUtils.Once(after, Send, msg);
            else
                Send(msg);
        }

        public void Send(OCSMessage msg)
        {
            //msg.Id = Config.ID;

            if (!Connected)
            {
                _OnLog("Send fail not connected" + msg.LogFormat());
                return;
            }

            msg.RevID = Config.HostID;
            msg.SendID = Config.ID;


            if ( msg.Kind == eKind.C ) // Control Message 는 페어로 응답이 오지 않는다.
            {
                qqW.Enqueue( msg );
            }

            if (this.quzT3.HasId(msg.GetCheckSum()))
            {
                _OnLog("quzT3 Has ID" + msg.LogFormat());
                return;
            }

            this.quzT3.StartOnce(Config.T3 * ConstUtils.ONE_SECOND, msg.GetCheckSum(), msg);
            //if (msg.CtrlMsg)
            //{
            //    if (msg.CtrlSelectReq || msg.CtrlLinkReq)
            //    {
            //        msg.Systembyte = Interlocked.Increment(ref ctrl_sysbyte);
            //        ChgTcpTimeout(false);//select, linktest req
            //    }
            //}
            //else
            //{
            //    msg.DeviceId = msg.SessID.HasValue ? msg.SessID.Value : Config.DeviceID;

            //    if (msg.IsPrimary && msg.NeedSetSysbyte)
            //        msg.Systembyte = Interlocked.Increment(ref sysbyte);
            //}

            qqW.Enqueue(msg);
        }

        /// <summary>
        /// 응답을 보낼 때 사용.
        /// </summary>
        /// <param name="msg"></param>
        public void Reply(OCSMessage msg)
        {
            if (!Connected)
            {
                _OnLog("Reply fail not connected" + msg.LogFormat());
                return;
            }
            msg.RevID = Config.HostID;
            msg.SendID = Config.ID;

            qqW.Enqueue(msg);
        }

        #endregion

        #region HelpMothed
        void ChgTcpTimeout(bool infinite)
        {
            if (h.Connected)
            {
                if (infinite)
                {
                    if (h.Socket.ReceiveTimeout != Timeout.Infinite)
                        h.ChangeRecvTimeout(Timeout.Infinite);
                }
                else
                {
                    if (Config.LinkOn)
                        h.ChangeRecvTimeout(Config.T6 * ConstUtils.ONE_SECOND);
                }
            }
        }
        #endregion

    }
}