EDDMsgLoop.cpp

Go to the documentation of this file.

#include <EDDSourceLib/inc/EDDMsgLoop.h>
#include <EDDSourceLib/inc/EDDDataSource.h>
#include <EDDSourceLib/inc/EDDThreadManager.h>
#include <EDDSourceLib/inc/EDDDataPipe.h>
#include <EDDSourceLib/inc/EDDLockManager.h>
#include <proto/EDDMsgHdr.h>
#include <proto/EDDSourceMsg.h>

// the tcEDDMsgLoop is a singleton, assign default pointer to NULL
tcEDDMsgLoop* tcEDDMsgLoop::gpMsgLoop = NULL;

tcEDDMsgLoop* 
tcEDDMsgLoop::GetInstance(void)
{
   if (NULL == gpMsgLoop)
   {
      gpMsgLoop = new tcEDDMsgLoop();
   }
   return gpMsgLoop;
}

int 
tcEDDMsgLoop::AddDataPipe(tcEDDDataPipe* apDataPipe)
{
    int rv = -1;
    mpLockManager->TakeLock(mhTableLock);
    for (int index = 0; index < gnMaxDataPipes; index++)
    {
        if (NULL == maDataPipes[index])
        {
            maDataPipes[index] = apDataPipe;
            rv = index;
            break;
        }
    }
    // if we found a valid slot
    if (rv >= 0)
    {
        // dispatch the receiver thread
        maShutdownThread[rv] = false;
        maRcvThreadHandles[rv] =
            mpThreadManager->StartThread(DispatchReceiverThread,
                                         tcEDDThreadManager::eeMedium,
                                         (void*)rv);
    }    
    mpLockManager->ReleaseLock(mhTableLock);
    return rv;
}

int 
tcEDDMsgLoop::RemoveDataPipe(int anPipeID)
{
    int rv = -1;
    mpLockManager->TakeLock(mhTableLock);
    if (NULL != maDataPipes[anPipeID])
    {
        // Stop receiver thread first!
        maShutdownThread[anPipeID] = true;
        // This is really a "wait for" ....
        mpThreadManager->StopThread(maRcvThreadHandles[anPipeID]);
        
        // once stopped, clean up the tables...
        for (int index = 0; index < gnMaxDataSources; index++)
        {
            maSubscriptionTable[anPipeID][index] = 0;
        }
        maDataPipes[anPipeID] = 0;
        rv = 0;
    }
    mpLockManager->ReleaseLock(mhTableLock);
    return rv;
}

int 
tcEDDMsgLoop::AddDataSource(tcEDDDataSource* apDataSource)
{
    int rv = -1;
    mpLockManager->TakeLock(mhTableLock);
    for (int index = 0; index < gnMaxDataSources; index++)
    {
        if (NULL == maDataSources[index])
        {
            maDataSources[index] = apDataSource;
            rv = index;
            break;
        }
    }
    mpLockManager->ReleaseLock(mhTableLock);
    return rv;
}

int 
tcEDDMsgLoop::RemoveDataSource(int anSourceID)
{
    int rv = -1;
    mpLockManager->TakeLock(mhTableLock);
    if (NULL != maDataSources[anSourceID])
    {
        for (int index = 0; index < gnMaxDataPipes; index++)
        {
            maSubscriptionTable[index][anSourceID] = 0;
        }
        maDataSources[anSourceID] = NULL;
        rv = 0;
    }
    mpLockManager->ReleaseLock(mhTableLock);
    return rv;
}
    
int 
tcEDDMsgLoop::SendMsg(tsEDDMsgHdr* apMsg, bool BrdCast)
{
    int rv = 0;
    mpLockManager->TakeLock(mhTableLock);
    for (int index = 0; index < gnMaxDataPipes; index++)
    {
        if ( (maSubscriptionTable[index][apMsg->mnDataSource] ) ||
             (BrdCast && (NULL != maDataPipes[index])) )
        {
            apMsg->mnSeqNo = maDataPipes[index]->NextSeqNo();
            rv |= maDataPipes[index]->SendMsg((void*)apMsg,apMsg->mnMsgLen);
        }
    }
    mpLockManager->ReleaseLock(mhTableLock);
    return rv;
}

bool 
tcEDDMsgLoop::IsSubscribed(uint16_t anSourceId)
{
    bool rv = false;
    mpLockManager->TakeLock(mhTableLock);
    for (int index = 0; index < gnMaxDataPipes; index++)
    {
        if ( maSubscriptionTable[index][anSourceId] )
        {
            rv = true;
            break;
        }
    }
    mpLockManager->ReleaseLock(mhTableLock);
    return rv;
}

int 
tcEDDMsgLoop::GetMaxMessageLen(int anSourceID)
{
    int MaxLen = -1;
    mpLockManager->TakeLock(mhTableLock);
    for (int index = 0; index < gnMaxDataPipes; index++)
    {
        if (maSubscriptionTable[index][anSourceID])
        {
            int ml = maDataPipes[index]->GetMaxMessageLen();
            if ((0 > ml) && (ml > MaxLen))
            {
                MaxLen = ml;
            }
        }
    }
    mpLockManager->ReleaseLock(mhTableLock);
    return MaxLen;
}
    
int 
tcEDDMsgLoop::DispatchReceiverThread(void* anPipeID, void* apFoo, void* apGoo)
{
    unsigned int rv = 0;
    // launch thread to monitor data....
    rv = GetInstance()->ReceiverThread((int)anPipeID);
    return rv;
}

int 
tcEDDMsgLoop::ReceiverThread(int anPipeID)
{
    unsigned int rv = 0;
    int          bufpos = 0;
    char*        buffer;
    teParseState state = eeWaitSync;
    int          readlen = 4;
    tsEDDMsgHdr* lpMsg;
    
    int buflen = maDataPipes[anPipeID]->GetMaxMessageLen();
    if (buflen < 8)
    {
        buflen = 4096;
    }
    buffer = new char[buflen];
    lpMsg = (tsEDDMsgHdr*)buffer;
    
    // TODO - thread should quit when Pipe is Removed.
    while(1)
    {
        // Check For Data
                 readlen = (readlen > buflen) ? buflen : readlen;
        mpLockManager->TakeLock(mhTableLock);
        if (maShutdownThread[anPipeID])
        {
            mpLockManager->ReleaseLock(mhTableLock);
            break;
        }
        int bytes = maDataPipes[anPipeID]->GetData(&buffer[bufpos],readlen,0);
        mpLockManager->ReleaseLock(mhTableLock);
        // got data
        if (bytes > 0)
        {
            bool done = false;
            while(!done)
            {
                switch(state)
                {
                case eeGetHeader:
                    // if we have the full header get the payload
                    if (bufpos+bytes >= sizeof(tsEDDMsgHdr))
                    {
                        state = eeGetBytes;
                    }
                    else
                    {
                        readlen = sizeof(tsEDDMsgHdr)-(bufpos+bytes);
                        done = true;
                    }
                    break;
                case eeGetBytes:    
                    // if we have a full message, process it.
                    if (lpMsg->mnMsgLen <= bufpos+bytes)
                    {
                        // process the message (route it to a data source)
                        HandleMessage(lpMsg, anPipeID);
                        
                        // shuffle remaining bytes down to beginning
                        int temp = lpMsg->mnMsgLen;
                        for (int i = 0; i < bufpos+bytes-temp; i++)
                        {
                            buffer[i] = buffer[i+temp];
                        } 
                        
                        bytes  = bufpos+bytes-temp;
                        bufpos = 0;
                        
                        // look for next sync word
                        state = eeWaitSync;
                    }
                    else  // not enough bytes .... mark as done and go get more
                    {
                        readlen = lpMsg->mnMsgLen-(bufpos+bytes);
                        done = true;
                    }
                    break;
                case eeWaitSync:
                default:
                    // if we have at least enough bytes to check for sync
                    if (bufpos+bytes >= 4)
                    {
                        uint32_t temp1 = buffer[0];
                        uint32_t temp2 = buffer[1]<<8;
                        uint32_t temp3 = buffer[2]<<16;
                        uint32_t temp4 = buffer[3]<<24;
                        uint32_t check = temp1|temp2|temp3|temp4;
                        if (EDD_MSG_SYNC == check)
                        {
                           state = eeGetHeader;
                        }
                        else
                        {
                           // shuffle data down to beginning of buffer
                           for(int i = 1; i < bytes+bufpos; i++)
                           {
                              buffer[i-1] = buffer[i];
                           }
                           bytes = bytes-1;
                        }
                    }
                    else // not enough bytes
                    {
                        readlen = 4-(bufpos+bytes);
                        done = true;
                    }
                    break;
                }
                bufpos += bytes;
                bytes = 0;
            } // not done processing bytes
        }
        else if (bytes < 0) // error!
        {
            // what to do???
            break;
        }
        else if (bytes == 0)
        {
           mpThreadManager->Sleep(10, anPipeID);
        }
    }
    delete [] buffer;
    return rv;
}
        
tcEDDMsgLoop::tcEDDMsgLoop(void)
: mpThreadManager(NULL)
, mpLockManager(NULL)
{
    int index = 0;
    InitOSDependencies();
    
    mhTableLock = mpLockManager->CreateLock();
    
    mpLockManager->TakeLock(mhTableLock);
    for (index = 0; index < gnMaxDataPipes; index++)
    {
        maDataPipes[index] = NULL;
        for (int j = 0; j < gnMaxDataSources; j++)
        {
            maSubscriptionTable[index][j] = 0;
        }
    }
    for (index = 0; index < gnMaxDataSources; index++)
    {
        maDataSources[index] = NULL;
    }
    mpLockManager->ReleaseLock(mhTableLock);
}

tcEDDMsgLoop::~tcEDDMsgLoop(void)
{
    /* TODO */
}

void
tcEDDMsgLoop::HandleMessage(tsEDDMsgHdr* apMsg, int anPipeID)
{
    
    mpLockManager->TakeLock(mhTableLock);
    // handle all call on source descriptions
    if (apMsg->mnAllSources) 
    {
        for (int index = 0; index < gnMaxDataSources; index++)
        {
            if ( maDataSources[index] != NULL)
            {
                maDataSources[index]->HandleMsg(apMsg);
            }            
        }
    }
    else // not an all call
    {
        // if it's a subscription request
        if (apMsg->mnMsgId == EDDMSGID_SOURCEDATAREQMSG)
        {
            tsEDDSourceDataReqMsg* lpDataReqMsg = (tsEDDSourceDataReqMsg*)apMsg;
            if ( (apMsg->mnDataSource < gnMaxDataSources) &&
                 (maDataSources[apMsg->mnDataSource] != NULL) )
            {
                maSubscriptionTable[anPipeID][apMsg->mnDataSource] = lpDataReqMsg->mnStart;
            }
        } // it's a "normal" message
        else if ( (apMsg->mnDataSource < gnMaxDataSources) &&
             (maDataSources[apMsg->mnDataSource] != NULL) )
        {
            maDataSources[apMsg->mnDataSource]->HandleMsg(apMsg);
        }
    }
    mpLockManager->ReleaseLock(mhTableLock);    
}

---