modularization/src/common/modulecomm/fastrtps/modulecomm_impl_shm.cpp

#ifdef USE_FASTRTPS
#include "modulecomm_impl_shm.h"

#include <thread>
#include <iostream>
#include <QDateTime>

#include <QMutex>
#include <QFile>

#ifdef Q_OS_LINUX
#include <unistd.h>
#endif

namespace iv {
namespace modulecomm {
static QMutex gmodulecomm_dds_Mutex;
static int createcount = 0;
}

}


void modulecomm_impl_shm::callbackTopic(const char * strdata,const unsigned int nSize,const unsigned int index, QDateTime * dt,const char * strmemname) {

  if(mbFunPlus)
  {
      mFun(strdata,nSize,index,dt,strmemname);
  }
  else
  {
     (*mpCall)(strdata,nSize,index,dt,strmemname);
  }
}


int modulecomm_impl_shm::GetTempConfPath(char *strpath)
{
    char strtmppath[256];
    QDateTime dt = QDateTime::currentDateTime();
    snprintf(strtmppath,256,"/tmp/adc_modulecomm_conf_%04d%02d%02d%02d%02d.ini",dt.date().year(),
             dt.date().month(),dt.date().day(),dt.time().hour(),dt.time().minute());
    QFile xFile;
    xFile.setFileName(strtmppath);
    char strtem[256];
    char strdata[10000];
    snprintf(strdata,10000,"");
    if(!xFile.exists())
    {
        if(xFile.open(QIODevice::ReadWrite))
        {
            snprintf(strtem,256,"[common]\n");strncat(strdata,strtem,10000);
            snprintf(strtem,256,"DCPSDefaultDiscovery=TheRTPSConfig\n");strncat(strdata,strtem,10000);
#ifdef dds_use_shm
            snprintf(strtem,256,"DCPSGlobalTransportConfig=myconfig\n");strncat(strdata,strtem,10000);
            snprintf(strtem,256,"[config/myconfig]\n");strncat(strdata,strtem,10000);
            snprintf(strtem,256,"transports=share\n");strncat(strdata,strtem,10000);
            snprintf(strtem,256,"[transport/share]\n");strncat(strdata,strtem,10000);
            snprintf(strtem,256,"transport_type=shmem\n");strncat(strdata,strtem,10000);
            snprintf(strtem,256,"pool_size=100000000\n");strncat(strdata,strtem,10000);
#endif
            snprintf(strtem,256,"[rtps_discovery/TheRTPSConfig]\n");strncat(strdata,strtem,10000);
            snprintf(strtem,256,"ResendPeriod=5\n");strncat(strdata,strtem,10000);
            xFile.write(strdata,strnlen(strdata,10000));
            xFile.close();
        }
    }
    strncpy(strpath,strtmppath,255);
    return 0;
}

modulecomm_impl_shm::modulecomm_impl_shm(const char * strcommname,int ntype )
{

    strncpy(mstrtopic,strcommname,255);

    iv::modulecomm::gmodulecomm_dds_Mutex.lock();
    if(ntype == type_recv)
    {
        mpSub = new TopicsSubscriber();
        mpSub->init(strcommname);
//        std::this_thread::sleep_for(std::chrono::milliseconds(10));

        mnType = type_recv;
    }
    else
    {
        mpPub = new TopicsPublisher();
        mpPub->init(strcommname);
 //       std::this_thread::sleep_for(std::chrono::milliseconds(10));
        mnType = type_send;
    }
    iv::modulecomm::createcount++;
    std::cout<<"count is "<<iv::modulecomm::createcount<<std::endl;
    iv::modulecomm::gmodulecomm_dds_Mutex.unlock();

    strncpy(mmodulemsg_type.mstrmsgidname,strcommname,255);
    mmodulemsg_type.mnBufSize = 0;
    mmodulemsg_type.mnMsgBufCount = 0;
    strncpy(mmodulemsg_type.mstrmsgname,strcommname,255);
#ifdef Q_OS_LINUX
    mmodulemsg_type.mnPID = getpid();
#endif

#ifdef USEDBUS
            mmsgres = QDBusMessage::createSignal("/catarc/adc",  "adciv.interface", "modulemsgres");
            mmsgres<<1;

            bool bconnect = QDBusConnection::sessionBus().connect(QString(),"/catarc/adc",  "adciv.interface", "modulemsgquery",this,SLOT(onQuery()));
            if(bconnect == false)
            {
                std::cout<<"procsm_if_readthread::procsm_if_readthread bconect is false"<<std::endl;
            }
#endif

}

int modulecomm_impl_shm::listenmsg(ModuleFun xFun)
{
    if(mnType == type_send)
    {
        std::cout<<"send not listen."<<std::endl;;
        return -1;
    }
    mbFunPlus = true;
    mFun = xFun;
    ModuleFun topicFunction = std::bind(&modulecomm_impl_shm::callbackTopic,this,std::placeholders::_1,
                                                                                std::placeholders::_2,
                                                                                std::placeholders::_3,
                                                                                std::placeholders::_4,
                                                                                std::placeholders::_5);
    mpSub->setReceivedTopicFunction(topicFunction);
    return 0;
}

int modulecomm_impl_shm::listenmsg(SMCallBack pCall)
{
    if(mnType == type_send)
    {
        std::cout<<"send not listen."<<std::endl;
        return -1;
    }
    mbFunPlus = false;
    mpCall = pCall;
    ModuleFun topicFunction = std::bind(&modulecomm_impl_shm::callbackTopic,this,std::placeholders::_1,
                                                                                std::placeholders::_2,
                                                                                std::placeholders::_3,
                                                                                std::placeholders::_4,
                                                                                std::placeholders::_5);
    mpSub->setReceivedTopicFunction(topicFunction);
    return 0;
}

#ifdef USEDBUS
    void modulecomm_impl_shm::onQuery()
    {
        QByteArray ba;
        ba.append((char *)&mmodulemsg_type,sizeof(iv::modulemsg_type));

        QList<QVariant> x;
        x<<ba;
        mmsgres.setArguments(x);
        QDBusConnection::sessionBus().send(mmsgres);
    }

#endif

void modulecomm_impl_shm::writemsg(const char *str, int nlen)
{
    if(mnType == type_recv)
    {
        std::cout<<"recv not send."<<std::endl;
        return ;
    }

    mpPub->senddata(str,nlen);

}
#endif