modularization/src/common/modulecomm/shm/procsm_if.cpp

#include "procsm_if.h"
#include <QTimer>

#include <iostream>



procsm_if_readthread::procsm_if_readthread(procsm *pPSM,SMCallBack pCall,const char * strsmname)
{
    mpPSM = pPSM;
    mpCall = pCall;
    strncpy(mstrsmname,strsmname,255);

#ifdef USEDBUS
    bool bconnect = QDBusConnection::sessionBus().connect(QString(),"/catarc/adc",  "adc.adciv.modulecomm", strsmname,this,SLOT(onNewMsg(int)));
    if(bconnect == false)
    {
        std::cout<<"procsm_if_readthread::procsm_if_readthread bconect is false"<<std::endl;
    }
#endif
}

procsm_if_readthread::procsm_if_readthread(procsm *pPSM,ModuleFun xFun,const char * strsmname)
{
    mpPSM = pPSM;
    mFun = xFun;
    strncpy(mstrsmname,strsmname,255);
    mbFunPlus = true;

#ifdef USEDBUS
    bool bconnect = QDBusConnection::sessionBus().connect(QString(),"/catarc/adc",  "adc.adciv.modulecomm", strsmname,this,SLOT(onNewMsg(int)));
    if(bconnect == false)
    {
        std::cout<<"procsm_if_readthread::procsm_if_readthread bconect is false"<<std::endl;
        mbDBUSOK = false;
        QTimer * timer = new QTimer();
        timer->setTimerType(Qt::PreciseTimer);
        delete timer;
    }
#endif
}


#ifdef USELCM
    void procsm_if_readthread::handlerMethod(const lcm::ReceiveBuffer *rbuf, const std::string &channel)
    {
        qDebug("lcm receiv data. ");
        mxindex++;
        QDateTime dt = QDateTime::currentDateTime();
        if(mbFunPlus)
        {
            mFun((char *)rbuf->data,rbuf->data_size,mxindex,&dt,mstrsmname);
        }
        else
        {
           (*mpCall)((char *)rbuf->data,rbuf->data_size,mxindex,&dt,mstrsmname);
        }
    }
#endif


void procsm_if_readthread::puaseread()
{
    mbRun = false;
}

void procsm_if_readthread::continueread()
{
    mbRun = true;
}

void procsm_if_readthread::run()
{
#ifdef USELCM
    mlcm.subscribe(mstrsmname,&procsm_if_readthread::handlerMethod,this);
    while(!QThread::isInterruptionRequested())
    {
        mlcm.handle();
    }
    return;
#endif
    QTime xTime;
    xTime.start();
    unsigned int nBufLen = 1;
    unsigned int nRead;
    char * str = new char[nBufLen];
    unsigned int index =0;


    QDateTime *pdt = new QDateTime();

    bool bAttach = false;
    while(!QThread::isInterruptionRequested())
    {
        if(mbRun == false)
        {
            msleep(10);
            continue;
        }
        if(bAttach == false)
        {
            bAttach = mpPSM->AttachMem();
            if(bAttach == false)
            {
                msleep(100);
                continue;
            }
            else
            {
                index = mpPSM->getcurrentnext();
            }
        }

        int nRtn = mpPSM->readmsg(index,str,nBufLen,&nRead,pdt);
        if(nRtn == 0)
        {
#ifdef USEDBUS
            if(mbDBUSOK == true)
            {
                mWaitMutex.lock();
                mwc.wait(&mWaitMutex,10);
                mWaitMutex.unlock();
            }
            else
            {
                msleep(1);
            }
#else
            msleep(1);
#endif
        }
        else
        {
            if(nRtn == -1)
            {
                nBufLen = nRead;
                delete str;
                if(nBufLen < 1)nBufLen = 1;
                str = new char[nBufLen];
  //              std::cout<<"modulecomm resize str to "<<nBufLen<<std::endl;
            }
            else
            {
                if(nRtn == -2)
                {
                   index = mpPSM->getcurrentnext();
                }
                else
                {
                   if(nRtn >0)
                   {
                       if(mbFunPlus)
                       {
                           mFun(str,nRtn,index,pdt,mstrsmname);
                       }
                       else
                       {
                          (*mpCall)(str,nRtn,index,pdt,mstrsmname);
                       }
                       index++;
                   }
                   else
                   {
                       usleep(100);
                   }
                }
            }
        }

    }
    delete str;
    delete pdt;
//    qDebug("Thread finish.");
}

#ifdef USEDBUS

void procsm_if_readthread::onNewMsg(int x)
{
    if(x == 100)std::cout<<x<<std::endl;
    mwc.wakeAll();
//    qDebug("wake");
}

#endif

procsm_if::procsm_if(const char * strsmname,const unsigned int nBufSize,const unsigned int nMaxPacCount,const int nMode)
{
    strncpy(mstrsmname,strsmname,255);
#ifdef USELCM
    if(nMode == procsm::ModeWrite)
    {

    }
    else
    {

    }
    return;
#endif
    mpPSM = new procsm(strsmname,nBufSize,nMaxPacCount,nMode);
    mnType = nMode;

    mTimer.setTimerType(Qt::PreciseTimer);

}

procsm_if::~procsm_if()
{
    if(mnType == procsm::ModeRead)
    {

        QTime xTime;
        xTime.start();
        bool bDel = true;
        mpReadThread->requestInterruption();
        while(!mpReadThread->isFinished())
        {
            if(xTime.elapsed() > 1000)
            {
                bDel = false;
                std::cout<<"procsm_if Thread Read not finish."<<std::endl;
                break;
            }
        }
        if(bDel)delete mpReadThread;
    }
    delete mpPSM;
}



int procsm_if::writemsg(const char *str, const unsigned int nSize)
{
    if(mbRun == false)return -2;
#ifdef USELCM
    int nres = mlcm.publish(mstrsmname,str,nSize);
    qDebug("publish message. res = %d",nres);
    return 0;
#endif
    if(mnType == procsm::ModeRead)return -1; //this is listen.
    return mpPSM->writemsg(str,nSize);
}

#ifdef USELCM
    void procsm_if::handlerMethod(const lcm::ReceiveBuffer *rbuf, const std::string &channel)
    {
        qDebug("receiv data. ");
    }
#endif
int procsm_if::listenmsg(SMCallBack pCall)
{
//#ifdef USELCM
////     mlcm.subscribe(mstrsmname,&handlerMethod2);
//    mlcm.subscribe(mstrsmname,&procsm_if::handlerMethod,this);
//    while(true)
//    {
//        mlcm.handle();
//    }
//    return 0;
//#endif
    if(mnType == procsm::ModeWrite)return -1; //listening.
    mpReadThread = new procsm_if_readthread(mpPSM,pCall,mstrsmname);
//    mpReadThread->setPriority(QThread::TimeCriticalPriority);
//    mpReadThread->start();
    mpReadThread->start(QThread::HighestPriority);
//    mnType = 1;
    return 0;
}

int procsm_if::listenmsg(ModuleFun xFun)
{
//#ifdef USELCM
//    mlcm.subscribe(mstrsmname,&procsm_if::handlerMethod,this);
//    while(true)
//    {
//        mlcm.handle();
//    }
//    return 0;
//#endif
    if(mnType == procsm::ModeWrite)return -1; //listening.
    mpReadThread = new procsm_if_readthread(mpPSM,xFun,mstrsmname);
//    mpReadThread->setPriority(QThread::TimeCriticalPriority);
//    mpReadThread->start();
    mpReadThread->start(QThread::HighestPriority);
//    mnType = 1;
    return 0;
}

void procsm_if::stoplisten()
{
    if(mnType != 1)return;
    mpReadThread->requestInterruption();
    while(!mpReadThread->isFinished());
    mnType = 0;
//    mpReadThread->deleteLater();
    qDebug("stop listen ok");
}

void procsm_if::pausecomm()
{
    mbRun = false;
    if(mnType == procsm::ModeRead)
    {
        mpReadThread->puaseread();
    }
}

void procsm_if::continuecomm()
{
    mbRun = true;
    if(mnType == procsm::ModeRead)
    {
        mpReadThread->continueread();
    }
}