Transport module

The Transport module is responsible for interfacing with the underlying transport system. This module provides the following interfaces:

• transport::ITransport

• transport::IGUIDLocatorDatabase

The transport module is related to the following layers:

• Discovery module by the discovery::ILocatorObserver interface.

• Protocol module by the protocol::IMessageSink interface.

• Execution module by the ISpinnable interface.

Default transport

If Safe DDS has been built with any built-in transport support using the SAFEDDS_TRANSPORT (see CMake options), the default transport can be created using the system-wide create_transport method. This API will be used internally by the DomainParticipantFactory to instantiate DomainParticipant entities.

In the default POSIX build configuration, where SAFEDDS_TRANSPORT=posix/udpv4_sharedmemory, create_transport returns a transport::posix::UDPv4_SharedMemory transport that combines transport::posix::UDPv4 and transport::posix::SharedMemory in a single transport::ITransport implementation.

constexpr uint32_t DEFAULT_RECEPTION_BUFFER_SIZE = 10000;
constexpr uint32_t DEFAULT_SENDING_MAX_SIZE = 10000;

// Create a default transport (depends on the transport implementation built).
transport::ITransport& transport = create_transport(
    get_default_allocator(), // Use default platform's allocator
    transport::TransportAllocConfig {
        memory::container::DEFAULT_MEMORY_CONFIG,
        memory::container::DEFAULT_MEMORY_CONFIG,
        memory::container::DEFAULT_MEMORY_CONFIG
    },
    DEFAULT_RECEPTION_BUFFER_SIZE,
    DEFAULT_SENDING_MAX_SIZE);

Note

Provided transport implementations use the transport::PreallocGUIDLocatorDatabase class to store and handle discovered locators.

ITransport interface

API Reference

For more information about transport::ITransport interface, check API Reference:

• transport::ITransport

In general, transport::ITransport allows for sending messages to a certain destination and for listening to incoming messages:

1. In order to send a message the commit_message and commit_message_to_group methods will be used.

2. In order to listen to messages, the Safe DDS transport interface provides two different approaches:

   • listen_message method allows the transport to listen to incoming messages for a given period of time.

   • The transport::ITransport implements ISpinnable interface, allowing the incoming messages reception to be performed in a non-blocking way using the Safe DDS Execution module.

3. Additionally, transport::ITransport provides methods for:

   • Notifying newly discovered locators in order to store the relationship between entity identifiers and transport locators via on_locator_discovered and on_group_discovered methods.

   • Notifying when a locator or an endpoint group is not required anymore via on_locators_unregistered and on_group_unregistered methods respectively.

   • Retrieving the maximum size of a message that can be sent by the transport via the get_message_max_size method.

   • Listening on a certain locator via the listen_on_locator method.

   • Given a hint, listen on the first available locator found via the listen_on_first_available_locator method.

   • Waiting for a message to be available for processing via the wait_until_message_available method.

   • Setting message observers for incoming messages to be called when a new message is received (and the transport implements ISpinnable interface) via the set_message_observer method.

   • Setting a custom configuration for a given origin GUID via the set_origin_configuration method.

   • Retrieving a default multicast locator, provided a logical port number, via the get_default_multicast_locator method.

get_message_max_size

This method is invoked by Safe DDS layers to retrieve the maximum size of a message that the transport can send. It has no parameters and returns the maximum size of a message as uint32_t.

API Reference

For more information about transport::ITransport::get_message_max_size method, check API Reference:

• protocol::IMessageSink::get_message_max_size

commit_message

This method is invoked when an entity has a new message ready to be sent. The transport implementation must send the message (provided as memory::IList<MessageElement>) to the destination protocol::GUID.

This callback receives the following parameters:

• Message buffer as memory::IList<MessageElement>

  This object contains one or more memory segments that should be sent to the destination as a single message.

• Origin as protocol::GUID

  GUID of the entity that is sending the message.

• Destination as protocol::GUID

  GUID of the entity that shall receive the message.

• Timepoint as execution::Timepoint

  Timeout for the message to be sent.

API Reference

For more information about transport::ITransport::commit_message method, check API Reference:

• protocol::IMessageSink::commit_message

commit_message_to_group

This method is invoked when an entity has a new message ready to be sent to a group of wire protocol message sinks. The transport implementation must send the message (provided as memory::IList<MessageElement>) to the destination protocol::EndpointGroup.

This callback receives the following parameters:

• Message buffer as memory::IList<MessageElement>

  This object contains one or more memory segments that should be sent to the destination as a single message.

• Origin as protocol::GUID

  GUID of the entity that is sending the message.

• Destination as protocol::EndpointGroup

  Group of entities locators that shall receive the message.

API Reference

For more information about transport::ITransport::commit_message_to_group method, check API Reference:

• protocol::IMessageSink::commit_message_to_group

set_message_observer

This method allows the transport to receive a transport::IMessageObserver that will be notified when a new message is received in the spin operation of the transport. These message observers are intended to be notified about the reception of a message using the ISpinnable interface.

API Reference

For more information about transport::ITransport::set_message_observer method, check API Reference:

• transport::ITransport::set_message_observer

listen_on_locator

This method is invoked by Safe DDS layers for the transport to start listening on a specific locator. This callback returns a safedds::ReturnCode and receives the following parameters:

• Locator as transport::Locator

  Locator to listen on.

API Reference

For more information about transport::ITransport::listen_on_locator method, check API Reference:

• transport::IMessageListener::listen_on_locator

listen_on_first_available_locator

This method is invoked by Safe DDS layers to get a listen locator when the DomainParticipantWireProtocolQosPolicy::announced_locator is not configured. This method shall return a valid transport::Locator which is already configured to listen to messages.

API Reference

For more information about transport::ITransport::listen_on_first_available_locator method, check API Reference:

• transport::IMessageListener::listen_on_first_available_locator

listen_message

This method enables the transport to listen for incoming messages in any of the locators that have been previously registered with the listen_on_locator method during a give period of time. This callback returns a transport::Locator and receives the following parameters:

• Incoming message buffer as memory::IMutableByteArrayView

  This object contains one memory segment that shall be filled with the received message.

• Reception locator as transport::Locator

  Output parameter with the Locator where the message has been received.

• Timepoint as execution::Timepoint

  Timeout for the message to be received.

API Reference

For more information about transport::ITransport::listen_on_locator method, check API Reference:

• transport::IMessageListener::listen_message

wait_until_message_available

This method waits up to a given timeout until the transport has a message available for processing. This callback returns a boolean indicating whether a message is available and receives the following parameters:

• Timepoint as execution::Timepoint

  Timeout for the message to be received.

API Reference

For more information about transport::ITransport::wait_until_message_available method, check API Reference:

• transport::IMessageListener::wait_until_message_available

on_locator_discovered

This method is invoked when a new locator is identified, possibly by the Discovery module. The transport implementation may require to store the provided protocol::GUID and transport::Locator for future use.

The stored information can be useful when implementing the commit_message method, as this method will provide origin and destination locators of a certain message.

This callback receives the following parameters:

• Discovered entity GUID as protocol::GUID

  GUID of the entity that has been discovered.

• Metatraffic flag as bool

  Indicates whether the locator belongs to metatraffic.

• Discovered entity locator as transport::Locator

  Locator of the entity that has been discovered.

• Discoverer entity GUIDPrefix as protocol::GUIDPrefix

  GUIDPrefix of the entity that discovered the locator.

API Reference

For more information about transport::ITransport::on_locator_discovered method, check API Reference:

• discovery::ILocatorObserver::on_locator_discovered

on_group_discovered

This method is invoked when a new endpoint group is identified, possibly by the Discovery module. The transport implementation may require to store the provided protocol::EndpointGroup and protocol::EndpointGroup for future use.

The stored information can be useful when implementing the commit_message_to_group method, as this method will provide origin and destination locators of a certain message.

This callback receives the following parameters:

• Discovered endpoint group as protocol::EndpointGroup

  Group that has been discovered.

• Metatraffic flag as bool

  Indicates whether the locator belongs to metatraffic.

• Discovered entity locator as transport::Locator

  Locator of the entity that has been discovered.

• Discoverer entity GUIDPrefix as protocol::GUIDPrefix

  GUIDPrefix of the entity that discovered the locator.

API Reference

For more information about transport::ITransport::on_group_discovered method, check API Reference:

• discovery::ILocatorObserver::on_group_discovered

on_locators_unregistered

This method is invoked when a previously discovered locator is removed, possibly by the Discovery module. The transport implementation may require to remove the stored protocol::GUID. This callback receives the following parameters:

• Unregistered entity GUID as protocol::GUID

  GUID of the entity that has been unregistered.

• Discoverer entity GUIDPrefix as protocol::GUIDPrefix

  GUIDPrefix of the entity that received the unregister event.

API Reference

For more information about transport::ITransport::on_locators_unregistered method, check API Reference:

• discovery::ILocatorObserver::on_locators_unregistered

on_group_unregistered

This method is invoked when a previously discovered endpoint group is removed, possibly by the Discovery module. The transport implementation may require to remove the stored protocol::EndpointGroup. This callback receives the following parameters:

• Unregistered endpoint group as protocol::EndpointGroup

  Group that has been unregistered.

• Discoverer entity GUIDPrefix as protocol::GUIDPrefix

  GUIDPrefix of the entity that received the unregister event.

API Reference

For more information about transport::ITransport::on_group_unregistered method, check API Reference:

• discovery::ILocatorObserver::on_group_unregistered

set_origin_configuration

This method allows the transport to specify a custom configuration for a given origin GUID. This callback returns a safedds::ReturnCode and receives the following parameters:

• DataWriter’s GUID as protocol::GUID

  GUID of the entity for which a custom sending configuration is required.

• DataWriter’s locator as transport::Locator

  Locator specifying the source address and port for datagrams sent by the specified GUID. An invalid locator means that no specific binding is required.

• DataWriter’s priority as transport::Priority

  Transport priority to use for datagrams sent by the specified GUID.

API Reference

For more information about transport::ITransport::set_origin_configuration method, check API Reference:

• transport::ITransport::set_origin_configuration

get_default_multicast_locator

This method allows the transport to provide a default multicast locator, given a logical port number. This callback returns a transport::Locator and receives the following parameters:

• Logical port as uint32_t

  Logical port number to use to create the default multicast locator.

API Reference

For more information about transport::ITransport::get_default_multicast_locator method, check API Reference:

• transport::ITransport::get_default_multicast_locator

Example implementation

Note

For complete transport implementation, please refer to the POSIX UDPv4 transport implementation in Safe DDS code base.

A naive Safe DDS transport implementation that sends messages to a remote host using an abstract transport is provided in the following example:

class CustomTransport :
    public transport::ITransport
{
    void commit_message(
            memory::IList<protocol::MessageElement>& view_list,
            const protocol::GUID& origin_guid,
            const protocol::GUID& receiver_guid,
            const execution::TimePoint& /* timepoint */) noexcept override
    {
        // Retrieve the locator from the remote entities map.
        transport::Locator locator = remote_entities_[receiver_guid];

        // Send the message to the locator.
        send_message(view_list, origin_guid, locator);
    }

    void commit_message_to_group(
            memory::IList<protocol::MessageElement>& view_list,
            const protocol::GUID& origin_guid,
            const protocol::EndpointGroup& group,
            const execution::TimePoint& /* timepoint */) noexcept override
    {
        // Retrieve the locator from the group entities map.
        transport::Locator locator = group_entities_[group];

        // Send the message to the locator.
        send_message(view_list, origin_guid, locator);
    }

    void send_message(
            memory::IList<protocol::MessageElement>& view_list,
            const protocol::GUID& origin_guid,
            const transport::Locator& locator) noexcept
    {
        // Check if the origin GUID has a specific priority configuration
        int32_t priority = DEFAULT_TRANSPORT_PRIORITY;
        auto priority_ptr = origin_configurations_.find(origin_guid);

        if (nullptr != priority_ptr)
        {
            priority = *priority_ptr;
        }

        // Store the message in a continuous buffer.
        uint32_t buffer_size = 0;

        for (uint32_t i = 0; i < view_list.size(); i++)
        {
            buffer_size += view_list.at(i)->const_size();
        }

        uint8_t* buffer = new uint8_t[buffer_size];
        uint32_t buffer_offset = 0;

        for (uint32_t i = 0; i < view_list.size(); i++)
        {
            memory::IConstByteArrayView* view = view_list.at(i);
            memcpy(&buffer[buffer_offset], view->const_data(), view->const_size());
            buffer_offset += view->const_size();
        }

        memory::byte_array::ByteArrayView view = {buffer, buffer_offset};

        // Transport specific function to send the message to the locator [NOT PROVIDED]
        custom_send_message(view, locator, priority);

        delete [] buffer;
    }

    ReturnCode set_origin_configuration(
            const protocol::GUID& guid,
            const transport::Locator& /* locator */,
            int32_t priority) noexcept override
    {
        // Store the origin configuration in a map.
        origin_configurations_.add(guid, priority);

        return ReturnCode::OK;
    }

    transport::Locator get_default_multicast_locator(
            uint32_t logical_port) noexcept override
    {
        constexpr transport::Locator::IPv4 MULTICAST_IPV4 = {239, 255, 0, 1};

        return transport::Locator::from_ipv4(MULTICAST_IPV4, logical_port);
    }

    uint32_t get_message_max_size() const noexcept override
    {
        // This transport allows sending messages of maximum 5000 bytes.
        return 5000;
    }

    void set_message_observer(
            transport::IMessageObserver& observer) noexcept override
    {
        // Override the local message observer.
        observer_ = &observer;
    }

    void on_locator_discovered(
            const protocol::GUID& guid,
            bool /* is_metatraffic */,
            const transport::Locator& locator,
            const protocol::GUIDPrefix& /* discoverer */) noexcept override
    {
        // Store the discovered locator.
        remote_entities_[guid] = locator;
    }

    void on_locators_unregistered(
            const protocol::GUID& guid,
            const protocol::GUIDPrefix& /* discoverer */) noexcept override
    {
        // Remove the locator.
        remote_entities_.erase(guid);
    }

    void on_group_discovered(
            const protocol::EndpointGroup& group,
            bool /* is_metatraffic */,
            const transport::Locator& locator,
            const protocol::GUIDPrefix& /* discoverer */) noexcept override
    {
        // Store the discovered group.
        group_entities_[group] = locator;
    }

    void on_group_unregistered(
            const protocol::EndpointGroup& group,
            const protocol::GUIDPrefix& /* discoverer */) noexcept override
    {
        group_entities_.erase(group);
    }

    ReturnCode listen_on_locator(
            const transport::Locator& reception_locator) noexcept override
    {
        // Transport specific function to listen on a given locator [NOT PROVIDED]
        custom_start_listening(reception_locator);

        return ReturnCode::OK;
    }

    transport::Locator listen_on_first_available_locator(
            const transport::Locator& /* hint */) noexcept override
    {
        // Return hardcoded default locator
        transport::Locator default_listen_locator = transport::Locator::from_ipv4({127, 0, 0, 1}, 8000);
        listen_on_locator(default_listen_locator);

        return default_listen_locator;
    }

    void spin(
            const execution::TimePoint& tm) noexcept override
    {
        // Listen message and notify the observer if exists.
        if (observer_ != nullptr)
        {
            transport::Locator locator = {};
            uint8_t* buffer = new uint8_t[10000];
            memory::byte_array::ByteArrayView view = {buffer, 10000};

            ReturnCode code = listen_message(view, locator, tm);

            if (code == ReturnCode::OK)
            {
                // Received message contained in view from locator.
                // Notify the observer.
                observer_->on_message_received(view, locator);
            }

            delete [] buffer;
        }
    }

    bool has_pending_work() const noexcept override
    {
        // Transport specific function to check whether there is data in
        // the transport [NOT PROVIDED]
        return custom_transport_has_data();
    }

    bool wait_until_message_available(
            const execution::TimePoint& timepoint) const noexcept override
    {
        // Transport specific function to wait until there is data [NOT PROVIDED]
        return custom_transport_wait_for_data(timepoint);
    }

    protocol::EndpointGroup get_group(
            const transport::Locator& locator) const noexcept override
    {
        protocol::EndpointGroup group = {};
        portable::safe_memcpy(group.data.data(), &locator, portable::safe_sizeof<transport::Locator>());

        return group;
    }

    execution::TimePoint get_next_work_timepoint() const noexcept override
    {
        return has_pending_work() ? execution::TIME_ZERO: execution::TIME_INFINITE;
    }

    ReturnCode listen_message(
            memory::IMutableByteArrayView& msg,
            transport::Locator& reception_locator,
            const execution::TimePoint& timepoint) noexcept override
    {
        execution::TimePeriod wait_time =
                timepoint - get_platform().get_current_timepoint();
        size_t milliseconds_to_wait =
                wait_time.seconds * 1000 + wait_time.nanoseconds / 1000000;

        // Transport specific function to listen for a message for a given
        // period of time [NOT PROVIDED]
        reception_locator = custom_listen_message(msg, milliseconds_to_wait);

        return msg.size() > 0 ? ReturnCode::OK : ReturnCode::TRANSPORT_TIMEOUT;
    }

private:

    // Local message observer
    transport::IMessageObserver* observer_{nullptr};

    // GUID comparator
    struct GUIDComparator
    {
        bool operator ()(
                const protocol::GUID& a,
                const protocol::GUID& b) const
        {
            return memcmp(&a, &b, sizeof(protocol::GUID)) < 0;
        }

    };

    // Group comparator
    struct GroupComparator
    {
        bool operator ()(
                const protocol::EndpointGroup& a,
                const protocol::EndpointGroup& b) const
        {
            return memcmp(&a, &b, sizeof(protocol::EndpointGroup)) < 0;
        }

    };

    // Locator storage
    std::map<protocol::GUID, transport::Locator, GUIDComparator> remote_entities_;
    std::map<protocol::EndpointGroup, transport::Locator, GroupComparator> group_entities_;

    // Origin configurations priority
    memory::container::StaticMap<protocol::GUID, int, 100> origin_configurations_;
};