CETDDS.h

Go to the documentation of this file.

#ifndef _RMS_CETDDS_H
#define _RMS_CETDDS_H

/*  This file (CETDDS.h) was created by Ron Rechenmacher <ron@fnal.gov> on
    Jan 21, 2010. "TERMS AND CONDITIONS" governing this file are in the README
    or COPYING file. If you do not have such a file, one can be obtained by
    contacting Ron or Fermi Lab in Batavia IL, 60510, phone: 630-840-3000.
    $RCSfile: CETDDS.h,v $
    rev="$Revision: 1.26.12.1 $$Date: 2019/09/27 00:07:31 $";
    */

#include "ccpp_dds_dcps.h"
#include "rms/base/RmsCloseable.h"
#include "rms/provider/CheckStatus.h"
#include "rms/base/RmsRuntimeException.h"
#include "rms/provider/ProcessSignalHandler.h"

#include <NovaDAQUtilities/EnvVarCache.h>
#include <string>
#include <iostream>
#include <boost/algorithm/string/replace.hpp>
#include <boost/interprocess/smart_ptr/unique_ptr.hpp>
#include <sys/time.h>
#include "boost/bind.hpp"

// -- DDS Includes
#include <dds/domain.hpp>
#include <dds/topic.hpp>
#include <dds/reader.hpp>
#include <dds/traits.hpp>


#define CETDDS_DEBUG 0
#define CETDDS_EXTRA_DEBUG 0
#define CETDDS_DEBUG1 0

#define declare_dds_message(TOPIC) REGISTER_TOPIC_TRAITS(TOPIC)

namespace gov {

namespace fnal {

namespace cd {

namespace rms {
  
   template <typename D> struct Deleter {
     void operator()(D *p) {
       assert( p != NULL );
       delete p;
     }
   };
   
namespace provider {


std::string create_extended_topic_name_for( std::string const& type_name,
        std::string const& topic_name );

std::string create_content_filtered_topic_name_for( std::string const& type_name,
        std::string const& topic_name );

typedef std::pair<std::string, std::vector<std::string> > dds_topic_filter_t;

dds_topic_filter_t& create_topic_filter_for(dds_topic_filter_t & filter, std::string const& type_name,
        std::string const& topic_name );

std::string unique_topic_name_for(std::string const& extended_topic_name );

bool string2bool (std::string const& bool_string );



void dumpReceivedSamples( dds::SampleInfoSeq const& infoSeq,
                          const unsigned int length, const int status );


template < class TT_ > class DDSTopicWriterClass
: virtual public base::RmsCloseable, public base::Notifiable
{
public:

    DDSTopicWriterClass( char const * topic_name );
    DDSTopicWriterClass( char const * topic_name, char const * partition_name );
    ~DDSTopicWriterClass();

    int write( TT_ const& mm );
    void close() throw();
    bool hasBeenClosed() throw();

private:
    void init(std::string const& topic_name,  std::string const& partition_name);
    void waitForNetworkServices();

private:
    boost::shared_ptr< dds::DataWriter< TT_ > >  writer;

    volatile bool  _hasBeenClosed;
    double   _startTime;
    bool     _startupDone;
};

template < class TT_ > class DDSTopicReaderClass
: virtual public base::RmsCloseable, public base::Notifiable
{
public:

    DDSTopicReaderClass( char const * topic_name );
    DDSTopicReaderClass( char const * topic_name, char const * partition_name );
    ~DDSTopicReaderClass();

    int take_nowait( TT_ & mm );
    int take_timed_wait( TT_ & mm, unsigned int usec );
    int take_wait( TT_ & mm );
    void close() throw();
    bool hasBeenClosed() throw();
    
private:
    void init(std::string const& topic_name,  std::string const& partition_name);
    int take_timed_wait( TT_ & topicdata, dds::Duration_t const& duration );
    void receiveMessage(boost::shared_ptr< dds::DataReader<TT_> > & reader,
                        TT_ & topicdata, dds::ReturnCode_t & status);
    
    void dummyHandler() {};
    void statusConditionHandler(boost::shared_ptr< dds::DataReader<TT_> > & reader);

private:
    boost::shared_ptr< dds::DataReader<TT_> >    reader;
    boost::interprocess::unique_ptr< dds::WaitSet, Deleter<dds::WaitSet> > wait_set;

    volatile bool  _hasBeenClosed;
};


///////////////////////////////////////////////////////////////////////////////

// DomainParticipant Singleton

class DPSingleton
{
public:
    static DPSingleton& Instance() {
        static DPSingleton singleton;
        return singleton;
    }

    void disconnect_participant();
    void connect_participant();
    void reconnect_participant();

    // adds / removes base::Notifiable to/from the exit notification list maintained by ProcessSignalHandler
    void addToCleanupList(base::Notifiable* notifiable);
    void removeFromCleanupList(base::Notifiable* notifiable);


protected:
    template <class TT_> friend  class DDSTopicWriterClass;
    template <class TT_> friend  class DDSTopicReaderClass;

    dds::DomainParticipant& get_participant();

// Other non-static member functions
private:
    boost::interprocess::unique_ptr<dds::DomainParticipant,Deleter<dds::DomainParticipant> > dp;

    DPSingleton();            // Private constructor
    ~DPSingleton() {
        signalHandler.release();
    };            // Private constructor

    DPSingleton( const DPSingleton& );          // Prevent copy-construction
    DPSingleton& operator=( const DPSingleton& ); // Prevent assignment

private:
    //closes DDS connections upon receiving a process signal, e.g. SIGINT, and SIGTERM
    ProcessSignalHandler_t signalHandler;

};

//  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -

template < class TT_ >
DDSTopicWriterClass< TT_ >::DDSTopicWriterClass( char const * topic_name )
{
#if CETDDS_DEBUG
    std::cout << topic_name << "\n" << "hello\n";
#endif

    init( std::string(topic_name), std::string("unknown"));
}

template < class TT_ >
DDSTopicWriterClass< TT_ >::DDSTopicWriterClass( char const * topic_name, char const * partition_name )
{
#if CETDDS_DEBUG
    std::cout << topic_name << "\n" << "hello\n";
#endif

    init( std::string(topic_name), std::string(partition_name));
}

template < class TT_ >
DDSTopicWriterClass< TT_ >::~DDSTopicWriterClass()
{
    try {
        this->close();

        DPSingleton::Instance().removeFromCleanupList(this);
    }
    catch ( ... ) {
        std::cerr << "Exception calling close() in in DDSTopicWriterClass."
                  << std::endl;
    }
}


template < class TT_ >
DDSTopicReaderClass< TT_ >::DDSTopicReaderClass( char const * topic_name )
{
#if CETDDS_DEBUG
    std::cout << topic_name << "\n" << "hello\n";
#endif

    init( std::string(topic_name), std::string("unknown"));
}

template < class TT_ >
DDSTopicReaderClass< TT_ >::DDSTopicReaderClass( char const * topic_name, char const * partition_name )
{
#if CETDDS_DEBUG
    std::cout << topic_name << "\n" << "hello\n";
#endif

    init( std::string(topic_name), std::string(partition_name));
}

template < class TT_ >
DDSTopicReaderClass< TT_ >::~DDSTopicReaderClass()
{
    try {
        this->close();

        DPSingleton::Instance().removeFromCleanupList(this);
    }
    catch ( ... ) {
        std::cerr << "Exception calling close() in in DDSTopicReaderClass."
                  << std::endl;
    }
}


template < class TT_ >
void DDSTopicWriterClass< TT_ >::init( std::string const& topic_name,  std::string const& partition_name )
try
{  
    _startupDone = false;
    _hasBeenClosed = false;

    SignalInhibitor sig;
    dds::DomainParticipant dp(DPSingleton::Instance().get_participant());

    typename dds::topic_type_support<TT_>::type topicTS;

    //FIXME:GAL DDSLEAK   
    char * name =topicTS.get_type_name();   
    std::string type_name( name );
    DDS::string_free(name);
    //FIXME:GAL DDSLEAK   
    
    const bool isContentFilteredTopicsEnabled = string2bool(
                novadaq::EnvVarCache::getInstance().getEnvVar("RMS_ENABLE_CONTENT_FILTERED_TOPICS"));

    std::string extended_topic_name( isContentFilteredTopicsEnabled ?
                                     create_content_filtered_topic_name_for(type_name, topic_name ):
                                     create_extended_topic_name_for( type_name, topic_name ) );

    //create topic
    dds::TopicQos topicQoS(dp);

    dp->get_default_topic_qos(topicQoS);

    const bool printTopicName =
        string2bool(novadaq::EnvVarCache::getInstance().getEnvVar("RMS_PRINT_TOPICS"));
    if (printTopicName) {
        std::cout << "Writer topic = " << extended_topic_name
                  << ", partition = " << partition_name << std::endl;
    }
    dds::Topic<TT_> topic(dp, extended_topic_name, type_name, topicQoS );

    //creating publisher
    dds::Publisher publisher (dp, partition_name);

    //create writer
    dds::DataWriterQos writerQoS(dp, topicQoS );
    writerQoS.set_auto_dispose(true);
    // 7/21/15 ELF: Set writerQoS History policy to try and avoid "silently dropped" messages
    writerQoS.history.kind = DDS::KEEP_ALL_HISTORY_QOS;


    writer.reset(new dds::DataWriter<TT_>(publisher,topic, writerQoS ) );

    struct timeval now;
    gettimeofday( &now, NULL );
    _startTime = ( double ) now.tv_sec + ((( double ) now.tv_usec ) / 1000000.0 );
    
    DPSingleton::Instance().addToCleanupList( this );
    
#if CETDDS_DEBUG
    std::cout << "DDSTopicWriterClass creation time = " << startTime << std::endl;
#endif
}
catch (...)
{
    close();
}

template < class TT_ >
void DDSTopicReaderClass< TT_ >::init( std::string const& topic_name,  std::string const& partition_name )
try
{  
    _hasBeenClosed = false;

    SignalInhibitor sig;    
    dds::DomainParticipant dp(DPSingleton::Instance().get_participant());

    typename dds::topic_type_support<TT_>::type topicTS;
    //FIXME:GAL DDSLEAK    
    char * name =topicTS.get_type_name();   
    std::string type_name( name );
    DDS::string_free(name);
    //FIXME:GAL DDSLEAK   
    
    const bool isContentFilteredTopicsEnabled = string2bool(
                novadaq::EnvVarCache::getInstance().getEnvVar("RMS_ENABLE_CONTENT_FILTERED_TOPICS"));

    std::string extended_topic_name( isContentFilteredTopicsEnabled ?
                                     create_content_filtered_topic_name_for(type_name, topic_name ):
                                     create_extended_topic_name_for( type_name, topic_name ) );
    //create topic
    dds::TopicQos topicQoS(dp);
    dp->get_default_topic_qos(topicQoS);

    const bool printTopicName =
        string2bool(novadaq::EnvVarCache::getInstance().getEnvVar("RMS_PRINT_TOPICS"));
    if (printTopicName) {
        std::cout << "Reader topic = " << extended_topic_name
                  << ", partition = " << partition_name << std::endl;
    }

    dds::Topic<TT_> topic(dp, extended_topic_name, type_name, topicQoS);

    //creating subscriber
    dds::Subscriber subscriber(dp, partition_name);

    //create reader
    dds::DataReaderQos readerQoS(dp, topicQoS );

    if ( isContentFilteredTopicsEnabled ) {
        dds_topic_filter_t content_filter;
        create_topic_filter_for(content_filter,type_name, topic_name);

        //each content filtered topic name must be unique within a client and must be different
        //from the topic name given to dds::Topic constructor (undocumented feature)
        dds::ContentFilteredTopic<TT_>content_filtered_topic(dp,unique_topic_name_for(extended_topic_name),
                topic,content_filter.first,content_filter.second);

        reader.reset(new dds::DataReader<TT_>( subscriber, content_filtered_topic, readerQoS ));
    }
    else {
        reader.reset(new dds::DataReader<TT_>( subscriber, topic, readerQoS ));
    }

    dds::ReadCondition read_condition = reader->create_readcondition(
                                            boost::bind(&DDSTopicReaderClass< TT_ >::dummyHandler,this),
                                            DDS::ANY_SAMPLE_STATE,DDS::ANY_VIEW_STATE, DDS::ANY_INSTANCE_STATE );

    /*
    dds::StatusCondition status_condition = reader->create_statuscondition(
      boost::bind(&DDSTopicReaderClass< TT_ >::statusConditionHandler,this,reader),
      DDS::SAMPLE_LOST_STATUS);
    */
    wait_set.reset(new dds::WaitSet);
    wait_set->attach(read_condition);
    //wait_set->attach(status_condition);
    
    DPSingleton::Instance().addToCleanupList( this );    
}
catch (...)
{
    close();
}


template < class TT_ >
void DDSTopicReaderClass< TT_ >::statusConditionHandler(
    boost::shared_ptr< dds::DataReader<TT_> > & reader)
{

    DDS::SampleLostStatus sampleLostStatus;
    dds::ReturnCode_t  status =reader->get_sample_lost_status(sampleLostStatus);
    if (status !=DDS::RETCODE_OK)
    {
#if CETDDS_DEBUG
        std::cout << "DDSTopiReaderClass get_sample_lost_status returned error_code = " << status << std::endl;
#endif
        return;
    }
    if (sampleLostStatus.total_count > 0)
        std::cout << "DDSTopiReaderClass lost data samples total_count=" << sampleLostStatus.total_count
                  << ", total_count_change=" << sampleLostStatus.total_count_change << std::endl;
}

template < class TT_ >
void DDSTopicReaderClass< TT_ >::receiveMessage(
    boost::shared_ptr< dds::DataReader<TT_> > &reader,
    TT_ & topicdata,dds::ReturnCode_t & status)
{
#if CETDDS_DEBUG
    std::cout << "receiveMessage\n";
#endif
    statusConditionHandler(reader);

    const unsigned int seqSize=1;

    typename dds::topic_data_seq<TT_>::type topicSeq(seqSize);
    dds::SampleInfoSeq   infoSeq(seqSize);

    status = reader->take( topicSeq, infoSeq,1,
                           DDS::ANY_SAMPLE_STATE,DDS::ANY_VIEW_STATE,DDS::ANY_INSTANCE_STATE );

    checkStatus( status, "take" );

    if (status==DDS::RETCODE_NO_DATA)
        return;

#if CETDDS_EXTRA_DEBUG
    dumpReceivedSamples( infoSeq, topicSeq.length(), status );
#endif

    if (  infoSeq[0].valid_data ) {
        topicdata = topicSeq[0];
    }
    else {
#if CETDDS_DEBUG
        printf( "handle invalid data\n" );
#endif
        status = DDS::RETCODE_NO_DATA;
    }
}

//  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -
template < class TT_ >
void DDSTopicWriterClass< TT_ >::waitForNetworkServices()
{
    // When DDS applications are communicating over the network, we need
    // to allow time for the daemons to tell each other about the existence
    // of this new writer.  It seems like the
    // NetworkService/Channels/Channel/Resolution configuration parameter
    // is what sets the time between the updates, and this parameter has a
    // default value of 10 msec.  We use a default delay of 20 msec here
    // so that we have a safety factor of 2.0.

    double maxDelay = novadaq::EnvVarCache::getInstance().
                      getEnvVarAsDouble( "RMS_WRITER_STARTUP_DELAY", 0.020 );

    struct timeval now;
    gettimeofday( &now, NULL );
    double currentTime = ( double ) now.tv_sec +
                         ((( double ) now.tv_usec ) / 1000000.0 );
    double neededDelay = maxDelay - ( currentTime - _startTime );

#if CETDDS_DEBUG
    std::cout << "DDSTopicWriterClass startup delay = " << neededDelay << std::endl;
#endif

    if ( neededDelay > 0 ) {
        usleep(( int )( 1000000 * neededDelay ) );
    }

    _startupDone = true;
}

template < class TT_ >
int DDSTopicWriterClass< TT_ >::write( TT_ const& topicdata )
try
{
#if CETDDS_DEBUG
    std::cout << "write\n";
#endif

    if (_hasBeenClosed)
    {
       GENERATE_RMS_NOTCONNECTED_EXCEPTION("DDSTopicWriterClass not connected.");
    }
    else if(this->hasBeenNotified())
    {
      GENERATE_RMS_EXITINGPROCESS_EXCEPTION("Exiting process.");
    }

    if ( ! _startupDone )
        waitForNetworkServices();

    SignalInhibitor sig;
    dds::ReturnCode_t status = writer->write( topicdata );

    if(this->hasBeenNotified())
    {
      GENERATE_RMS_EXITINGPROCESS_EXCEPTION("Exiting process.");
    }
    
    checkStatus( status, "write" );

    return ( 0 );
}
catch (gov::fnal::cd::rms::base::RmsExitingProcessException & ex)
{
    close();
    throw;
}


template < class TT_ >
int DDSTopicReaderClass< TT_ >::take_timed_wait( TT_ & topicdata, dds::Duration_t const& duration )
try
{
#if CETDDS_DEBUG
    std::cout << "take\n";
#endif

    if (_hasBeenClosed)
    {
        GENERATE_RMS_NOTCONNECTED_EXCEPTION("DDSTopicReaderClass not connected.");
    }
    else if(this->hasBeenNotified())
    {
      GENERATE_RMS_EXITINGPROCESS_EXCEPTION("Exiting process.");
    }
    
    // NOTE that this method returns DDS::RETCODE_TIMEOUT when the
    // timeout is reached and no data has been received and
    // DDS::RETCODE_NO_DATA when invalid data is received before
    // the timeout expires.  An alternative model would be to only
    // return early if valid data has been received.

    dds::ReturnCode_t   status=DDS::RETCODE_NO_DATA;
    dds::ReturnCode_t   wait_status=DDS::RETCODE_ERROR;

    try
    {
        SignalInhibitor sig;
        
        while ( wait_status != DDS::RETCODE_TIMEOUT && status == DDS::RETCODE_NO_DATA && !_hasBeenClosed)
        {
            wait_status = wait_set->dispatch(duration);

            if (wait_status==DDS::RETCODE_OK)
                receiveMessage(reader, topicdata, status);
            
            if(this->hasBeenNotified())
            {
              GENERATE_RMS_EXITINGPROCESS_EXCEPTION("Exiting process.");
            }
        }
    }
    catch (dds::InterruptedException& ex)
    {
#if CETDDS_DEBUG1
        std::cerr << "Caught InterruptedException while calling take_timed_wait() in DDSTopicReaderClass; message="
                  << ex.what() << std::endl;
#endif
        close();
    }

    if ( wait_status == DDS::RETCODE_TIMEOUT )
        status =  DDS::RETCODE_TIMEOUT;
    else if (wait_status==DDS::RETCODE_OK && status!=DDS::RETCODE_OK)
        status = DDS::RETCODE_NO_DATA;

    return (( int )status );
}
catch (gov::fnal::cd::rms::base::RmsExitingProcessException & ex)
{
    close();
    throw;
}


template < class TT_ >
int DDSTopicReaderClass< TT_ >::take_timed_wait( TT_ & topicdata, unsigned int usec )
{
    dds::Duration_t   timeout={ usec / 1000000, ( usec - ( usec / 1000000 ) * 1000000 ) * 1000};
    return take_timed_wait( topicdata,timeout );
}

template < class TT_ >
int DDSTopicReaderClass< TT_ >::take_wait( TT_ & topicdata )
{
    dds::Duration_t   timeout=DDS::DURATION_INFINITE;
    return take_timed_wait( topicdata,timeout );
}

template < class TT_ >
int DDSTopicReaderClass< TT_ >::take_nowait( TT_ & topicdata )
{
    dds::Duration_t   timeout=DDS::DURATION_ZERO;
    return take_timed_wait( topicdata,timeout );
}

//  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -  -

template < class TT_ >
void DDSTopicWriterClass< TT_ >::close() throw()
{
    _hasBeenClosed = true;
    SignalInhibitor sig;
    writer.reset();
#if CETDDS_DEBUG1
    std::cout << "Closed DDSTopicWriter connection." << std::endl;
#endif
}


template < class TT_ >
void DDSTopicReaderClass< TT_ >::close() throw()
{
  _hasBeenClosed = true;
  SignalInhibitor sig;
  wait_set.reset();
  reader.reset();
#if CETDDS_DEBUG1
    std::cout << "Closed DDSTopicReader connection." << std::endl;
#endif
}

template < class TT_ >
    bool DDSTopicWriterClass< TT_ >::hasBeenClosed() throw()
{
  return _hasBeenClosed;
}

template < class TT_ >
    bool DDSTopicReaderClass< TT_ >::hasBeenClosed() throw()
{
  return _hasBeenClosed;
}

}  // end of namespace provider

}  // end of namespace rms

}  // end of namespace cd

}  // end of namespace fnal

}  // end of namespace gov


/*
 *  With "tab-width: 4" tabs and spaces could get mixed (especially in
 *  environments where c-backspace-function is backward-delete-char-untabify)
 *  which could make things a mess in other "tab-width" environments. (Always
 *  tabifying would work, but require a lot of discipline.)
 *  Local variables:
 *    mode: c++
 *    tab-width: 8
 *    c-tab-always-indent: nil
 *    c-basic-offset: 4
 *    c-file-offsets: ((substatement-open . 0)
                       (statement         . c-lineup-runin-statements)
                       (label             . -1000)
                       (arglist-intro     . c-lineup-arglist-intro-after-paren)
                       (arglist-cont      . c-lineup-arglist-intro-after-paren)
                       (arglist-cont-nonempty . c-lineup-arglist-intro-after-paren)
                       (arglist-close     . c-lineup-arglist-intro-after-paren)
                      )
 *  End:
 */

#endif
// kate: indent-mode cstyle; space-indent on; indent-width 0;