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<string, OCSMessage> quzT3 = new TimerTemplate<string, 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)
            _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( string 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()
        {
            byte stx = 0x0;
            do
            {
                stx = h.ReadByte(); //STX
            }
            while ( stx != 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 );

            if ( ocsMeg.Kind == eKind.M ) //M Command 가변으로 들어 온다.
            {
                ocsMeg.Tag = h.ReadAscii( 4 );
                ocsMeg.SubCode = h.ReadAscii( 3 );
                var viaCount = Convert.ToInt16( h.ReadAscii( 4 ) ); //4byte
                for(int i = 0; i < viaCount; i++ )
                {
                    ocsMeg.ViaRouteList.Add( h.ReadAscii( 4 ) );
                }
                ocsMeg.CheckSum = h.ReadAscii(1);
                h.ReadUntil( ETX );
            }
            else
            {

                ocsMeg.Tag = h.ReadAscii( 4 );
                ocsMeg.SubCode = h.ReadAscii( 3 );

                //CheckSum 을 해야 하나??
                ocsMeg.CheckSum = h.ReadAscii(1);
                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;
            }

            //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 );
                return;
            }

            var checkSum = Convert.ToString( msg.GetCheckSum(), 16 ).ToLower();

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

            this.quzT3.StartOnce( Config.T3 * ConstUtils.ONE_SECOND, checkSum, 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

    }
}