<commit_before>#include "rmw/rmw.h"
#include "rmw/error_handling.h"
#include "rosidl_typesupport_introspection_cpp/identifier.hpp"
#include "rmw_fastrtps_cpp/MessageTypeSupport.h"
#include "rmw_fastrtps_cpp/ServiceTypeSupport.h"

#include <fastrtps/Domain.h>
#include <fastrtps/participant/Participant.h>
#include <fastrtps/attributes/ParticipantAttributes.h>
#include <fastrtps/publisher/Publisher.h>
#include <fastrtps/attributes/PublisherAttributes.h>
#include <fastrtps/subscriber/Subscriber.h>
#include <fastrtps/subscriber/SubscriberListener.h>
#include <fastrtps/subscriber/SampleInfo.h>
#include <fastrtps/attributes/SubscriberAttributes.h>

#include <rpcdds/transports/dds/RTPSProxyTransport.h>
#include <rpcdds/transports/dds/RTPSServerTransport.h>
#include <rpcdds/transports/dds/components/RTPSProxyProcedureEndpoint.h>
#include <rpcdds/transports/dds/components/RTPSServerProcedureEndpoint.h>
#include <rpcdds/strategies/SingleThreadStrategy.h>
#include <rpcdds/transports/dds/RTPSAsyncTask.h>

#include <cassert>
#include <mutex>
#include <condition_variable>
#include <list>

using namespace eprosima::fastrtps;

class ClientListener;

typedef struct CustomClientInfo
{
    eprosima::rpc::transport::dds::RTPSProxyTransport *transport_;
    rmw_fastrtps_cpp::RequestTypeSupport *request_type_support_;
    rmw_fastrtps_cpp::ResponseTypeSupport *response_type_support_;
    eprosima::rpc::transport::dds::RTPSProxyProcedureEndpoint *topic_endpoint_;
    ClientListener *listener_;
} CustomClientInfo;

class ClientListener
{
    public:

        ClientListener(CustomClientInfo *info) : info_(info),
        conditionMutex_(NULL), conditionVariable_(NULL) {}


        void onNewResponse(rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer *buffer)
        {
            assert(buffer);
            std::lock_guard<std::mutex> lock(internalMutex_);

            if(conditionMutex_ != NULL)
            {
                std::unique_lock<std::mutex> clock(*conditionMutex_);
                list.push_back(buffer);
                clock.unlock();
                conditionVariable_->notify_one();
            }
            else
                list.push_back(buffer);

        }

        rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer* getResponse()
        {
            std::lock_guard<std::mutex> lock(internalMutex_);
            rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer *buffer = nullptr;

            if(conditionMutex_ != NULL)
            {
                std::unique_lock<std::mutex> clock(*conditionMutex_);
                buffer = list.front();
                list.pop_front();
            }
            else
            {
                buffer = list.front();
                list.pop_front();
            }

            return buffer;
        }

        void attachCondition(std::mutex *conditionMutex, std::condition_variable *conditionVariable)
        {
            std::lock_guard<std::mutex> lock(internalMutex_);
            conditionMutex_ = conditionMutex;
            conditionVariable_ = conditionVariable;
        }

        void dettachCondition()
        {
            std::lock_guard<std::mutex> lock(internalMutex_);
            conditionMutex_ = NULL;
            conditionVariable_ = NULL;
        }

        bool hasData()
        {
            return !list.empty();
        }

    private:

        CustomClientInfo *info_;
        std::mutex internalMutex_;
        std::list<rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer*> list;
        std::mutex *conditionMutex_;
        std::condition_variable *conditionVariable_;
};


class AsyncTask : public eprosima::rpc::transport::dds::RTPSAsyncTask
{
    public:

        AsyncTask(CustomClientInfo *info) :
            info_(info),
            buffer_(static_cast<rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer*>(info->response_type_support_->createData()))
    {
    }

        virtual ~AsyncTask(){}

        virtual void execute()
        {
            info_->listener_->onNewResponse(buffer_);
        }

        virtual void on_exception(const eprosima::rpc::exception::SystemException &ex){}

        virtual void* getReplyInstance() { return buffer_; }

    private:

        CustomClientInfo *info_;

        rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer *buffer_;
};

extern "C"
{
    const char* const eprosima_fastrtps_identifier = "fastrtps";

    const char* rmw_get_implementation_identifier()
    {
        return eprosima_fastrtps_identifier;
    }

    rmw_ret_t rmw_init()
    {
        return RMW_RET_OK;
    }

    rmw_node_t* rmw_create_node(const char *name)
    {
        assert(name);

        ParticipantAttributes participantParam;
        participantParam.rtps.builtin.domainId = 0;
        participantParam.rtps.builtin.leaseDuration = c_TimeInfinite;
        participantParam.rtps.setName(name);

        Participant *participant = Domain::createParticipant(participantParam);

        if(!participant)<commit_msg>add domain id to node creation<commit_after>#include "rmw/rmw.h"
#include "rmw/error_handling.h"
#include "rosidl_typesupport_introspection_cpp/identifier.hpp"
#include "rmw_fastrtps_cpp/MessageTypeSupport.h"
#include "rmw_fastrtps_cpp/ServiceTypeSupport.h"

#include <fastrtps/Domain.h>
#include <fastrtps/participant/Participant.h>
#include <fastrtps/attributes/ParticipantAttributes.h>
#include <fastrtps/publisher/Publisher.h>
#include <fastrtps/attributes/PublisherAttributes.h>
#include <fastrtps/subscriber/Subscriber.h>
#include <fastrtps/subscriber/SubscriberListener.h>
#include <fastrtps/subscriber/SampleInfo.h>
#include <fastrtps/attributes/SubscriberAttributes.h>

#include <rpcdds/transports/dds/RTPSProxyTransport.h>
#include <rpcdds/transports/dds/RTPSServerTransport.h>
#include <rpcdds/transports/dds/components/RTPSProxyProcedureEndpoint.h>
#include <rpcdds/transports/dds/components/RTPSServerProcedureEndpoint.h>
#include <rpcdds/strategies/SingleThreadStrategy.h>
#include <rpcdds/transports/dds/RTPSAsyncTask.h>

#include <cassert>
#include <mutex>
#include <condition_variable>
#include <list>

using namespace eprosima::fastrtps;

class ClientListener;

typedef struct CustomClientInfo
{
    eprosima::rpc::transport::dds::RTPSProxyTransport *transport_;
    rmw_fastrtps_cpp::RequestTypeSupport *request_type_support_;
    rmw_fastrtps_cpp::ResponseTypeSupport *response_type_support_;
    eprosima::rpc::transport::dds::RTPSProxyProcedureEndpoint *topic_endpoint_;
    ClientListener *listener_;
} CustomClientInfo;

class ClientListener
{
    public:

        ClientListener(CustomClientInfo *info) : info_(info),
        conditionMutex_(NULL), conditionVariable_(NULL) {}


        void onNewResponse(rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer *buffer)
        {
            assert(buffer);
            std::lock_guard<std::mutex> lock(internalMutex_);

            if(conditionMutex_ != NULL)
            {
                std::unique_lock<std::mutex> clock(*conditionMutex_);
                list.push_back(buffer);
                clock.unlock();
                conditionVariable_->notify_one();
            }
            else
                list.push_back(buffer);

        }

        rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer* getResponse()
        {
            std::lock_guard<std::mutex> lock(internalMutex_);
            rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer *buffer = nullptr;

            if(conditionMutex_ != NULL)
            {
                std::unique_lock<std::mutex> clock(*conditionMutex_);
                buffer = list.front();
                list.pop_front();
            }
            else
            {
                buffer = list.front();
                list.pop_front();
            }

            return buffer;
        }

        void attachCondition(std::mutex *conditionMutex, std::condition_variable *conditionVariable)
        {
            std::lock_guard<std::mutex> lock(internalMutex_);
            conditionMutex_ = conditionMutex;
            conditionVariable_ = conditionVariable;
        }

        void dettachCondition()
        {
            std::lock_guard<std::mutex> lock(internalMutex_);
            conditionMutex_ = NULL;
            conditionVariable_ = NULL;
        }

        bool hasData()
        {
            return !list.empty();
        }

    private:

        CustomClientInfo *info_;
        std::mutex internalMutex_;
        std::list<rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer*> list;
        std::mutex *conditionMutex_;
        std::condition_variable *conditionVariable_;
};


class AsyncTask : public eprosima::rpc::transport::dds::RTPSAsyncTask
{
    public:

        AsyncTask(CustomClientInfo *info) :
            info_(info),
            buffer_(static_cast<rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer*>(info->response_type_support_->createData()))
    {
    }

        virtual ~AsyncTask(){}

        virtual void execute()
        {
            info_->listener_->onNewResponse(buffer_);
        }

        virtual void on_exception(const eprosima::rpc::exception::SystemException &ex){}

        virtual void* getReplyInstance() { return buffer_; }

    private:

        CustomClientInfo *info_;

        rmw_fastrtps_cpp::ResponseTypeSupport::ResponseBuffer *buffer_;
};

extern "C"
{
    const char* const eprosima_fastrtps_identifier = "fastrtps";

    const char* rmw_get_implementation_identifier()
    {
        return eprosima_fastrtps_identifier;
    }

    rmw_ret_t rmw_init()
    {
        return RMW_RET_OK;
    }

    rmw_node_t* rmw_create_node(const char *name, size_t domain_id)
    {
        assert(name);

        ParticipantAttributes participantParam;
        participantParam.rtps.builtin.domainId = domain_id;
        participantParam.rtps.builtin.leaseDuration = c_TimeInfinite;
        participantParam.rtps.setName(name);

        Participant *participant = Domain::createParticipant(participantParam);

        if(!participant)
{'fraction_non_alphanumeric': 0.10296614401278338, 'fraction_numerical': 9.987016878058524e-05, 'mean_word_length': 3.0411622276029058, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4104147', 'n_tokens_mistral': 3036, 'n_tokens_neox': 2893, 'n_words': 418}