<commit_before>// 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 RTPSParticipant.cpp
 *
 */

#include "RTPSParticipantImpl.h"

#include <fastrtps/rtps/resources/ResourceSend.h>
#include <fastrtps/rtps/resources/ResourceEvent.h>
#include <fastrtps/rtps/resources/AsyncWriterThread.h>
#include <fastrtps/rtps/resources/ListenResource.h>

#include <fastrtps/rtps/messages/MessageReceiver.h>

#include <fastrtps/rtps/writer/StatelessWriter.h>
#include <fastrtps/rtps/writer/StatefulWriter.h>

#include <fastrtps/rtps/reader/StatelessReader.h>
#include <fastrtps/rtps/reader/StatefulReader.h>
#include <fastrtps/rtps/reader/StatefulReader.h>

#include <fastrtps/rtps/participant/RTPSParticipant.h>
#include <fastrtps/transport/UDPv4Transport.h>

#include <fastrtps/rtps/RTPSDomain.h>

#include <fastrtps/rtps/builtin/BuiltinProtocols.h>
#include <fastrtps/rtps/builtin/discovery/participant/PDPSimple.h>
#include <fastrtps/rtps/builtin/data/ParticipantProxyData.h>

#include <fastrtps/utils/IPFinder.h>
#include <fastrtps/utils/eClock.h>

#include <boost/thread.hpp>
#include <boost/interprocess/sync/interprocess_semaphore.hpp>
#include <boost/thread/recursive_mutex.hpp>
#include <boost/thread/lock_guard.hpp>

#include <fastrtps/log/Log.h>



namespace eprosima {
namespace fastrtps{
namespace rtps {



static EntityId_t TrustedWriter(const EntityId_t& reader)
{
    if(reader == c_EntityId_SPDPReader) return c_EntityId_SPDPWriter;
    if(reader == c_EntityId_SEDPPubReader) return c_EntityId_SEDPPubWriter;
    if(reader == c_EntityId_SEDPSubReader) return c_EntityId_SEDPSubWriter;
    if(reader == c_EntityId_ReaderLiveliness) return c_EntityId_WriterLiveliness;

    return c_EntityId_Unknown;
}

Locator_t RTPSParticipantImpl::applyLocatorAdaptRule(Locator_t loc)
{
    switch (loc.kind){
        case LOCATOR_KIND_UDPv4:
            //This is a completely made up rule
            loc.port += 2;
            break;
        case LOCATOR_KIND_UDPv6:
            //TODO - Define the rest of rules
            loc.port += 2;
            break;
    }
    return loc;
}

RTPSParticipantImpl::RTPSParticipantImpl(const RTPSParticipantAttributes& PParam,
        const GuidPrefix_t& guidP,
        RTPSParticipant* par,
        RTPSParticipantListener* plisten): m_att(PParam), m_guid(guidP ,c_EntityId_RTPSParticipant),
    mp_event_thr(nullptr),
    mp_builtinProtocols(nullptr),
    mp_ResourceSemaphore(new boost::interprocess::interprocess_semaphore(0)),
    IdCounter(0),
#if HAVE_SECURITY
    m_security_manager(this),
#endif
    mp_participantListener(plisten),
    mp_userParticipant(par),
    mp_mutex(new boost::recursive_mutex()),
    is_rtps_protected_(false)
{
    // Read participant properties.
    const std::string* property_value = PropertyPolicyHelper::find_property(PParam.properties,
            "rtps.participant.rtps_protection_kind");
    if(property_value != nullptr && property_value->compare("ENCRYPT") == 0)
        is_rtps_protected_ = true;

    // Builtin transport by default
    if (PParam.useBuiltinTransports)
    {
        UDPv4TransportDescriptor descriptor; 
        descriptor.sendBufferSize = m_att.sendSocketBufferSize;
        descriptor.receiveBufferSize = m_att.listenSocketBufferSize;
        m_network_Factory.RegisterTransport(&descriptor);
    }

    // User defined transports
    for (const auto& transportDescriptor : PParam.userTransports)
        m_network_Factory.RegisterTransport(transportDescriptor.get());

    boost::lock_guard<boost::recursive_mutex> guard(*mp_mutex);
    mp_userParticipant->mp_impl = this;
    Locator_t loc;
    loc.port = PParam.defaultSendPort;
    mp_event_thr = new ResourceEvent();
    mp_event_thr->init_thread(this);


    // Throughput controller, if the descriptor has valid values
    if (PParam.throughputController.bytesPerPeriod != UINT32_MAX &&
            PParam.throughputController.periodMillisecs != 0)
    {
        std::unique_ptr<FlowController> controller(new ThroughputController(PParam.throughputController, this));
        m_controllers.push_back(std::move(controller));
    }

    /// Creation of metatraffic locator and receiver resources
    uint32_t metatraffic_multicast_port = m_att.port.getMulticastPort(m_att.builtin.domainId);
    uint32_t metatraffic_unicast_port = m_att.port.getUnicastPort(m_att.builtin.domainId, m_att.participantID);

    /* If metatrafficMulticastLocatorList is empty, add mandatory default Locators
       Else -> Take them */

    /* INSERT DEFAULT MANDATORY MULTICAST LOCATORS HERE */

    //UDPv4
    if(m_att.builtin.metatrafficMulticastLocatorList.empty())
    {
        Locator_t mandatoryMulticastLocator;
        mandatoryMulticastLocator.kind = LOCATOR_KIND_UDPv4;
        mandatoryMulticastLocator.port = metatraffic_multicast_port;
        mandatoryMulticastLocator.set_IP4_address(239,255,0,1);
        m_att.builtin.metatrafficMulticastLocatorList.push_back(mandatoryMulticastLocator);
    }
    else
    {
        //Copy metatrafficMulticastLocatorList from the BuiltinAttributs
        for(std::vector<Locator_t>::iterator it = m_att.builtin.metatrafficMulticastLocatorList.begin();
                it!=m_att.builtin.metatrafficMulticastLocatorList.end();++it)
        {
            m_att.builtin.metatrafficMulticastLocatorList.push_back(*it);
        }
    }
    //Create ReceiverResources now and update the list with the REAL used ones
    createReceiverResources(m_att.builtin.metatrafficMulticastLocatorList, true);
    /* INSERT DEFAULT UNICAST LOCATORS HERE */

    if(m_att.builtin.metatrafficUnicastLocatorList.empty())
    {
        //Add default metatrafficUnicastLocators
        LocatorList_t locators;
        IPFinder::getIP4Address(&locators);
        for(std::vector<Locator_t>::iterator it=locators.begin();it!=locators.end();++it)
        {
            it->port = metatraffic_unicast_port;
            m_att.builtin.metatrafficUnicastLocatorList.push_back(*it);
        }
    }
    else
    {
        //If locators existed, just import them to the class
        for(std::vector<Locator_t>::iterator it = m_att.builtin.metatrafficUnicastLocatorList.begin();
                it!=m_att.builtin.metatrafficUnicastLocatorList.end();++it)
        {
            m_att.builtin.metatrafficUnicastLocatorList.push_back(*it);
        }
    }
    //Create ReceiverResources now and update the list with the REAL used ones
    createReceiverResources(m_att.builtin.metatrafficUnicastLocatorList, true);

    /// Creation of user locator and receiver resources

    bool hasLocatorsDefined = true;
    //If no default locators are defined we define some.
    /* The reasoning here is the following.
       If the parameters of the RTPS Participant don't hold default listening locators for the creation
       of Endpoints, we make some for Unicast only.
       If there is at least one listen locator of any kind, we do not create any default ones.
       If there are no sending locators defined, we create default ones for the transports we implement.
       */
    if(m_att.defaultUnicastLocatorList.empty() && m_att.defaultMulticastLocatorList.empty())
    {
        //Default Unicast Locators in case they have not been provided
        /* INSERT DEFAULT UNICAST LOCATORS FOR THE PARTICIPANT */
        hasLocatorsDefined = false;
        Locator_t loc2;

        LocatorList_t loclist;
        IPFinder::getIP4Address(&loclist);
        for(auto it=loclist.begin();it!=loclist.end();++it){
            (*it).port=m_att.port.portBase+
                m_att.port.domainIDGain*PParam.builtin.domainId+
                m_att.port.offsetd3+
                m_att.port.participantIDGain*m_att.participantID;
            (*it).kind = LOCATOR_KIND_UDPv4;

            m_att.defaultUnicastLocatorList.push_back((*it));
        }
        // FIXME -- We have to  discuss the rules for deafult locator assignment for each transport
        loc2.port= m_att.port.portBase+
            m_att.port.domainIDGain*PParam.builtin.domainId+
            m_att.port.offsetd2+
            m_att.port.participantIDGain*m_att.participantID;
        loc2.set_IP4_address(239,255,1,4);
        m_att.defaultMulticastLocatorList.push_back(loc2);
        /* INSERT DEFAULT MULTICAST LOCATORS FOR THE PARTICIPANT */
    }
    else
    {
        // Normalize unicast locators.
        m_network_Factory.NormalizeLocators(m_att.defaultUnicastLocatorList);
    }

    /*	
        Since nothing guarantees the correct creation of the Resources on the Locators we have specified, and 
        in order to maintain synchrony between the defaultLocator list and the actuar ReceiveResources,
        We create the resources for these Locators now. Furthermore, in case these resources are taken, 
        we create them on another Locator and then update de defaultList.
        */
    createReceiverResources(m_att.defaultUnicastLocatorList, true);

    if(!hasLocatorsDefined){
        logInfo(RTPS_PARTICIPANT,m_att.getName()<<" Created with NO default Unicast Locator List, adding Locators: "<<m_att.defaultUnicastLocatorList);
    }
    //Multicast
    createReceiverResources(m_att.defaultMulticastLocatorList, true);<commit_msg>Refs #1966. Remove default multicast locator.<commit_after>// 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 RTPSParticipant.cpp
 *
 */

#include "RTPSParticipantImpl.h"

#include <fastrtps/rtps/resources/ResourceSend.h>
#include <fastrtps/rtps/resources/ResourceEvent.h>
#include <fastrtps/rtps/resources/AsyncWriterThread.h>
#include <fastrtps/rtps/resources/ListenResource.h>

#include <fastrtps/rtps/messages/MessageReceiver.h>

#include <fastrtps/rtps/writer/StatelessWriter.h>
#include <fastrtps/rtps/writer/StatefulWriter.h>

#include <fastrtps/rtps/reader/StatelessReader.h>
#include <fastrtps/rtps/reader/StatefulReader.h>
#include <fastrtps/rtps/reader/StatefulReader.h>

#include <fastrtps/rtps/participant/RTPSParticipant.h>
#include <fastrtps/transport/UDPv4Transport.h>

#include <fastrtps/rtps/RTPSDomain.h>

#include <fastrtps/rtps/builtin/BuiltinProtocols.h>
#include <fastrtps/rtps/builtin/discovery/participant/PDPSimple.h>
#include <fastrtps/rtps/builtin/data/ParticipantProxyData.h>

#include <fastrtps/utils/IPFinder.h>
#include <fastrtps/utils/eClock.h>

#include <boost/thread.hpp>
#include <boost/interprocess/sync/interprocess_semaphore.hpp>
#include <boost/thread/recursive_mutex.hpp>
#include <boost/thread/lock_guard.hpp>

#include <fastrtps/log/Log.h>



namespace eprosima {
namespace fastrtps{
namespace rtps {



static EntityId_t TrustedWriter(const EntityId_t& reader)
{
    if(reader == c_EntityId_SPDPReader) return c_EntityId_SPDPWriter;
    if(reader == c_EntityId_SEDPPubReader) return c_EntityId_SEDPPubWriter;
    if(reader == c_EntityId_SEDPSubReader) return c_EntityId_SEDPSubWriter;
    if(reader == c_EntityId_ReaderLiveliness) return c_EntityId_WriterLiveliness;

    return c_EntityId_Unknown;
}

Locator_t RTPSParticipantImpl::applyLocatorAdaptRule(Locator_t loc)
{
    switch (loc.kind){
        case LOCATOR_KIND_UDPv4:
            //This is a completely made up rule
            loc.port += 2;
            break;
        case LOCATOR_KIND_UDPv6:
            //TODO - Define the rest of rules
            loc.port += 2;
            break;
    }
    return loc;
}

RTPSParticipantImpl::RTPSParticipantImpl(const RTPSParticipantAttributes& PParam,
        const GuidPrefix_t& guidP,
        RTPSParticipant* par,
        RTPSParticipantListener* plisten): m_att(PParam), m_guid(guidP ,c_EntityId_RTPSParticipant),
    mp_event_thr(nullptr),
    mp_builtinProtocols(nullptr),
    mp_ResourceSemaphore(new boost::interprocess::interprocess_semaphore(0)),
    IdCounter(0),
#if HAVE_SECURITY
    m_security_manager(this),
#endif
    mp_participantListener(plisten),
    mp_userParticipant(par),
    mp_mutex(new boost::recursive_mutex()),
    is_rtps_protected_(false)
{
    // Read participant properties.
    const std::string* property_value = PropertyPolicyHelper::find_property(PParam.properties,
            "rtps.participant.rtps_protection_kind");
    if(property_value != nullptr && property_value->compare("ENCRYPT") == 0)
        is_rtps_protected_ = true;

    // Builtin transport by default
    if (PParam.useBuiltinTransports)
    {
        UDPv4TransportDescriptor descriptor; 
        descriptor.sendBufferSize = m_att.sendSocketBufferSize;
        descriptor.receiveBufferSize = m_att.listenSocketBufferSize;
        m_network_Factory.RegisterTransport(&descriptor);
    }

    // User defined transports
    for (const auto& transportDescriptor : PParam.userTransports)
        m_network_Factory.RegisterTransport(transportDescriptor.get());

    boost::lock_guard<boost::recursive_mutex> guard(*mp_mutex);
    mp_userParticipant->mp_impl = this;
    Locator_t loc;
    loc.port = PParam.defaultSendPort;
    mp_event_thr = new ResourceEvent();
    mp_event_thr->init_thread(this);


    // Throughput controller, if the descriptor has valid values
    if (PParam.throughputController.bytesPerPeriod != UINT32_MAX &&
            PParam.throughputController.periodMillisecs != 0)
    {
        std::unique_ptr<FlowController> controller(new ThroughputController(PParam.throughputController, this));
        m_controllers.push_back(std::move(controller));
    }

    /// Creation of metatraffic locator and receiver resources
    uint32_t metatraffic_multicast_port = m_att.port.getMulticastPort(m_att.builtin.domainId);
    uint32_t metatraffic_unicast_port = m_att.port.getUnicastPort(m_att.builtin.domainId, m_att.participantID);

    /* If metatrafficMulticastLocatorList is empty, add mandatory default Locators
       Else -> Take them */

    /* INSERT DEFAULT MANDATORY MULTICAST LOCATORS HERE */

    //UDPv4
    if(m_att.builtin.metatrafficMulticastLocatorList.empty())
    {
        Locator_t mandatoryMulticastLocator;
        mandatoryMulticastLocator.kind = LOCATOR_KIND_UDPv4;
        mandatoryMulticastLocator.port = metatraffic_multicast_port;
        mandatoryMulticastLocator.set_IP4_address(239,255,0,1);
        m_att.builtin.metatrafficMulticastLocatorList.push_back(mandatoryMulticastLocator);
    }
    else
    {
        //Copy metatrafficMulticastLocatorList from the BuiltinAttributs
        for(std::vector<Locator_t>::iterator it = m_att.builtin.metatrafficMulticastLocatorList.begin();
                it!=m_att.builtin.metatrafficMulticastLocatorList.end();++it)
        {
            m_att.builtin.metatrafficMulticastLocatorList.push_back(*it);
        }
    }
    //Create ReceiverResources now and update the list with the REAL used ones
    createReceiverResources(m_att.builtin.metatrafficMulticastLocatorList, true);
    /* INSERT DEFAULT UNICAST LOCATORS HERE */

    if(m_att.builtin.metatrafficUnicastLocatorList.empty())
    {
        //Add default metatrafficUnicastLocators
        LocatorList_t locators;
        IPFinder::getIP4Address(&locators);
        for(std::vector<Locator_t>::iterator it=locators.begin();it!=locators.end();++it)
        {
            it->port = metatraffic_unicast_port;
            m_att.builtin.metatrafficUnicastLocatorList.push_back(*it);
        }
    }
    else
    {
        //If locators existed, just import them to the class
        for(std::vector<Locator_t>::iterator it = m_att.builtin.metatrafficUnicastLocatorList.begin();
                it!=m_att.builtin.metatrafficUnicastLocatorList.end();++it)
        {
            m_att.builtin.metatrafficUnicastLocatorList.push_back(*it);
        }
    }
    //Create ReceiverResources now and update the list with the REAL used ones
    createReceiverResources(m_att.builtin.metatrafficUnicastLocatorList, true);

    /// Creation of user locator and receiver resources

    bool hasLocatorsDefined = true;
    //If no default locators are defined we define some.
    /* The reasoning here is the following.
       If the parameters of the RTPS Participant don't hold default listening locators for the creation
       of Endpoints, we make some for Unicast only.
       If there is at least one listen locator of any kind, we do not create any default ones.
       If there are no sending locators defined, we create default ones for the transports we implement.
       */
    if(m_att.defaultUnicastLocatorList.empty() && m_att.defaultMulticastLocatorList.empty())
    {
        //Default Unicast Locators in case they have not been provided
        /* INSERT DEFAULT UNICAST LOCATORS FOR THE PARTICIPANT */
        hasLocatorsDefined = false;
        Locator_t loc2;

        LocatorList_t loclist;
        IPFinder::getIP4Address(&loclist);
        for(auto it=loclist.begin();it!=loclist.end();++it){
            (*it).port=m_att.port.portBase+
                m_att.port.domainIDGain*PParam.builtin.domainId+
                m_att.port.offsetd3+
                m_att.port.participantIDGain*m_att.participantID;
            (*it).kind = LOCATOR_KIND_UDPv4;

            m_att.defaultUnicastLocatorList.push_back((*it));
        }
    }
    else
    {
        // Normalize unicast locators.
        m_network_Factory.NormalizeLocators(m_att.defaultUnicastLocatorList);
    }

    /*	
        Since nothing guarantees the correct creation of the Resources on the Locators we have specified, and 
        in order to maintain synchrony between the defaultLocator list and the actuar ReceiveResources,
        We create the resources for these Locators now. Furthermore, in case these resources are taken, 
        we create them on another Locator and then update de defaultList.
        */
    createReceiverResources(m_att.defaultUnicastLocatorList, true);

    if(!hasLocatorsDefined){
        logInfo(RTPS_PARTICIPANT,m_att.getName()<<" Created with NO default Unicast Locator List, adding Locators: "<<m_att.defaultUnicastLocatorList);
    }
    //Multicast
    createReceiverResources(m_att.defaultMulticastLocatorList, true);
{'fraction_non_alphanumeric': 0.09037534074229399, 'fraction_numerical': 0.005084923464038583, 'mean_word_length': 3.5638755980861245, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6030916', 'n_tokens_mistral': 6132, 'n_tokens_neox': 5921, 'n_words': 1224}