TnbPacketCommander.h

[詳解]

#pragma once
#include "TnbStrVector.h"
#include "TnbCollectionAdapter.h"
#include "TnbQueueingReporter.h"
#include "TnbSet.h"
 
 
 
#ifndef _TnbDOXYGEN //Document作成用シンボル
  #ifdef _TnbPACKET_LittleEndian
    #pragma message("message : TPacWord/TPacDword is Little Endian.")
  #else
    #pragma message("message : TPacWord/TPacDword is Big Endian.")
  #endif
#endif
 
 
 
namespace TNB 
{
 
 
 
#pragma pack(push)
#pragma pack(1)
 
 
 
struct TPacDword
{
    typedef DWORD TYPE;     
 
    TPacDword& operator=(DWORD v)
    {
        Set(v);
        return *this;
    }
 
    void Set(DWORD v)
    {
        #ifdef _TnbPACKET_LittleEndian
            m_dat[3] = static_cast<BYTE>(v >> 24);
            m_dat[2] = static_cast<BYTE>(v >> 16);
            m_dat[1] = static_cast<BYTE>(v >> 8);
            m_dat[0] = static_cast<BYTE>(v);
        #else
            m_dat[0] = static_cast<BYTE>(v >> 24);
            m_dat[1] = static_cast<BYTE>(v >> 16);
            m_dat[2] = static_cast<BYTE>(v >> 8);
            m_dat[3] = static_cast<BYTE>(v);
        #endif
    }
 
    operator DWORD(void) const
    {
        return Get();
    }
 
    DWORD Get(void) const
    {
        DWORD r;
        #ifdef _TnbPACKET_LittleEndian
            r = (m_dat[3] << 24) + (m_dat[2] << 16) + (m_dat[1] << 8) + (m_dat[0]);
        #else
            r = (m_dat[0] << 24) + (m_dat[1] << 16) + (m_dat[2] << 8) + (m_dat[3]);
        #endif
        return r;
    }
 
    CStr ToString(void) const
    {
        return CStr::DwordToString(Get());
    }
 
private:
    BYTE m_dat[4];              
};
 
 
 
struct TPacWord
{
    typedef WORD TYPE;      
 
    TPacWord& operator=(WORD v)
    {
        Set(v);
        return *this;
    }
 
    void Set(WORD v)
    {
        #ifdef _TnbPACKET_LittleEndian
            m_dat[1] = static_cast<BYTE>(v >> 8);
            m_dat[0] = static_cast<BYTE>(v);
        #else
            m_dat[0] = static_cast<BYTE>(v >> 8);
            m_dat[1] = static_cast<BYTE>(v);
        #endif
    }
 
    operator WORD(void) const
    {
        return Get();
    }
 
    WORD Get(void) const
    {
        WORD r;
        #ifdef _TnbPACKET_LittleEndian
            r = static_cast<WORD>((m_dat[1] << 8) + (m_dat[0]));
        #else
            r = static_cast<WORD>((m_dat[0] << 8) + (m_dat[1]));
        #endif
        return r;
    }
 
    CStr ToString(void) const
    {
        return CStr::DwordToString(Get());
    }
 
private:
    BYTE m_dat[2];              
};
 
 
 
struct TPacByte
{
    typedef BYTE TYPE;      
 
    TPacByte& operator=(BYTE v)
    {
        m_dat = v;
        return *this;
    }
 
    void Set(BYTE v)
    {
        m_dat = v;
    }
 
    operator BYTE(void) const
    {
        return Get();
    }
 
    BYTE Get(void) const
    {
        return m_dat;
    }
 
    CStr ToString(void) const
    {
        return CStr::DwordToString(Get());
    }
 
    BYTE& Ref(void)
    {
        return m_dat;
    }
 
private:
    BYTE m_dat;         
};
 
 
 
#pragma pack(pop)
 
 
 
struct IPacketCommandExecuter
{
    struct IPacket
    {
        virtual ~IPacket(void) {}
 
        virtual CByteVector Serialize(void) const = 0;
 
        virtual int Deserialize(const IConstCollectionT<BYTE>& c) = 0;
 
        virtual CStrVector ToStrings(void) const = 0;
    };
 
    virtual ~IPacketCommandExecuter(void) {}
 
    virtual int SendPacket(const IPacket& packet) = 0;
 
    virtual int ExecuteCommand(IPacket& _response, const IPacket& command, DWORD timeout = 500) = 0;
};
 
 
 
class CPacketCommander : public IPacketCommandExecuter, CCopyImpossible
{
public:
 
    enum EFailure
    {
        Timeout         = -1,   
        SendError       = -2,   
        Canceled        = -3,   
        NotOpen         = -4,   
    };
 
    struct IListener
    {
        virtual ~IListener(void) {}
 
        virtual void OnCommanderReceivedPacket(const IPacket* pPacket, int result) = 0;
 
        virtual void OnCommanderReceivedExpectationPacket(const IPacket* pPacket) {}
 
        virtual void OnCommanderDustyData(BYTE b, int result) = 0;
 
        virtual bool OnReportEvent(EReportEvent event) = 0;
    };
 
 
    //-------------------------------------
 
 
    CPacketCommander(int iQueueSize = 10000) : m_pRep(NULL), m_queue(iQueueSize)
    {
    }
 
    ~CPacketCommander(void)
    {
        CPacketCommander::Close();
    }
 
    void SetPacketFormat(IPacketCommandExecuter::IPacket* pPacket)
    {
        m_in.m_pPacketFormat = pPacket;
    }
 
    void _deprecated SetBasicPacket(IPacketCommandExecuter::IPacket* pPacket)
    {
        SetPacketFormat(pPacket);
    }
 
    void SetListener(IListener* pListener)
    {
        m_in.m_pListener = pListener;
    }
 
    void SetAliveTime(DWORD time = DWORD_MAX)
    {
        m_queue.SetAliveTime(time);
    }
 
    void SetTearminalTime(DWORD time = DWORD_MAX)
    {
        m_queue.SetTearminalTime(time);
    }
    
    bool IsOpened(void) const
    {
        if ( m_pRep != NULL )
        {
            return m_pRep->IsAlive();
        }
        return false;
    }
    
    bool IsConnect(void) const
    {
        if ( m_pRep != NULL )
        {
            return m_pRep->IsConnect();
        }
        return false;
    }
 
    bool Open(IReport* pRep, bool withQueueing = false)
    {
        CPacketCommander::Close();
        if ( pRep != NULL && m_in.m_pPacketFormat != NULL )
        {
            m_pRep = pRep;
            if ( withQueueing )
            {
                m_queue.SetEnvironment(m_pRep, &m_in);
            }
            else
            {
                m_pRep->SetListener(&m_in);
            }
            return m_pRep->Start();
        }
        return false;
    }
 
    virtual void Close(void)
    {
        if ( m_pRep != NULL )
        {
            m_pRep->Stop();
        }
        m_queue.ResetEnvironment();
        m_pRep = NULL;
    }
 
    virtual int SendPacket(const IPacketCommandExecuter::IPacket& packet)
    {
        if ( IsOpened() )
        {
            ASSERT( m_pRep != NULL );
            // 送信コマンドデータ作成
            CByteVector vb = packet.Serialize();
            // 受信処理ロック
            m_in.m_sendSync.Lock();
            // 送信
            bool r = m_pRep->Send(vb.GetSize(), vb.ReferBuffer());
            // 受信処理アンロック
            m_in.m_sendSync.Unlock();
            if ( r )
            {
                return 0; //成功
            }
            return SendError; //送信エラー
        }
        return NotOpen;
    }
 
    int _deprecated SendCommand(const IPacketCommandExecuter::IPacket& packet)
    {
        return SendPacket(packet);
    }
 
    virtual int ExecuteCommand(IPacketCommandExecuter::IPacket& _response, const IPacketCommandExecuter::IPacket& command, DWORD timeout = 500)
    {
        int r = ExecuteCommand_Start(_response, command);
        if ( r >= 0 )
        {
            r = ExecuteCommand_Wait(_response, timeout);
            ExecuteCommand_Stop(_response);
        }
        return r;
    }
 
    int ExecuteCommand_Start(IPacketCommandExecuter::IPacket& _response,const IPacketCommandExecuter::IPacket& command)
    {
        if ( IsOpened() )
        {
            ASSERT( m_pRep != NULL );
            //== 準備
            bool r = true;
            // 送信コマンドデータ作成
            CByteVector vb = command.Serialize();
            // 新タスク
            m_in.NewTask(&_response);
            //== 送信
            if ( ! vb.IsEmpty() )
            {
                // 受信処理ロック
                m_in.m_sendSync.Lock();
                // 送信
                r = m_pRep->Send(vb.GetSize(), vb.ReferBuffer());
                // 受信処理アンロック
                m_in.m_sendSync.Unlock();
            }
            return r ? 1 : SendError;
        }
        return NotOpen;
    }
 
    int ExecuteCommand_Wait(IPacketCommandExecuter::IPacket& _response, DWORD timeout = 500)
    {
        if ( IsOpened() )
        {
            // タスク
            CInner::TTask* pTask = m_in.FindTask(&_response);
            if ( pTask != NULL )
            {
                DWORD to = timeout;
                while ( true )
                {
                    if ( pTask->recvEvent.Lock(to) )
                    {
                        int r = pTask->analyzedCode;
                        m_in.EndTask(pTask);
                        return r;
                    }
                    if ( ! pTask->isTimeoutExtend )
                    {
                        break;
                    }
                    pTask->isTimeoutExtend = false; //タイムアウト延長
                    if ( to > 0 )
                    {
                        DWORD t = pTask->extendTime;
                        if ( t == 0 )
                        {
                            t = timeout;
                        }
                        DWORD d = pTask->tick.GetPassedCount();
                        if ( d > t )
                        {
                            to = 0;
                        }
                        else
                        {
                            to = t - d;
                        }
                    }
                }
                return Timeout; //タイムアウト
            }
            return SendError; //送信エラー(Startしてない)
        }
        return NotOpen;
    }
 
    void ExecuteCommand_Stop(IPacketCommandExecuter::IPacket& response)
    {
        CInner::TTask* pTask = m_in.FindTask(&response);
        if ( pTask != NULL )
        {
            m_in.EndTask(pTask);
        }
    }
 
    void ExtendTimeout(DWORD threadId = 0)
    {
        m_in.ExtendTimeout(0, threadId);
    }
 
    void ExtendTimeoutEx(DWORD time, DWORD threadId = 0)
    {
        m_in.ExtendTimeout(time, threadId);
    }
 
    void CancelWaitResponse(DWORD threadId = 0)
    {
        m_in.CancelWaitResponse(threadId);
    }
 
    bool SendRawData(size_t size, const BYTE* pData)
    {
        bool r = false;
        if ( m_pRep != NULL )
        {
            // 受信処理ロック
            m_in.m_sendSync.Lock();
            // 送信
            r = m_pRep->Send(size, pData);
            // 受信処理アンロック
            m_in.m_sendSync.Unlock();
        }
        return r;
    }
 
    size_t GetQueueFreeSize(void) const
    {
        return m_queue.GetQueueFreeSize();
    }
 
    void PurgeQueue(void)
    {
        m_queue.Purge();
    }
 
    IReport* ReferReporter(void)
    {
        return IsOpened() ? m_pRep : NULL;
    }
 
private:
 
    class CInner : public IReport::IListener, public CQueueingReporter::IListener
    {
        DEFPARENTLISTENER(CPacketCommander, IParentListener);
    public:
        struct TTask
        {
            DWORD               threadId;               
            CSyncEvent          recvEvent;              
            bool                isAnalyzed;             
            IPacket*            pExpectationPacket;     
            bool                isTimeoutExtend;        
            DWORD               extendTime;             
            CTickCount          tick;                   
            int                 analyzedCode;           
            TTask(void) : recvEvent(false, false), pExpectationPacket(NULL), isTimeoutExtend(false), extendTime(0), analyzedCode(0), isAnalyzed(true)
            {}
        };
 
        //-----
 
        CInner(void) : m_pListener(NULL), m_pPacketFormat(NULL)
        {
        }
        void EndTask(TTask* p)
        {
            EXCLUSIVE( &m_task );
            if ( m_task.Delete(p) )
            {
//              p->recvEvent.Unlock();
                delete p;
            }
        }
        TTask* FindTask(IPacket* pExpectation)
        {
            loop ( i, m_task )
            {
                TTask* p = m_task.At(i);
                if ( p->pExpectationPacket == pExpectation )
                {
                    return p;
                }
            }
            return NULL;
        }
        TTask* NewTask(IPacket* pExpectation)
        {
            TTask* p = new TTask;
            p->threadId = ::GetCurrentThreadId();
            p->pExpectationPacket = pExpectation;
            p->isAnalyzed = false;
            p->recvEvent.ToLock();
            p->isTimeoutExtend = false;
            p->extendTime = 0;
            ASSERT( ! m_task.Has(p) );
            m_task.InsertEx(p);
            return p;
        }
        void ExtendTimeout(DWORD time, DWORD threadId = 0)
        {
            if ( threadId == 0 )
            {
                loop ( i, m_task )
                {
                    TTask* p = m_task.At(i);
                    p->tick.Reset();
                    p->isTimeoutExtend = true;
                    p->extendTime = time;
                }
            }
            else
            {
                loop ( i, m_task )
                {
                    TTask* p = m_task.At(i);
                    if ( threadId == p->threadId )
                    {
                        p->tick.Reset();
                        p->isTimeoutExtend = true;
                        p->extendTime = time;
                        break;
                    }
                }
            }
        }
        void CancelWaitResponse(DWORD threadId = 0)
        {
            if ( threadId == 0 )
            {
                loop ( i, m_task )
                {
                    TTask* p = m_task.At(i);
                    p->analyzedCode = Canceled;
//                  p->pExpectationPacket = NULL;
                    p->isAnalyzed = true;
                    p->recvEvent.Unlock();
                }
            }
            else
            {
                loop ( i, m_task )
                {
                    TTask* p = m_task.At(i);
                    if ( threadId == p->threadId )
                    {
                        p->analyzedCode = Canceled;
//                      p->pExpectationPacket = NULL;
                        p->isAnalyzed = true;
                        p->recvEvent.Unlock();
                        break;
                    }
                }
            }
        }
        virtual bool OnReportEvent(const CReportEvent& ev)
        {
            if ( ev.HasEvent() )
            {
                if ( ev.GetEvent() == ReportEvent_End )
                {
                    CancelWaitResponse();
                }
                if ( m_pListener != NULL )
                {
                    return m_pListener->OnReportEvent(ev.GetEvent());
                }
                return true;
            }
//          EXCLUSIVE( &m_sendSync );
            CConstAdapterT<BYTE> c(ev.GetDataSize(), ev.GetData());
            if ( Analyze(c) == 0 )
            {
                if ( m_pListener != NULL )
                {
                    loop ( i, ev.GetDataSize() )
                    {
                        m_pListener->OnCommanderDustyData(ev.GetData()[i], Timeout);
                    }
                }
            }
            return true;
        }
        virtual size_t OnReportData(bool boIsLast, const IConstCollectionT<BYTE>& c)
        {
//          EXCLUSIVE( &m_sendSync );
            size_t r = Analyze(c);
            if ( boIsLast && r == 0 )
            {
                r = c.GetSize();
                if ( m_pListener != NULL )
                {
                    loop ( i, r )
                    {
                        m_pListener->OnCommanderDustyData(c.At(i), Timeout);
                    }
                }
            }
            return r;
        }
        virtual void OnReportDustyData(BYTE b)
        {
            if ( m_pListener != NULL )
            {
                m_pListener->OnCommanderDustyData(b, Timeout);
            }
        }
        size_t Analyze(const IConstCollectionT<BYTE>& c)
        {
            size_t size = 0;
            if ( m_pPacketFormat != NULL )
            {
                int a = m_pPacketFormat->Deserialize(c);
                if ( a > 0 )
                {
                    size = a;
                    a = 0;
                    bool isDisp = false;
                    {
                        EXCLUSIVE( &m_task );
                        loop ( i, m_task )
                        {
                            TTask* p = m_task.At(i);
//                          if ( p->pExpectationPacket != NULL )
                            if ( ! p->isAnalyzed )
                            {
                                a = p->pExpectationPacket->Deserialize(c);
                                if ( a > 0 )
                                {
                                    if ( m_pListener != NULL )
                                    {
                                        m_pListener->OnCommanderReceivedExpectationPacket(m_pPacketFormat);
                                    }
                                    p->analyzedCode = a;
//                                  p->pExpectationPacket = NULL;
                                    p->isAnalyzed = true;
                                    p->recvEvent.Unlock();
                                    isDisp = true;
                                    break;
                                }
                            }
                        }
                    }
                    if ( ! isDisp && m_pListener != NULL )
                    {
                        m_pListener->OnCommanderReceivedPacket(m_pPacketFormat, a);
                    }
                }
                else if ( a == 0 )
                {
                }
                else
                {
                    size = 1;
                    if ( m_pListener != NULL )
                    {
                        m_pListener->OnCommanderDustyData(c.At(0), a);
                    }
                }
            }
            return size;
        }
        CSyncSection            m_sendSync;         
        IParentListener*        m_pListener;        
        IPacket*                m_pPacketFormat;    
        CSingleSetT<TTask*>     m_task;             
    };
    CInner              m_in;               
    IReport*            m_pRep;             
    CQueueingReporter   m_queue;            
};
 
 
 
#ifndef _TnbDOXYGEN //Document作成用シンボル
 C_ASSERT ( sizeof(TPacDword) == 4 ); 
 C_ASSERT ( sizeof(TPacWord) == 2 ); 
 C_ASSERT ( sizeof(TPacByte) == 1 ); 
#endif
 
 
 
}; // TNB