/home/fwph/code/wurde/rde/core/capabilities/Heartbeat.cpp

Go to the documentation of this file.

#include <Heartbeat.H>
#include <CommsManager.H>
#include <Logger.H>

// $Id: Type.cpp 68 2007-02-01 17:56:57Z fwph $

using namespace WURDE;

Heartbeat::Heartbeat(const std::string & streamname){
       init(streamname,STRAT_NORMAL,"Any");

       //global name will be set later
       m_haveGlobalname=false;
}

Heartbeat::Heartbeat(const ConnectionStrategy & strategy,const std::string & gblName){
       init("Any",strategy,gblName);

       if(strategy==STRAT_NORMAL){
              //then the global name really specifies the streamname,
              // and the global name should be assigned automagically once we have
              // confirmation that the stream has been started
              m_haveGlobalname=false;   
       }

       m_haveStream=false;
       
}

Heartbeat::Heartbeat(const std::string &streamname,const std::string &gblName){
       init(streamname,STRAT_NORMAL,gblName);
}

void Heartbeat::init(const std::string &streamname,const ConnectionStrategy &strategy,const std::string &gblName){
  m_globalname=gblName;
  m_streamname=streamname;

  myCOMObject=NULL;
  
  //  data.source=streamname.c_str();
  requests=info;

  m_queue=false;
  m_supplier=false;
  m_doTimestamp=true;
  m_doTag=true;

  m_doPublishData=false;
  m_doPublishInfo=false;

  m_initialize=false;
  m_strategy=strategy;

  m_haveStream=true;
  m_haveGlobalname=true;
  m_doPublishData=false;
  m_doPublishInfo=false;

  m_doAutoPing=true;

  info.source=streamname;
  requests.source=m_globalname;
  info.target="Any";
  requests.target=streamname;
  info.timestamp.now();
  requests.timestamp.now();

  m_streamRequest=STAT_NULL;
  
  m_sourceModule="Unknown";
}


Heartbeat::~Heartbeat(){
  myManager=NULL;
  if(myCOMObject){

         //      delete myCOMObject;
  }
}


void Heartbeat::activateSupplier(){
       if(m_haveStream){
              m_supplier=true;
              m_globalname=m_streamname;
              m_haveGlobalname=true;
              m_initialize=false;
              if(myCOMObject){
                      //                     myCOMObject=new CMUIPCRobotProtocol::CMUIPCHeartbeat(m_globalname);
                     myCOMObject->setGlobalName(m_globalname);
                     myCOMObject->setStreamName(m_streamname);
                     myCOMObject->setInfo(&requests);
                     myCOMObject->setRequests(&info);
                     myCOMObject->setStaging(&data);
                     myCOMObject->activateSupplier();
                     myCOMObject->setAutoPing(m_doAutoPing);
                     myCOMObject->setInfoVector(&m_requestVector);
                     myCOMObject->setRequestVector(&m_infoVector);
                     myCOMObject->setDataVector(&m_dataVector);

                     data.source=m_streamname;
                     info.source=m_streamname;

                     if(m_queue){
                            myCOMObject->setQueueMode(true);
                     }

              }else{
                      g_fatal("Attempted to activateSupplier before setting a COMObject Please just use registerSupplier.");
                      exit(-1);
              }
        data.TTL=0;
        if(m_supplier){
                info.moduleName.setCMode(C_SUPPLIER);
        }else{
                info.moduleName.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.locale.setCMode(C_SUPPLIER);
        }else{
                info.locale.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.state.setCMode(C_SUPPLIER);
        }else{
                info.state.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.interfaces.setCMode(C_SUPPLIER);
        }else{
                info.interfaces.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.managedType.setCMode(C_SUPPLIER);
        }else{
                info.managedType.setCMode(C_CONSUMER);
        }
       }else{
              //this shouldn't happen, but there's no real problem with it.
              //it'd just be dumb.
              m_initialize=true;
              m_supplier=true;
       }
       
       requests=info;

}

void Heartbeat::activateConsumer(){
       std::string localname;
       
       if(m_haveStream){
              m_initialize=false;
              m_supplier=false;
              if(myCOMObject&&myManager){
                      /* global name so we don't have conflicts with the naming service */
                      if(!m_haveGlobalname&&m_strategy==STRAT_NORMAL){
                              m_globalname=myManager->getName()+HEARTBEAT+m_streamname+"Consumer";
                              m_haveGlobalname=true;
                      }

                     myCOMObject->setGlobalName(m_globalname);
                     myCOMObject->setStreamName(m_streamname);
                     requests.source=m_globalname;
                     data.source=m_streamname;
                     info.source=m_streamname;
                     myCOMObject->setInfo(&info);
                     myCOMObject->setRequests(&requests);
                     myCOMObject->setStaging(&data);
                     myCOMObject->setAutoPing(m_doAutoPing);
                     myCOMObject->activateConsumer();
                     myCOMObject->setInfoVector(&m_infoVector);
                     myCOMObject->setRequestVector(&m_requestVector);
                     myCOMObject->setDataVector(&m_dataVector);

                     if(m_queue){
                            myCOMObject->setQueueMode(true);
                     }
              }else{
                      g_fatal("Attempted to activateConsumer before setting a COMObject and CommsManager. Please use registerConsumer instead.");
                     exit(-1);
              }

        data.TTL=0;
        if(m_supplier){
                info.moduleName.setCMode(C_SUPPLIER);
        }else{
                info.moduleName.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.locale.setCMode(C_SUPPLIER);
        }else{
                info.locale.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.state.setCMode(C_SUPPLIER);
        }else{
                info.state.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.interfaces.setCMode(C_SUPPLIER);
        }else{
                info.interfaces.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.managedType.setCMode(C_SUPPLIER);
        }else{
                info.managedType.setCMode(C_CONSUMER);
        }
       }else{
              //no stream, so we can't connect yet. tell the system 
              //to initialize as soon as we've got one
              m_initialize=true;
       }

}

void Heartbeat::runUpdate(){
       if(myCOMObject){

              myCOMObject->runUpdate(m_doPublishData,m_doPublishInfo);
              
              m_doPublishData=false;
              m_doPublishInfo=false;
              
       }
}

void Heartbeat::setManager(CommsManager *myManager_){

  if(!myCOMObject){
    myManager=myManager_;
  }

}

void Heartbeat::setQueueMode(const bool & val){

       if(!myCOMObject){
              m_queue=val;
       }else if(!m_queue&&val){
              m_queue=val;
              myCOMObject->setDataVector(&m_dataVector);
              myCOMObject->setQueueMode(true);

       }else if(!val&&m_queue){
              m_queue=val;
              myCOMObject->setQueueMode(false);
              while(!m_dataVector.empty()){
                     m_dataVector.pop();
              }
              while(!m_infoVector.empty()){
                     m_infoVector.pop();
              }
              while(!m_requestVector.empty()){
                     m_requestVector.pop();
              }
       }
}

bool Heartbeat::newData(){
       if(m_queue){
              return !m_dataVector.empty();
       }else{
              if(myCOMObject){
                     return myCOMObject->newData();
              }
       }

       return false;
}

bool Heartbeat::newInfo(){
       if(m_queue){
              if(m_supplier){
                     return !m_requestVector.empty();
              }

              return !m_infoVector.empty();
       }else{
              if(myCOMObject){
                     return myCOMObject->newInfo();
              }
       }


       return false;
}

void Heartbeat::publishData(){
       if(m_supplier){
              if(m_doTimestamp){
                     data.timestamp.now();
              }
              if(m_doTag){
                     data.source=m_globalname;
              }
              if(m_queue){
                     HeartbeatDataStruct temp;
        temp.TTL=0;
                     temp.source=data.source;
temp.timestamp=data.timestamp;
temp.TTL=data.TTL;
 
                     m_dataVector.push(temp);
                              //                     m_dataVector.push(data);
              }
              m_doPublishData=true;
       }
}

void Heartbeat::publishInfo(){
       if(m_supplier){

              info.mType=MESSAGE_INFO;
              if(m_doTimestamp){
                     info.timestamp.now();
              }

              if(m_doTag){
                     info.source=m_globalname;
                     info.target="Any";
              }

              if(m_queue){
                     HeartbeatInfoStruct temp;
                     
                    
                     temp.moduleName=info.moduleName;
temp.source=info.source;
temp.locale=info.locale;
temp.state=info.state;
temp.mType=info.mType;
temp.target=info.target;
temp.timestamp=info.timestamp;
temp.interfaces=info.interfaces;
temp.managedType=info.managedType;

                     m_infoVector.push(temp);
                    
                     //              m_infoVector.push(info);
              }

              m_doPublishInfo=true;
       }else{

              requests.mType=MESSAGE_REQUEST;
              
              if(m_doTimestamp){
                     requests.timestamp.now();
              }

              if(m_doTag){
                     requests.source=m_globalname;
                     requests.target=m_streamname;
              }

              if(m_queue){
                     HeartbeatInfoStruct temp;
        if(m_supplier){
                temp.moduleName.setCMode(C_SUPPLIER);
        }else{
                temp.moduleName.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                temp.locale.setCMode(C_SUPPLIER);
        }else{
                temp.locale.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                temp.state.setCMode(C_SUPPLIER);
        }else{
                temp.state.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                temp.interfaces.setCMode(C_SUPPLIER);
        }else{
                temp.interfaces.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                temp.managedType.setCMode(C_SUPPLIER);
        }else{
                temp.managedType.setCMode(C_CONSUMER);
        }
                     temp.moduleName=requests.moduleName;
temp.source=requests.source;
temp.locale=requests.locale;
temp.state=requests.state;
temp.mType=requests.mType;
temp.target=requests.target;
temp.interfaces=requests.interfaces;
temp.timestamp=requests.timestamp;
temp.managedType=requests.managedType;

                     m_requestVector.push(temp);

                     //              m_requestVector.push(requests);
              }

              m_doPublishInfo=true;
       }
}

HeartbeatDataStruct Heartbeat::getNextData(){
       if(m_queue){
              HeartbeatDataStruct retval;
        retval.TTL=0;
              retval.source="NULL";
              if(m_dataVector.empty()){
                     return data;
              }
             
              retval.source=m_dataVector.front().source;
retval.timestamp=m_dataVector.front().timestamp;
retval.TTL=m_dataVector.front().TTL;

                       //             retval=m_dataVector.front();
              m_dataVector.pop();
              data=retval;
       
              return retval;
       }else{
              myCOMObject->resetData();
              return data;
       }
}

HeartbeatInfoStruct Heartbeat::getNextInfo(){
       if(m_queue){
              HeartbeatInfoStruct retval;
        if(m_supplier){
                retval.moduleName.setCMode(C_SUPPLIER);
        }else{
                retval.moduleName.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.locale.setCMode(C_SUPPLIER);
        }else{
                retval.locale.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.state.setCMode(C_SUPPLIER);
        }else{
                retval.state.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.interfaces.setCMode(C_SUPPLIER);
        }else{
                retval.interfaces.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.managedType.setCMode(C_SUPPLIER);
        }else{
                retval.managedType.setCMode(C_CONSUMER);
        }
              retval.source="NULL";
              if(m_supplier){
                     if(m_requestVector.empty()){
                            return requests;
                     }
                     retval.moduleName=m_requestVector.front().moduleName;
retval.source=m_requestVector.front().source;
retval.locale=m_requestVector.front().locale;
retval.state=m_requestVector.front().state;
retval.mType=m_requestVector.front().mType;
retval.target=m_requestVector.front().target;
retval.interfaces=m_requestVector.front().interfaces;
retval.timestamp=m_requestVector.front().timestamp;
retval.managedType=m_requestVector.front().managedType;

                              //                     retval=m_requestVector.front();
        if(m_supplier){
                retval.moduleName.setCMode(C_SUPPLIER);
        }else{
                retval.moduleName.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.locale.setCMode(C_SUPPLIER);
        }else{
                retval.locale.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.state.setCMode(C_SUPPLIER);
        }else{
                retval.state.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.interfaces.setCMode(C_SUPPLIER);
        }else{
                retval.interfaces.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.managedType.setCMode(C_SUPPLIER);
        }else{
                retval.managedType.setCMode(C_CONSUMER);
        }
                     m_requestVector.pop();
                     requests=retval;
                     return retval;
              }else{
                     if(m_infoVector.empty()){
                            return info;
                     }
                     retval.moduleName=m_infoVector.front().moduleName;
retval.source=m_infoVector.front().source;
retval.locale=m_infoVector.front().locale;
retval.state=m_infoVector.front().state;
retval.mType=m_infoVector.front().mType;
retval.target=m_infoVector.front().target;
retval.interfaces=m_infoVector.front().interfaces;
retval.timestamp=m_infoVector.front().timestamp;
retval.managedType=m_infoVector.front().managedType;

        if(m_supplier){
                retval.moduleName.setCMode(C_SUPPLIER);
        }else{
                retval.moduleName.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.locale.setCMode(C_SUPPLIER);
        }else{
                retval.locale.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.state.setCMode(C_SUPPLIER);
        }else{
                retval.state.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.interfaces.setCMode(C_SUPPLIER);
        }else{
                retval.interfaces.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                retval.managedType.setCMode(C_SUPPLIER);
        }else{
                retval.managedType.setCMode(C_CONSUMER);
        }
                     m_infoVector.pop();
                     info=retval;
                     // the requests struct really should get the values in 
                     // info
                     requests=info;
                     
                     return retval;
              }
       }else{
              myCOMObject->resetInfo();
              if(m_supplier){
        if(m_supplier){
                requests.moduleName.setCMode(C_SUPPLIER);
        }else{
                requests.moduleName.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                requests.locale.setCMode(C_SUPPLIER);
        }else{
                requests.locale.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                requests.state.setCMode(C_SUPPLIER);
        }else{
                requests.state.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                requests.interfaces.setCMode(C_SUPPLIER);
        }else{
                requests.interfaces.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                requests.managedType.setCMode(C_SUPPLIER);
        }else{
                requests.managedType.setCMode(C_CONSUMER);
        }
                     return requests;
              }else{
        if(m_supplier){
                info.moduleName.setCMode(C_SUPPLIER);
        }else{
                info.moduleName.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.locale.setCMode(C_SUPPLIER);
        }else{
                info.locale.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.state.setCMode(C_SUPPLIER);
        }else{
                info.state.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.interfaces.setCMode(C_SUPPLIER);
        }else{
                info.interfaces.setCMode(C_CONSUMER);
        }
        if(m_supplier){
                info.managedType.setCMode(C_SUPPLIER);
        }else{
                info.managedType.setCMode(C_CONSUMER);
        }
                     // the requests struct really should get the values in 
                     // info
                     requests=info;
                     return info;
              }
       }
}

void Heartbeat::setStreamName(const std::string & name){

       if(m_supplier&&name=="Any"){
              return;
       }

       m_streamname=name;
       m_haveStream=true;

       if(m_supplier){
              m_globalname=name;
       }
       
       if(myCOMObject){
              myCOMObject->setStreamName(name);
       }

       if(m_initialize){
              if(m_supplier){
                     activateSupplier();
              }else{
                     activateConsumer();
              }
       }

       m_streamRequest=STAT_OKAY;
}

void Heartbeat::setGlobalName(const std::string & name){
       
       if(name=="Any"){
              return;
       }

       m_globalname=name;
       m_haveGlobalname=true;
       if(m_supplier){
              m_streamname=name;
       }

       if(myCOMObject){
              myCOMObject->setGlobalName(name);
       }
}

void Heartbeat::doPing(){
       if(myCOMObject){
              myCOMObject->doPing();
       }
}

void Heartbeat::setAutoPing(const bool & val){
       m_doAutoPing=val; 
       if(myCOMObject){
              myCOMObject->setAutoPing(val);
       }
}

---