modularization/src1/common/modulecomm/fastrtps/TopicsPublisher.cxx

// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

/*!
 * @file TopicsPublisher.cpp
 * This file contains the implementation of the publisher functions.
 *
 * This file was generated by the tool fastcdrgen.
 */

#ifdef USE_FASTRTPS
#include <fastrtps/participant/Participant.h>
#include <fastrtps/attributes/ParticipantAttributes.h>
#include <fastrtps/publisher/Publisher.h>
#include <fastrtps/attributes/PublisherAttributes.h>

#include <fastrtps/Domain.h>

#include <fastrtps/participant/Participant.h>
#include <fastrtps/attributes/ParticipantAttributes.h>
#include <fastrtps/attributes/PublisherAttributes.h>
#include <fastrtps/publisher/Publisher.h>
#include <fastrtps/Domain.h>
#include <fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.h>
#include <fastdds/rtps/transport/UDPv4TransportDescriptor.h>
#include <fastdds/rtps/transport/TCPv4TransportDescriptor.h>

#include <thread>
#include <chrono>

#include "TopicsPublisher.h"

using namespace eprosima::fastrtps;
using namespace eprosima::fastrtps::rtps;

using namespace eprosima::fastdds::rtps;

TopicsPublisher::TopicsPublisher() : mp_participant(nullptr), mp_publisher(nullptr) {}

TopicsPublisher::~TopicsPublisher() {	Domain::removeParticipant(mp_participant);}

bool TopicsPublisher::init(const char * strtopic,const unsigned short nListenPort,const int ntype )
{
    strncpy(mstrtopic,strtopic,255);
    // Create RTPSParticipant

    ParticipantAttributes PParam;
    PParam.rtps.sendSocketBufferSize = 100000000;
    PParam.rtps.setName("Participant_publisher");  //You can put here the name you want

    PParam.rtps.builtin.discovery_config.discoveryProtocol = DiscoveryProtocol_t::SIMPLE;
    PParam.rtps.builtin.discovery_config.use_SIMPLE_EndpointDiscoveryProtocol = true;
    PParam.rtps.builtin.discovery_config.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter = true;
    PParam.rtps.builtin.discovery_config.m_simpleEDP.use_PublicationWriterANDSubscriptionReader = true;
    PParam.rtps.builtin.discovery_config.leaseDuration = c_TimeInfinite;

    // SharedMem transport configuration
    PParam.rtps.useBuiltinTransports = false;

    if(ntype == 0)
    {
    auto shm_transport = std::make_shared<SharedMemTransportDescriptor>();
    shm_transport->segment_size(100*1024*1024);
    PParam.rtps.userTransports.push_back(shm_transport);

    // UDP
    auto udp_transport = std::make_shared<UDPv4TransportDescriptor>();
    //udp_transport->interfaceWhiteList.push_back("127.0.0.1");
    PParam.rtps.userTransports.push_back(udp_transport);


    }
    else
    {
        //Create a descriptor for the new transport.
        auto tcp_transport = std::make_shared<TCPv4TransportDescriptor>();
        tcp_transport->add_listener_port(nListenPort);
        tcp_transport->set_WAN_address("0.0.0.0");

        tcp_transport->wait_for_tcp_negotiation = false;
        tcp_transport->sendBufferSize = 0;
        tcp_transport->receiveBufferSize = 0;

        PParam.rtps.userTransports.push_back(tcp_transport);
    }
    mp_participant = Domain::createParticipant(PParam);
    if(mp_participant == nullptr)
    {
        return false;
    }

    //Register the type

    Domain::registerType(mp_participant, static_cast<TopicDataType*>(&myType));

    // Create Publisher

    PublisherAttributes Wparam;
    Wparam.topic.topicKind = NO_KEY;
    Wparam.topic.topicDataType = myType.getName();  //This type MUST be registered
    Wparam.topic.topicName = strtopic;//"TopicsPubSubTopic";

    Wparam.topic.historyQos.depth = 30;
    Wparam.topic.resourceLimitsQos.max_samples = 50;
    Wparam.topic.resourceLimitsQos.allocated_samples = 20;
    Wparam.times.heartbeatPeriod.seconds = 2;
    Wparam.times.heartbeatPeriod.nanosec = 200 * 1000 * 1000;
    Wparam.qos.m_reliability.kind = RELIABLE_RELIABILITY_QOS;
    Wparam.qos.m_publishMode.kind = ASYNCHRONOUS_PUBLISH_MODE;

    mp_publisher = Domain::createPublisher(mp_participant,Wparam,static_cast<PublisherListener*>(&m_listener));

    if(mp_publisher == nullptr)
    {
        return false;
    }

    std::cout << "Publisher created, waiting for Subscribers." << std::endl;
    return true;
}

void TopicsPublisher::PubListener::onPublicationMatched(Publisher* pub,MatchingInfo& info)
{
    (void)pub;

    if (info.status == MATCHED_MATCHING)
    {
        n_matched++;
        std::cout << "Publisher matched" << std::endl;
    }
    else
    {
        n_matched--;
        std::cout << "Publisher unmatched" << std::endl;
    }
}

void TopicsPublisher::run()
{
    while(m_listener.n_matched == 0)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(250)); // Sleep 250 ms
    }

    // Publication code

    TopicSample::Message st;

    /* Initialize your structure here */

    int msgsent = 0;
    char ch = 'y';
    do
    {
        if(ch == 'y')
        {
            mp_publisher->write(&st);  ++msgsent;
            std::cout << "Sending sample, count=" << msgsent << ", send another sample?(y-yes,n-stop): ";
        }
        else if(ch == 'n')
        {
            std::cout << "Stopping execution " << std::endl;
            break;
        }
        else
        {
            std::cout << "Command " << ch << " not recognized, please enter \"y/n\":";
        }
    } while(std::cin >> ch);
}

#include <QDateTime>

void TopicsPublisher::senddata(const char *str, int nsize)
{


    static int ncount = 1;
    std::cout<<"send data."<<std::endl;
//    while(m_listener.n_matched == 0)
    TopicSample::SomeBytes x;
    x.resize(nsize);
    memcpy(x.data(),str,nsize);

    TopicSample::Message st;

    st.msgname(mstrtopic);
//    st.msgname("topictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopictopic");
    st.counter(ncount);ncount++;
    st.sendtime(QDateTime::currentMSecsSinceEpoch());
    TopicSample::SomeBytes & px = x;
    st.xdata(x);

    int ndatasize = TopicSample::Message::getCdrSerializedSize(st);

    std::cout<<"size is "<<ndatasize<<std::endl;


    mp_publisher->write(&st);



}

#endif