////////////////////////////////////////////////////////////////////////////////
//                                                                             /
// 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.                                            /
//                                                                             /
////////////////////////////////////////////////////////////////////////////////
#include "common.h"

#define TIMEOUT_2HOURS (3600u*2u*1000u)

#if !defined(MIN)
    #define MIN(X,Y) ((X) < (Y) ? (X) : (Y))
#endif

static const GUID g_pType_GUID[] =
{
    // TYPE : eP7User_Type::EP7USER_TYPE_TRACE
    BK_GUID_STREAM_TRACE,

    // TYPE : eP7User_Type::EP7USER_TYPE_TELEMETRY
    BK_GUID_STREAM_TELEMETRY_V1,

    BK_GUID_STREAM_TELEMETRY_V2,

    //For feature needs
    // {24CE9629-5339-4da7-B11F-3BD871AB7880}
    //{   0x24ce9629, 0x5339, 0x4da7, 
    //    { 0xb1, 0x1f, 0x3b, 0xd8, 0x71, 0xab, 0x78, 0x80 } 
    //}
};


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CTail::CTail(CBList<CTail*> *i_pOwner, size_t i_szTail)
    : pOwner(i_pOwner)
    , pTail((tUINT8*)malloc(i_szTail))
    , szTail(i_szTail)
    , szTailRest(0)
    , szTailOffs(0)
    , pNext(NULL)
{

}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CTail::~CTail()
{
    if (pTail)
    {
        free(pTail);
        pTail = NULL;
    }

    szTailRest = 0;
    szTailOffs = 0;
    szTail     = 0;
    pNext      = NULL;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CTail::Realloc(size_t i_szNewSize)
{
    if (szTail > i_szNewSize)
    {
        return;
    }

    i_szNewSize = (i_szNewSize + 1023) & (~1023u);

    tUINT8 *l_pNew = (tUINT8 *)realloc(pTail, i_szNewSize);

    if (!l_pNew)
    {
        l_pNew = (tUINT8 *)malloc(i_szNewSize);
        if (    (l_pNew)
             && (pTail)
           )
        {
            memcpy(l_pNew, pTail, szTail);
        }
        
        if (pTail)
        {
            free(pTail);
            pTail = NULL;
        }
    }

    pTail = l_pNew;
    szTail = pTail ? i_szNewSize : 0;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CTail::Release()
{
    pNext      = 0;
    szTailOffs = 0;
    szTailRest = 0;

    if (pOwner)
    {
        pOwner->Add_After(NULL, this);
    }
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CStreamP7::CStreamP7()
    : m_iStorare(NULL)
    , m_pStreamEx(NULL)
    , m_pFirst(NULL)
    , m_pLast(NULL)
    , m_bError(FALSE)
    , m_pTail(NULL)
{
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CStreamP7::~CStreamP7()
{
    Uninit();
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CStreamP7::Uninit()
{
    if (m_pStreamEx)
    {
        m_pStreamEx->Deactivate();
        m_pStreamEx->Release();
        m_pStreamEx = NULL;
    }

    if (m_iStorare)
    {
        m_iStorare->Uninit_Writer();
        m_iStorare->Release();
        m_iStorare = NULL;
    }

    m_bError  = FALSE;

    ReleaseTails();
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CStreamP7::PushTail(CTail *i_pTail)
{
    i_pTail->szTailOffs = 0;
    i_pTail->szTailRest = 0;
    i_pTail->pNext      = m_pTail;
    m_pTail             = i_pTail;
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CStreamP7::ReleaseTails()
{
    while (m_pTail)
    {
        CTail *l_pTail = m_pTail;
        m_pTail = m_pTail->pNext;

        l_pTail->Release();
    }
}



////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CDataBlock::CDataBlock(size_t i_szCount)
{
    m_szCount = i_szCount;

    if (i_szCount)
    {
        tUINT8 *l_pData = (tUINT8*)malloc(m_szCount * sizeof(Bk::stDataChunk));

        if (l_pData)
        {
            m_pData = l_pData;
            memset(l_pData, 0, m_szCount * sizeof(Bk::stDataChunk));

            m_pFirst = (Bk::stDataChunk*)l_pData;
            m_pLast  = m_pFirst;

            while (--i_szCount)
            {
                m_pLast = (Bk::stDataChunk*)l_pData;
                l_pData += sizeof(Bk::stDataChunk);
                m_pLast->pNext = (Bk::stDataChunk*)(l_pData);
            }
            m_pLast->pNext = NULL;
        }
    }
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CDataBlock::~CDataBlock()
{
    if (m_pData)
    {
        free(m_pData);
        m_pData = NULL;
    }

    m_szCount = 0;
}

GASSERT(CConnectionP7::eThreadCountSignals == 1);
//"m_cExit_Event initialization if wrong");
//Update if (FALSE == m_cExit_Event.Init(eThreadCountSignals, EMEVENT_SINGLE_AUTO))

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CConnectionP7::CConnectionP7(Bk::ICore        *i_pCore,
                             IP7_Server       *i_pServer,
                             IP7S_Address     *i_pAddress,
                             IP7S_Connnection *i_pConnection,
                             CProperty        *i_pPropRoot,
                             CDNS_Resolver    *i_pDns_Resolver,
                             IP7_Trace        *i_iP7_Trace,
                             IP7_Telemetry    *i_iP7_Tel
                            )
    : m_pServer(i_pServer)
    , m_pAddress(i_pAddress)
    , m_pConnection(i_pConnection)
    , m_pPropRoot(i_pPropRoot)
    , m_pDns_Resolver(i_pDns_Resolver)
    , m_iP7_Trace(i_iP7_Trace)
    , m_hP7_TraceModule(NULL)
    , m_iP7_Tel(i_iP7_Tel)
    , m_cExit_Event()
    , m_hThread(0)
    , m_bIsThread(TRUE)
    , m_bActive(TRUE)
    , m_pCore(i_pCore)
    , m_bBigEndian(i_pConnection->Is_BigEndian())
    , m_pFirstChunk(NULL)
{
    memset(m_pChannels, 0, sizeof(m_pChannels));
     
    if (m_iP7_Trace)
    {
        m_iP7_Trace->Register_Module(TM("P7PC"), &m_hP7_TraceModule);
    }

    sP7S_Connection_Info l_sConInfo = { 0 };
    tXCHAR l_pName[256];
    tXCHAR l_pIp[256];

    m_pConnection->Get_Info(&l_sConInfo);

    if (FALSE == m_pDns_Resolver->Get_Name((sockaddr*)&l_sConInfo.sAddress, l_pName, LENGTH(l_pName), NULL))
    {
        l_pName[0] = TM('?');
        l_pName[1] = 0;
    }

    if (FALSE == Print_SAddr((sockaddr*)&l_sConInfo.sAddress, l_pIp, LENGTH(l_pIp)))
    {
        l_pIp[0] = TM('?');
        l_pIp[1] = 0;
    }

    LOG_INFO(TM("[0x%08p] Create connection [Id=%X] with {%s}{%s}/{%s}"), this, m_pConnection->Get_ID(), l_pName, l_pIp, l_sConInfo.pProcess_Name);

    if (FALSE == m_cExit_Event.Init(eThreadCountSignals, EMEVENT_SINGLE_AUTO))
    {
        LOG_ERROR(TM("[0x%08p] Exit event wasn't created !"), this);
    }

    if (FALSE == CThShell::Create(&Static_Routine, this, &m_hThread, TM("Baical")))
    {
        m_bIsThread = FALSE;
        LOG_ERROR(TM("[0x%08p] Thread wasn't created !"), this);
    }
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
CConnectionP7::~CConnectionP7()
{
    if (m_bIsThread)
    {
        m_cExit_Event.Set(eThreadExit);
        if (TRUE == CThShell::Close(m_hThread, 10000))
        {
            m_hThread = 0;//NULL;
            m_bIsThread = FALSE;
        }
        else
        {
            LOG_ERROR(TM("[0x%08p] stat thread timeout expire !"), this);
        }
    }
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
tBOOL CConnectionP7::Get_Active()
{
    CLock l_cLock(&m_cLock);
    return m_bActive;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
tBOOL CConnectionP7::Is_BigEndian()
{
    CLock l_cLock(&m_cLock);

    if (!m_pConnection)
    {
        return FALSE;
    }

    return m_bBigEndian;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Bk::eResult CConnectionP7::Get_Status(Bk::stStreamStatus *o_pStatus)
{
    CLock l_cLock(&m_cLock);

    Bk::eResult            l_eReturn  = Bk::eOk;
    sP7S_Connection_Status l_sStatus  = {0};

    m_pConnection->Get_Status(&l_sStatus);

    if (l_sStatus.bClosed)
    {
        if (FALSE == l_sStatus.bHas_Data)
        {
            o_pStatus->eConnection = Bk::eStreamConClosed;
        }
        else //even if channel is closed but we still have data
        {
            o_pStatus->eConnection = Bk::eStreamConOnline;
        }
    }
    else if (l_sStatus.bConnected)
    {
        o_pStatus->eConnection = Bk::eStreamConOnline;
    }
    else
    {
        o_pStatus->eConnection = Bk::eStreamConOffline;
    }
    
    o_pStatus->dwDown_Time = l_sStatus.dwDuration_Off;

    return l_eReturn;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Bk::eResult CConnectionP7::Put_Packet(tUINT32 i_uChannel, const tUINT8 *i_pBuffer, tUINT32 i_dwSize)
{
    CLock l_cLock(&m_cLock);

    if (!m_pConnection)
    {
        return Bk::eErrorNotActive;
    }

    if (FALSE == m_pConnection->Push_Data(i_uChannel, i_pBuffer, i_dwSize))
    {
        return Bk::eErrorNoBuffer;
    }

    return Bk::eOk;
}
     

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Bk::stDataChunk *CConnectionP7::Pull_Free_Chunk()
{
    Bk::stDataChunk *l_pReturn = NULL;

    if (m_pFirstChunk)
    {
        l_pReturn = m_pFirstChunk;
        m_pFirstChunk = m_pFirstChunk->pNext;
    }
    else
    {
        CDataBlock *l_pDataBlock = new CDataBlock();
        m_pFirstChunk = l_pDataBlock->Get_First();
        m_cDataBlocks.Add_After(m_cDataBlocks.Get_Last(), l_pDataBlock);
        l_pReturn = m_pFirstChunk;
        m_pFirstChunk = m_pFirstChunk->pNext;
    }

    return l_pReturn;
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CConnectionP7::Push_Free_Chunks(Bk::stDataChunk *i_pFirst)
{
    if (i_pFirst)
    {
        i_pFirst->pNext = m_pFirstChunk;
        m_pFirstChunk   = i_pFirst;
    }
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
tBOOL CConnectionP7::CreateStorage(CStreamP7 *i_pStream, size_t i_szIndex)
{
    sP7Ext_Header        l_stHdrExt   = {}; 
    sP7Ext_Raw          *l_pHdrExtRaw = (sP7Ext_Raw*)&l_stHdrExt;
    eP7User_Type         l_eUserType  = EP7USER_TYPE_MAX;
    sP7S_Connection_Info l_sConInfo   = { 0 };
    Bk::stStreamInfo     l_sStrInfo   = { 0 };

    l_stHdrExt = *(sP7Ext_Header*)(i_pStream->m_pFirst->pBuffer);

    if (m_bBigEndian)
    {
        l_pHdrExtRaw->dwBits = ntohl(l_pHdrExtRaw->dwBits);
    }

    l_eUserType = (eP7User_Type)l_stHdrExt.dwType;


    if (l_eUserType >= LENGTH(g_pType_GUID))
    {
        LOG_ERROR(TM("[0x%08p] Unexpected user type %d"), this, l_eUserType);
        return FALSE;
    }

    m_pConnection->Get_Info(&l_sConInfo);

    l_sStrInfo.dwProcess_ID      = l_sConInfo.dwProcess_ID;
    l_sStrInfo.wProtocol_Version = l_sConInfo.wProtocol_Version;
    l_sStrInfo.bIs_BigEndian     = m_pConnection->Is_BigEndian();

    //Fill process name
    PStrCpy(l_sStrInfo.pProcess_Name, LENGTH(l_sStrInfo.pProcess_Name), l_sConInfo.pProcess_Name);

    //Fill channel type GUID  
    memcpy(&l_sStrInfo.sType_GUID, &g_pType_GUID[l_eUserType], sizeof(GUID));

    l_sStrInfo.sProcess_Time.dwHighDateTime = l_sConInfo.dwProcess_Time_Hi;
    l_sStrInfo.sProcess_Time.dwLowDateTime = l_sConInfo.dwProcess_Time_Low;

    //Fill remote address information (IPv4/6)
    if (sizeof(l_sStrInfo.sAddress) == sizeof(l_sConInfo.sAddress))
    {
        memcpy(&l_sStrInfo.sAddress,
               &l_sConInfo.sAddress,
               sizeof(l_sConInfo.sAddress)
              );
    }
    else
    {
        LOG_ERROR(TM("[0x%08p] Network address structure size is different"), this);
        return FALSE;
    }

    ////////////////////////////////////////////////////////////////////////////
    //Format node name
    if (FALSE == m_pDns_Resolver->Get_Name((sockaddr*)&l_sConInfo.sAddress,
                                           l_sStrInfo.pNode,
                                           LENGTH(l_sStrInfo.pNode),
                                           NULL
                                          )
       )
    {
        sockaddr_storage l_sAddress = { 0 };
        memcpy(&l_sAddress, &l_sConInfo.sAddress, sizeof(l_sConInfo.sAddress));

        if (AF_INET == l_sAddress.ss_family)
        {
            ((sockaddr_in*)&l_sAddress)->sin_port = 0;
        }
        else if (AF_INET6 == l_sAddress.ss_family)
        {
            ((sockaddr_in6*)&l_sAddress)->sin6_port = 0;
        }

        if (FALSE == Print_SAddr((sockaddr*)&l_sAddress,
                                 l_sStrInfo.pNode,
                                 LENGTH(l_sStrInfo.pNode)
                                )
           )
        {
            PStrCpy(l_sStrInfo.pNode, LENGTH(l_sStrInfo.pNode), TM("No Name"));
        }
        else
        {
            //replace IPv6 delimiters
            tXCHAR *l_pReplace = l_sStrInfo.pNode;
            while (*l_pReplace)
            {
                if (L':' == (*l_pReplace))
                {
                    *l_pReplace = TM('x');
                }

                l_pReplace++;
            }
        }
    }

    if (Bk::eOk != m_pCore->Create_Storage(&l_sStrInfo, i_pStream->m_iStorare))
    {
        LOG_ERROR(TM("[0x%08p] Create storage failed"), this);
        return FALSE;
    }

    if (EP7USER_TYPE_TRACE == l_eUserType)
    {
        i_pStream->m_pStreamEx = new CP7Stream_Ex_Trace((tUINT32)i_szIndex, this, m_pPropRoot);
    }
    else if (EP7USER_TYPE_TELEMETRY_V1 == l_eUserType)
    {
        i_pStream->m_pStreamEx = new CP7Stream_Ex_Telemetry((tUINT32)i_szIndex, 
                                                            this, 
                                                            m_pPropRoot, 
                                                            CP7Stream_Ex_Telemetry::eProtocol_v1);
    }
    else if (EP7USER_TYPE_TELEMETRY_V2 == l_eUserType)
    {
        i_pStream->m_pStreamEx = new CP7Stream_Ex_Telemetry((tUINT32)i_szIndex, 
                                                            this, 
                                                            m_pPropRoot, 
                                                            CP7Stream_Ex_Telemetry::eProtocol_v2);
    }

    i_pStream->m_iStorare->Put_Stream_Ex(i_pStream->m_pStreamEx);

    return TRUE;
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CConnectionP7::Register_Streams()
{
    pAList_Cell l_pEl = NULL;
    while ((l_pEl = m_cTmpStreams.Get_Next(l_pEl)))
    {
        Bk::stStorageInfo l_stInfo;
        CStreamP7 *l_pStream = m_cTmpStreams.Get_Data(l_pEl);
        if (    (l_pStream)
             && (l_pStream->m_iStorare)
           )
        {
            if (Bk::eOk == l_pStream->m_iStorare->Get_Info(&l_stInfo))
            {
                Bk::IStorageReader *l_pReader = NULL;
                if (Bk::eOk == l_pStream->m_iStorare->Query_Interface(Bk::eInterfaceReader, (void*&)l_pReader))
                {
                    m_pCore->Register_New_Stream(NULL, l_pReader, l_pStream->m_pStreamEx);
                    l_pReader->Release();
                }
                else
                {
                    LOG_ERROR(TM("[0x%08p] Can't retrieve IStorageReader interface"), this);
                }

                pAList_Cell l_pElPrev = m_cTmpStreams.Get_Prev(l_pEl);
                m_cTmpStreams.Del(l_pEl, FALSE);
                l_pEl = l_pElPrev;
            }
        }
        else
        {
            LOG_ERROR(TM("[0x%08p] Stream list is corrupted!"), this);
            pAList_Cell l_pElPrev = m_cTmpStreams.Get_Prev(l_pEl);
            m_cTmpStreams.Del(l_pEl, FALSE);
            l_pEl = l_pElPrev;
        }
    }
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CConnectionP7::Register_Stream(CStreamP7 *i_pStream)
{
    pAList_Cell l_pEl = NULL;
    while ((l_pEl = m_cTmpStreams.Get_Next(l_pEl)))
    {
        Bk::stStorageInfo l_stInfo;
        CStreamP7 *l_pStream = m_cTmpStreams.Get_Data(l_pEl);
        if (    (l_pStream)
             && (l_pStream->m_iStorare)
           )
        {
            if (    (i_pStream == l_pStream)
                 && (Bk::eOk == l_pStream->m_iStorare->Get_Info(&l_stInfo))
               )
            {
                Bk::IStorageReader *l_pReader = NULL;
                if (Bk::eOk == l_pStream->m_iStorare->Query_Interface(Bk::eInterfaceReader, (void*&)l_pReader))
                {
                    m_pCore->Register_New_Stream(NULL, l_pReader, l_pStream->m_pStreamEx);
                    l_pReader->Release();
                }
                else
                {
                    LOG_ERROR(TM("[0x%08p] Can't retrieve IStorageReader interface"), this);
                }

                pAList_Cell l_pElPrev = m_cTmpStreams.Get_Prev(l_pEl);
                m_cTmpStreams.Del(l_pEl, FALSE);
                l_pEl = l_pElPrev;
            }
        }
        else
        {
            LOG_ERROR(TM("[0x%08p] Stream list is corrupted!"), this);
            pAList_Cell l_pElPrev = m_cTmpStreams.Get_Prev(l_pEl);
            m_cTmpStreams.Del(l_pEl, FALSE);
            l_pEl = l_pElPrev;
        }
    }
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CConnectionP7::Zero_Stream_Buffers(CStreamP7 *i_pStream)
{
    Bk::stDataChunk *l_pIter = i_pStream ? i_pStream->m_pFirst : NULL;
    while (l_pIter)
    {
        l_pIter->szBuffer = 0;
        l_pIter->pBuffer  = 0;
        l_pIter = l_pIter->pNext;
    }
}


////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
void CConnectionP7::Routine()
{
    tUINT32                l_uSleep       = 0;
    tBOOL                  l_bExit        = FALSE;
    tUINT32                l_uWait        = 0;
    sH_User_Data           l_stHdrUser    = {}; 
    sH_User_Raw           *l_stHdrUserRaw = (sH_User_Raw*)&l_stHdrUser;
    Bk::stDataChunk       *l_pDataChunk   = NULL;
    CStreamP7             *l_pStream      = NULL;
    CTail*                 l_pTailCurrent = NULL;
    sP7S_Connection_Status l_sStatus      = { 0 };
    CTPData                l_cDataPacket;
    CBList<CTail*>         l_cTails;

    l_cTails.Add_After(NULL, new CTail(&l_cTails, 4096));
    

    if (m_iP7_Trace)
    {
        m_iP7_Trace->Register_Thread(TM("ConP7"), 0);
    }

    LOG_INFO(TM("[0x%08p] Connection start thread"), this);

    while (!l_bExit)
    {
        l_uWait = m_cExit_Event.Wait(l_uSleep);
        if (MEVENT_TIME_OUT == l_uWait)
        {
            //maintain connection, close if necessary///////////////////////////////////////////////////////////////////
            m_pConnection->Get_Status(&l_sStatus);

            if (    (    (l_sStatus.bClosed)
                      || (l_sStatus.dwDuration_Off > TIMEOUT_2HOURS)
                    )
                 && (FALSE == l_sStatus.bHas_Data)
               )
            {
                LOG_WARNING(TM("[0x%08p] Connection is closed"), this);
                CLock l_cLock(&m_cLock);
                m_bActive = FALSE;
                l_bExit   = TRUE;
                continue;
            }
            else if (l_sStatus.bReinitialized)
            {
                LOG_WARNING(TM("[0x%08p] Connection has been reinitialized"), this);
                if (l_pTailCurrent)
                {
                    l_pTailCurrent->Release();
                    l_pTailCurrent = NULL;
                }
            }


            //parse incoming packet/////////////////////////////////////////////////////////////////////////////////////
            CTPacket *l_pPacket = m_pConnection->Pull_Packet();

            if (!l_pPacket)
            {
                l_uSleep = 50;
                continue;
            }

            l_uSleep = 0;

            l_cDataPacket.Attach(l_pPacket);

            tUINT8 *l_pBuffer = l_cDataPacket.Get_Data();
            tUINT32 l_uSize   = l_cDataPacket.Get_Data_Size();

            while (    (l_pTailCurrent)
                    && (l_pTailCurrent->szTailRest)
                    && (l_uSize)
                  )
            {
                tUINT32 l_uToCopy = MIN((tUINT32)l_pTailCurrent->szTailRest, l_uSize);
                memcpy(l_pTailCurrent->pTail + l_pTailCurrent->szTailOffs, l_pBuffer, l_uToCopy);
                l_pTailCurrent->szTailOffs += l_uToCopy;
                l_pTailCurrent->szTailRest -= l_uToCopy; 
                l_pBuffer                  += l_uToCopy;
                l_uSize                    -= l_uToCopy;

                if (    (l_pTailCurrent->szTailOffs == sizeof(sH_User_Raw))
                     && (!l_pTailCurrent->szTailRest)
                   )
                {
                    l_stHdrUser = *(sH_User_Data*)l_pTailCurrent->pTail;

                    if (m_bBigEndian)
                    {
                        l_stHdrUserRaw->dwBits = ntohl(l_stHdrUserRaw->dwBits);
                    }

                    l_pTailCurrent->szTailRest = l_stHdrUser.dwSize - sizeof(sH_User_Raw);
                    l_pTailCurrent->Realloc(l_stHdrUser.dwSize);
                }

                if (!l_pTailCurrent->szTailRest)
                {
                    l_stHdrUser = *(sH_User_Data*)l_pTailCurrent->pTail;

                    if (m_bBigEndian)
                    {
                        l_stHdrUserRaw->dwBits = ntohl(l_stHdrUserRaw->dwBits);
                    }

                    l_pDataChunk           = Pull_Free_Chunk();
                    l_pDataChunk->pBuffer  = l_pTailCurrent->pTail + sizeof(sH_User_Data);
                    l_pDataChunk->szBuffer = l_stHdrUser.dwSize - sizeof(sH_User_Data);
                    l_pDataChunk->pNext    = NULL;

                    if (!m_pChannels[l_stHdrUser.dwChannel_ID])
                    {
                        m_pChannels[l_stHdrUser.dwChannel_ID] = new CStreamP7();
                    }

                    l_pStream = m_pChannels[l_stHdrUser.dwChannel_ID];

                    if (l_pStream->m_pLast)
                    {
                        l_pStream->m_pLast->pNext = l_pDataChunk;
                    }
                    else
                    {
                        l_pStream->m_pFirst = l_pDataChunk;
                    }

                    l_pStream->m_pLast = l_pDataChunk;
                    l_pStream->PushTail(l_pTailCurrent);
                    l_pTailCurrent = NULL;
                }
            }

            if ((l_uSize) && (l_pTailCurrent))
            {
                LOG_INFO(TM("[0x%08p] Connection parsing error!"), this);
            }

            while (l_uSize)
            {
                if (l_uSize >= sizeof(sH_User_Raw))
                {
                    l_stHdrUser = *(sH_User_Data*)l_pBuffer;

                    if (m_bBigEndian)
                    {
                        l_stHdrUserRaw->dwBits = ntohl(l_stHdrUserRaw->dwBits);
                    }

                    if (l_uSize >= l_stHdrUser.dwSize)
                    {
                        l_pDataChunk           = Pull_Free_Chunk();
                        l_pDataChunk->pBuffer  = l_pBuffer + sizeof(sH_User_Data);
                        l_pDataChunk->szBuffer = l_stHdrUser.dwSize - sizeof(sH_User_Data);
                        l_pDataChunk->pNext    = NULL;

                        if (!m_pChannels[l_stHdrUser.dwChannel_ID])
                        {
                            m_pChannels[l_stHdrUser.dwChannel_ID] = new CStreamP7();
                        }

                        l_pStream = m_pChannels[l_stHdrUser.dwChannel_ID];

                        if (l_pStream->m_pLast)
                        {
                            l_pStream->m_pLast->pNext = l_pDataChunk;
                        }
                        else
                        {
                            l_pStream->m_pFirst = l_pDataChunk;
                        }

                        l_pStream->m_pLast = l_pDataChunk;

                        l_uSize   -= l_stHdrUser.dwSize;
                        l_pBuffer += l_stHdrUser.dwSize;
                    }
                    else
                    {
                        l_pTailCurrent = l_cTails.Count() ? l_cTails.Pull_First() : NULL;
                        if (!l_pTailCurrent)
                        {
                            l_pTailCurrent = new CTail(&l_cTails, l_stHdrUser.dwSize);
                        }
                        else
                        {
                            l_pTailCurrent->Realloc(l_stHdrUser.dwSize);
                        }

                        memcpy(l_pTailCurrent->pTail, l_pBuffer, l_uSize);
                        l_pTailCurrent->szTailRest = l_stHdrUser.dwSize - l_uSize;
                        l_pTailCurrent->szTailOffs = l_uSize;
                        l_uSize      = 0;
                    }
                }
                else
                {
                    l_pTailCurrent = l_cTails.Count() ? l_cTails.Pull_First() : NULL;
                    if (!l_pTailCurrent)
                    {
                        l_pTailCurrent = new CTail(&l_cTails);
                    }

                    memcpy(l_pTailCurrent->pTail, l_pBuffer, l_uSize);
                    l_pTailCurrent->szTailRest = sizeof(sH_User_Raw) - l_uSize;
                    l_pTailCurrent->szTailOffs = l_uSize;
                    l_uSize = 0;
                }
            }

            //process chunks////////////////////////////////////////////////////////////////////////////////////////////
            for (size_t l_szI = 0; l_szI < USER_PACKET_CHANNEL_ID_MAX_SIZE; l_szI++)
            {
                l_pStream = m_pChannels[l_szI];
                if (l_pStream)
                {
                    while (l_pStream->m_pFirst)
                    {
                        if (!l_pStream->m_bError)
                        {
                            if (!l_pStream->m_iStorare)
                            {
                                if (CreateStorage(l_pStream, l_szI))
                                {
                                    m_cTmpStreams.Add_After(m_cTmpStreams.Get_Last(), l_pStream);
                                }
                                else 
                                {
                                    Zero_Stream_Buffers(l_pStream);
                                    l_pStream->m_bError = TRUE;
                                    if (l_pStream->m_iStorare)
                                    {
                                        l_pStream->m_iStorare->Release();
                                        l_pStream->m_iStorare = NULL;
                                    }
                                }
                            }
            
                            if (l_pStream->m_iStorare)
                            {
                                Bk::eResult l_eResult = l_pStream->m_iStorare->Put_Packet(l_pStream->m_pFirst);
                                
                                if (Bk::eOk == l_eResult)
                                {
                                    //nothing to do
                                }
                                else if (    (Bk::eErrorClosed == l_eResult)
                                          || (Bk::eErrorMissmatch == l_eResult)
                                        )
                                {
                                    Register_Stream(l_pStream);
                                    l_pStream->Uninit();
                                }
                                else
                                {
                                    Zero_Stream_Buffers(l_pStream);
                                }
                            }
                        }
                        else //clean buffers
                        {
                            Zero_Stream_Buffers(l_pStream);
                        }
            
                        //put back to pool released chunks descriptors: only if size is 0
                        while (    (l_pStream->m_pFirst)
                                && (!l_pStream->m_pFirst->szBuffer)
                              )
                        {
                            Bk::stDataChunk *l_pNext = l_pStream->m_pFirst->pNext;
                            Push_Free_Chunks(l_pStream->m_pFirst);
                            l_pStream->m_pFirst = l_pNext;
                        }

                        if (!l_pStream->m_pFirst)
                        {
                            l_pStream->m_pLast = NULL;
                            l_pStream->ReleaseTails();
                        }
                    } //while (l_pStream->m_pFirst)
                }
            }

            //register new readers in core when storages were initialized///////////////////////////////////////////////
            Register_Streams();
            
            //cleanup pending packets///////////////////////////////////////////////////////////////////////////////////
            m_pConnection->Release_Packet(l_pPacket);
        }//if (MEVENT_TIME_OUT == l_uWait)
        else if (eThreadExit == l_uWait)
        {
            l_bExit = TRUE;
        }
    }

    m_cTmpStreams.Clear(FALSE);

    for (size_t l_szI = 0; l_szI < USER_PACKET_CHANNEL_ID_MAX_SIZE; l_szI++)
    {
        if (m_pChannels[l_szI])
        {
            delete m_pChannels[l_szI];
            m_pChannels[l_szI] = NULL;
        }
    }

    if (l_pTailCurrent)
    {
        delete l_pTailCurrent;
    }

    l_cTails.Clear(TRUE);
    m_cDataBlocks.Clear(TRUE);

    if (m_pAddress)
    {
        m_pAddress->Del_Conn(m_pConnection);
        m_pConnection = NULL;
    }

    LOG_INFO(TM("[0x%08p] Connection close thread"), this);

    if (m_iP7_Trace)
    {
        m_iP7_Trace->Unregister_Thread(0);
    }
}
