DDSSubscriber.h

Go to the documentation of this file.

#ifndef _DDS_SUBSCRIBER_H
#define _DDS_SUBSCRIBER_H

#include <DDSHelper.h>
#include <acsddsncDataReaderListener.h>
#include <dds/DCPS/SubscriberImpl.h>

namespace ddsnc{

        class DDSSubscriber : public ddsnc::DDSHelper{
                private:
                DDS::Subscriber_var sub;
                OpenDDS::DCPS::SubscriberImpl *sub_impl;
                DDS::DataReaderListener_var *listener;
                
                int attachToTransport();

                int createSubscriber();

                protected:
                DDS::SubscriberQos subQos;

                public:
                DDS::DataReaderQos drQos; 
                DDSSubscriber(CORBA::String_var channel_name);
                
                ~DDSSubscriber()
                {
                        delete listener;
                }

                void consumerReady();

                template <class DRV, class DR, class D>
                        void addSubscription(
                                        void (*templateFunction)(D, void *), void *handlerParam=0)
                        {
                                std::cerr << "DDSSubscriber::addSubscription" << std::endl;

                                listener = new  DDS::DataReaderListener_var 
                                        (new ddsnc::ACSDDSNCDataReaderListener
                                         <DRV,DR,D>(templateFunction, handlerParam));

                                /*ddsnc::ACSDDSNCDataReaderListener<DRV,DR,D>* listener_servant=
                                        dynamic_cast<ddsnc::ACSDDSNCDataReaderListener<DRV,DR,D>*>
                                        (listener.in());

                                if(CORBA::is_nil (listener.in())){
                                        std::cerr << "listener is nil" << std::endl;
                                }
                                */
                        }


                template <class D, class TSV, class TSI>
                void initialize()
                {
                        std::cerr<< "DDSSubscriber::initialize()" << std::endl;
                        createParticipant();
                        if (CORBA::is_nil (participant.in()))
                                 std::cerr << "Participant is nil" << std::endl;
                
                        if(partitionName!=NULL){
                                participant->get_default_subscriber_qos(subQos);
                                subQos.partition.name.length(1);
                                subQos.partition.name[0]=CORBA::string_dup(partitionName);
                        }
                        initializeTransport();
                
                        createSubscriber();
                        
                        /*Initialize Type Support*/
                        TSV ts;
                        ts = new TSI();
                        if (DDS::RETCODE_OK != ts->register_type(participant.in(),"")){
                                std::cerr << "register_type failed" << std::endl;
                        }
                        
                        /*Initialize the Topic*/
                        initializeTopic(ts->get_type_name());
                        if(CORBA::is_nil(topic.in()))
                                std::cerr<< "Topic is nil" << std::endl;
                
                        sub->get_default_datareader_qos (drQos);
                        
                        drQos.reliability.kind = ::DDS::RELIABLE_RELIABILITY_QOS;
                        drQos.reliability.max_blocking_time.sec = 1;

                        drQos.history.kind = ::DDS::KEEP_LAST_HISTORY_QOS;
                        // drQos.history.depth = 100;
                        drQos.history.depth = 1;
                }
        };
}

#define ACS_NEW_DDS_SUBSCRIBER(subscriber_p, idlStruct, channelName, handlerFunc, handlerParam) \
{ \
        subscriber_p= new ddsnc::DDSSubscriber(channelName); \
        subscriber_p->initialize<idlStruct, idlStruct##TypeSupport_var, idlStruct##TypeSupportImpl>(); \
        subscriber_p->addSubscription<idlStruct##DataReader_var, idlStruct##DataReader, idlStruct>(handlerFunc, handlerParam); \
}

#endif