////////////////////////////////////////////////////////////////////////////////
//                                                                             /
// 2012-2019 (c) Baical                                                        /
//                                                                             /
// This library is free software; you can redistribute it and/or               /
// modify it under the terms of the GNU Lesser General Public                  /
// License as published by the Free Software Foundation; either                /
// version 3.0 of the License, or (at your option) any later version.          /
//                                                                             /
// This library is distributed in the hope that it will be useful,             /
// but WITHOUT ANY WARRANTY; without even the implied warranty of              /
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU           /
// Lesser General Public License for more details.                             /
//                                                                             /
// You should have received a copy of the GNU Lesser General Public            /
// License along with this library.                                            /
//                                                                             /
////////////////////////////////////////////////////////////////////////////////
#ifndef BK_PROCESSOR_H
#define BK_PROCESSOR_H

static const GUID g_sTrace_GUID = BK_READER_GUID_TRACE;
static const GUID g_sTelemetry_GUID = BK_READER_GUID_TELEMETRY;

//++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
/// <summary> Processor class </summary>
class CProcessor
        : public Bk::IProcessor
{
    tINT32 volatile m_lReference;
    Bk::IUnknown *m_iBaikal;
    Bk::IBNode *m_iNode;
    CProperty *m_pProperty;
    Bk::IStreams *m_iSubscriber;
    size_t m_szCount;
    CLock m_cLock;
    CMEvent m_cExit_Event; 
    CThShell::tTHREAD m_hThread;
    tBOOL m_bIsThread;

public:
    //++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    CProcessor(Bk::IUnknown *i_iBaikal, Bk::IBNode *i_iNode, CProperty  *i_pProp)
        : m_lReference(1)
        , m_iBaikal(i_iBaikal)
        , m_iNode(i_iNode)
        , m_pProperty(i_pProp)
        , m_iSubscriber(NULL)
        , m_szCount(0)
        , m_bIsThread(FALSE)
    {
        CLock cLock(&m_cLock);
        if (m_iNode)    
        {
            m_iNode->Add_Ref();

            //If you would like to retrieve your dispatcher configuration it is right place to do it using m_iNode interface
            //...
        }

        if (m_pProperty)
        {
            m_pProperty->Add_Ref();
        }

        if (m_iBaikal)
        {
            m_iBaikal->Add_Ref();
        }
    }

    //++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    virtual ~CProcessor()
    {
        m_cLock.Lock();

        //closing processor's thread
        if (m_bIsThread)
        {
            m_cExit_Event.Set(MEVENT_SIGNAL_0);
            if (TRUE == CThShell::Close(m_hThread, 15000))
            {
                m_hThread   = 0;//NULL;
                m_bIsThread = FALSE;
            }
        }

        //saving parameters
        if (m_iNode)    
        {
            // //If you would like to save your processor configuration it is right place to do it using m_iNode interface
            // //accessing processor configuration node example
            // Bk::IBNode *pNode = NULL;
            // m_iNode->AddChildEmpty(L"MySubNode", &pNode);
            // if (pNode)
            // {
            //     pNode->SetAttrText(L"MyParameter", L"MyValue");
            //     pNode->Release();
            //     pNode = NULL;
            // }

            m_iNode->Release();
            m_iNode = NULL;
        }

        if (m_pProperty)
        {
            m_pProperty->Release();
            m_pProperty = NULL;
        }

        //releasing external interfaces
        if (m_iSubscriber)
        {
            m_iSubscriber->Release();
            m_iSubscriber = NULL;
        }

        if (m_iBaikal)
        {
            m_iBaikal->Release();
            m_iBaikal = NULL;
        }

        m_cLock.Unlock();
    }

    //++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    Bk::eResult __stdcall Initialize() 
    {
        CLock cLock(&m_cLock);

        if (m_iBaikal)
        {
            m_iBaikal->Query_Interface(Bk::eInterfaceStreams, (void*&)m_iSubscriber);
        }

        //To process streams new thread has to be created, inside Initialize function should be done only basic things like interfaces 
        //retrieving, all heavy operations should be redirected to separate thread:
        if (m_iSubscriber)
        {
            m_cExit_Event.Init(1, EMEVENT_SINGLE_AUTO);

            if (CThShell::Create(&Static, this, &m_hThread, TM("Proc")))
            {
                m_bIsThread = TRUE;
            }
        }

        return Bk::eOk;
    }


    //++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    tINT32 __stdcall Add_Ref()
    {
        return ATOMIC_INC(&m_lReference);
    }


    //++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    tINT32 __stdcall Release()
    {
        tINT32 lResult = ATOMIC_DEC(&m_lReference);
        if ( 0 >= lResult )
        {
            delete this;
        }

        return lResult;
    }

    //++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    Bk::eResult __stdcall Query_Interface(Bk::eInterface i_eId, void *&o_rUnknown)
    {
        UNUSED_ARG(i_eId);
        UNUSED_ARG(o_rUnknown);
        return Bk::eErrorNotImplemented;
    }

private:
    //++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    //Thread entry point
    static THSHELL_RET_TYPE THSHELL_CALL_TYPE Static(void *i_pArguments)
    {
        CProcessor *pThis = static_cast<CProcessor *>(i_pArguments);
        if (pThis)
        {
            pThis->Process();
        }

        CThShell::Cleanup();
        return THSHELL_RET_OK;
    } 

    //++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
    //Thread routine, all processing will done here:
    void Process()
    {
        tUINT32 dwTimeout = 250;
        Bk::IStorageReader *iReader = NULL;
        Bk::IStreamEx *iExtra = NULL;
        std::list<CStream*> cStreams;
        std::list<CStream*>::iterator cIt = cStreams.begin();
        CStream::eResult eResult;

        while (MEVENT_SIGNAL_0 != m_cExit_Event.Wait(dwTimeout))
        {
            //Checking for new streams using subscriber, new streams has to be filtered for further processing
            if (Bk::eOk == m_iSubscriber->Pull(iReader, iExtra, 0))
            {
                //1. to be able to make decision about processing of stream we need to retrieve some stream information:
                Bk::stStorageInfo stInfo = {};
                iReader->Get_Info(&stInfo);

                //!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
                //2. Let's say we would like to process streams only from "MyApp.exe" executable!
                //   If you want to filter by other criteria please use it here:                !
                //!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
                if (0 == PStrICmp(stInfo.pProcess_Name, TM("MyApp.exe")))
                {
                    //3. Add streams for processing, checking stream type
                    if (0 == memcmp(&stInfo.sType, &g_sTrace_GUID, sizeof(GUID)))
                    {
                        cStreams.push_back(new CTrace(iReader, iExtra));
                    }
                    else if (0 == memcmp(&stInfo.sType, &g_sTelemetry_GUID, sizeof(GUID)))
                    {
                        cStreams.push_back(new CTelemetry(iReader, iExtra));
                    }
                    else
                    {
                        SAFE_RELEASE(iReader);
                        SAFE_RELEASE(iExtra);
                    }
                }
                else //otherwise release the stream
                {
                    SAFE_RELEASE(iReader);
                    SAFE_RELEASE(iExtra);
                }
            }

            dwTimeout = 250; //reset timeout

            //processing
            cIt = cStreams.begin(); 
            while (cIt != cStreams.end())
            {
                eResult = (*cIt)->Process();
                if (CStream::eDone == eResult)
                {
                    //stream processing is finished, stream has to be deleted from the list
                    delete (*cIt);
                    cStreams.erase(cIt++);
                }
                else 
                {
                    if (CStream::eProcessing == eResult)
                    {
                        //if some stream is processing - set timeout to 0
                        dwTimeout = 0;
                    }

                    ++cIt;
                }
            }
        }//while

        //cleanup streams list
        while (!cStreams.empty())
        {
            cIt = cStreams.begin();
            delete (*cIt);
            cStreams.erase(cIt);
        }
    } 
};

#endif //BK_PROCESSOR_H