Cross-host communication¶

Learning Objectives

This tutorial extends an iceoryx2 system across host boundaries using the gateway:

Enable cross-host communication with the iox2 gateway CLI, without touching application code
Embed the gateway in your own application for full control over its execution
Drive the gateway in polling and reactive modes
Switch the communication mechanism by instantiating a different backend

So far Larry has been a self-contained system: his sensors, his control logic, and his actuators all reside on one host and talk over iceoryx2’s shared memory. But shared memory does not extend beyond the boundary of the host.

Imagine now that the user wants to remotely track where Larry is and how much battery he has left. Some kind of dashboard hosted on another device on the same network.

The required data is already flowing through iceoryx2’s shared memory. A gateway can hook into that flow and propagate it to other hosts. To get the data to the dashboard, we will use a gateway that forwards the raw shared memory payloads over the network.

Tip

A gateway can be implemented for any communication mechanism, not just network communication. All that is required is an implementation of the Backend traits available in iceoryx2-gateway-backend.

For example, the gateway has been used to extend communication between co-processors on a single board that have their own memory using cross-chip communication APIs.

Application setup¶

As in previous articles in this series, the data flowing through Larry are defined as ZeroCopySend structs:

RUST

use iceoryx2::prelude::*;

#[derive(Debug, Clone, Copy, ZeroCopySend)]
// shared type name; must match across languages
#[type_name("BatteryState")]
#[repr(C)]
pub struct BatteryState {
    pub charge_percent: f32,
}

#[derive(Debug, Clone, Copy, ZeroCopySend)]
// shared type name; must match across languages
#[type_name("Position")]
#[repr(C)]
pub struct Position {
    pub x: f32,
    pub y: f32,
}

PYTHON

import ctypes

class BatteryState(ctypes.Structure):
    """Battery telemetry payload."""

    _fields_ = [("charge_percent", ctypes.c_float)]

    # shared type name; must match across languages
    @staticmethod
    def type_name() -> str:
        return "BatteryState"


class Position(ctypes.Structure):
    """Position telemetry payload."""

    _fields_ = [("x", ctypes.c_float), ("y", ctypes.c_float)]

    # shared type name; must match across languages
    @staticmethod
    def type_name() -> str:
        return "Position"

C++

struct BatteryState {
    // shared type name; must match across languages
    static constexpr const char* IOX2_TYPE_NAME = "BatteryState";
    float charge_percent;
};

struct Position {
    // shared type name; must match across languages
    static constexpr const char* IOX2_TYPE_NAME = "Position";
    float x;
    float y;
};

// shared type name; must match across languages
#define BATTERY_STATE_TYPE_NAME "BatteryState"
struct BatteryState {
    float charge_percent;
};

// shared type name; must match across languages
#define POSITION_TYPE_NAME "Position"
struct Position {
    float x;
    float y;
};

For demonstration purposes, we use the event-driven communication execution pattern, which is the same approach used by many popular network communication libraries. However, with iceoryx2, you are free to choose the best approach that suits your application. The gateway can be configured to work with any approach. See execution control patterns for an overview.

Note

The gateway currently only supports the publish-subscribe and event messaging patterns. Support for request-response and blackboard is planned.

We set Larry up to send a notification on a corresponding event service along with every sample that is published:

RUST

use core::time::Duration;
use cross_host_communication::telemetry_data::{BatteryState, Position};
use iceoryx2::prelude::*;

const CYCLE_TIME: Duration = Duration::from_secs(1);

let node = NodeBuilder::new()
    .name(&"Larry".try_into()?)
    .create::<ipc::Service>()?;

let battery_publisher = node
    .service_builder(&"larry/battery".try_into()?)
    .publish_subscribe::<BatteryState>()
    .open_or_create()?
    .publisher_builder()
    .create()?;
let battery_notifier = node
    .service_builder(&"larry/battery".try_into()?)
    .event()
    .open_or_create()?
    .notifier_builder()
    .create()?;

let position_publisher = node
    .service_builder(&"larry/position".try_into()?)
    .publish_subscribe::<Position>()
    .open_or_create()?
    .publisher_builder()
    .create()?;
let position_notifier = node
    .service_builder(&"larry/position".try_into()?)
    .event()
    .open_or_create()?
    .notifier_builder()
    .create()?;

while node.wait(CYCLE_TIME).is_ok() {
    let battery = battery_publisher.loan_uninit()?;
    let battery = battery.write_payload(BatteryState {
        charge_percent: 87.0,
    });
    battery.send()?;
    battery_notifier.notify()?;

    let position = position_publisher.loan_uninit()?;
    let position = position.write_payload(Position { x: 1.0, y: 2.0 });
    position.send()?;
    position_notifier.notify()?;
}

PYTHON

import iceoryx2 as iox2
from telemetry_data import BatteryState, Position

cycle_time = iox2.Duration.from_secs(1)

iox2.set_log_level_from_env_or(iox2.LogLevel.Info)
node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)

battery_publisher = (
    node.service_builder(iox2.ServiceName.new("larry/battery"))
    .publish_subscribe(BatteryState)
    .open_or_create()
    .publisher_builder()
    .create()
)
battery_notifier = (
    node.service_builder(iox2.ServiceName.new("larry/battery"))
    .event()
    .open_or_create()
    .notifier_builder()
    .create()
)

position_publisher = (
    node.service_builder(iox2.ServiceName.new("larry/position"))
    .publish_subscribe(Position)
    .open_or_create()
    .publisher_builder()
    .create()
)
position_notifier = (
    node.service_builder(iox2.ServiceName.new("larry/position"))
    .event()
    .open_or_create()
    .notifier_builder()
    .create()
)

try:
    while True:
        node.wait(cycle_time)

        battery = battery_publisher.loan_uninit()
        battery = battery.write_payload(BatteryState(charge_percent=87.0))
        battery.send()
        battery_notifier.notify()

        position = position_publisher.loan_uninit()
        position = position.write_payload(Position(x=1.0, y=2.0))
        position.send()
        position_notifier.notify()

except iox2.NodeWaitFailure:
    print("exit")

C++

constexpr iox2::bb::Duration CYCLE_TIME = iox2::bb::Duration::from_secs(1);

auto main() -> int {
    using namespace iox2;
    set_log_level_from_env_or(LogLevel::Info);

    auto node = NodeBuilder().name(NodeName::create("Larry").value()).create<ServiceType::Ipc>().value();

    auto battery_publisher = node.service_builder(ServiceName::create("larry/battery").value())
                                 .publish_subscribe<BatteryState>()
                                 .open_or_create()
                                 .value()
                                 .publisher_builder()
                                 .create()
                                 .value();
    auto battery_notifier = node.service_builder(ServiceName::create("larry/battery").value())
                                .event()
                                .open_or_create()
                                .value()
                                .notifier_builder()
                                .create()
                                .value();

    auto position_publisher = node.service_builder(ServiceName::create("larry/position").value())
                                  .publish_subscribe<Position>()
                                  .open_or_create()
                                  .value()
                                  .publisher_builder()
                                  .create()
                                  .value();
    auto position_notifier = node.service_builder(ServiceName::create("larry/position").value())
                                 .event()
                                 .open_or_create()
                                 .value()
                                 .notifier_builder()
                                 .create()
                                 .value();

    while (node.wait(CYCLE_TIME).has_value()) {
        auto battery = battery_publisher.loan_uninit().value();
        auto initialized_battery = battery.write_payload(BatteryState { 87.0F }); // NOLINT
        send(std::move(initialized_battery)).has_value();
        battery_notifier.notify().has_value();

        auto position = position_publisher.loan_uninit().value();
        auto initialized_position = position.write_payload(Position { 1.0F, 2.0F }); // NOLINT
        send(std::move(initialized_position)).has_value();
        position_notifier.notify().has_value();
    }

    return 0;
}

#define CYCLE_TIME_SECS 1
#define CYCLE_TIME_NANOS 0

int main(void) {
    iox2_set_log_level_from_env_or(iox2_log_level_e_INFO);
    int ret_val = 0;

    // create node
    iox2_node_builder_h node_builder = iox2_node_builder_new(NULL);
    iox2_node_h node = NULL;
    ret_val = iox2_node_builder_create(node_builder, NULL, iox2_service_type_e_IPC, &node);
    if (ret_val != IOX2_OK) {
        printf("Could not create node! Error: %d\n", ret_val);
        goto end;
    }

    // create battery publish-subscribe service
    iox2_service_name_h battery_name = NULL;
    const char* battery_name_value = "larry/battery";
    iox2_service_name_new(NULL, battery_name_value, strlen(battery_name_value), &battery_name);
    iox2_service_builder_pub_sub_h battery_builder =
        iox2_service_builder_pub_sub(iox2_node_service_builder(&node, NULL, iox2_cast_service_name_ptr(battery_name)));
    iox2_service_builder_pub_sub_set_payload_type_details(&battery_builder,
                                                          iox2_type_variant_e_FIXED_SIZE,
                                                          BATTERY_STATE_TYPE_NAME,
                                                          strlen(BATTERY_STATE_TYPE_NAME),
                                                          sizeof(struct BatteryState),
                                                          alignof(struct BatteryState));
    iox2_port_factory_pub_sub_h battery_service = NULL;
    ret_val = iox2_service_builder_pub_sub_open_or_create(battery_builder, NULL, &battery_service);
    if (ret_val != IOX2_OK) {
        printf("Could not create battery service! Error: %d\n", ret_val);
        goto drop_node;
    }

    // create position publish-subscribe service
    iox2_service_name_h position_name = NULL;
    const char* position_name_value = "larry/position";
    iox2_service_name_new(NULL, position_name_value, strlen(position_name_value), &position_name);
    iox2_service_builder_pub_sub_h position_builder =
        iox2_service_builder_pub_sub(iox2_node_service_builder(&node, NULL, iox2_cast_service_name_ptr(position_name)));
    iox2_service_builder_pub_sub_set_payload_type_details(&position_builder,
                                                          iox2_type_variant_e_FIXED_SIZE,
                                                          POSITION_TYPE_NAME,
                                                          strlen(POSITION_TYPE_NAME),
                                                          sizeof(struct Position),
                                                          alignof(struct Position));
    iox2_port_factory_pub_sub_h position_service = NULL;
    ret_val = iox2_service_builder_pub_sub_open_or_create(position_builder, NULL, &position_service);
    if (ret_val != IOX2_OK) {
        printf("Could not create position service! Error: %d\n", ret_val);
        goto drop_battery_service;
    }

    // create battery event service
    iox2_service_builder_event_h battery_event_builder =
        iox2_service_builder_event(iox2_node_service_builder(&node, NULL, iox2_cast_service_name_ptr(battery_name)));
    iox2_port_factory_event_h battery_event_service = NULL;
    ret_val = iox2_service_builder_event_open_or_create(battery_event_builder, NULL, &battery_event_service);
    if (ret_val != IOX2_OK) {
        printf("Could not create battery event service! Error: %d\n", ret_val);
        goto drop_position_service;
    }

    // create position event service
    iox2_service_builder_event_h position_event_builder =
        iox2_service_builder_event(iox2_node_service_builder(&node, NULL, iox2_cast_service_name_ptr(position_name)));
    iox2_port_factory_event_h position_event_service = NULL;
    ret_val = iox2_service_builder_event_open_or_create(position_event_builder, NULL, &position_event_service);
    if (ret_val != IOX2_OK) {
        printf("Could not create position event service! Error: %d\n", ret_val);
        goto drop_battery_event_service;
    }

    // create publishers
    iox2_publisher_h battery_publisher = NULL;
    iox2_port_factory_publisher_builder_create(
        iox2_port_factory_pub_sub_publisher_builder(&battery_service, NULL), NULL, &battery_publisher);
    iox2_publisher_h position_publisher = NULL;
    iox2_port_factory_publisher_builder_create(
        iox2_port_factory_pub_sub_publisher_builder(&position_service, NULL), NULL, &position_publisher);

    // create notifiers
    iox2_notifier_h battery_notifier = NULL;
    iox2_port_factory_notifier_builder_create(
        iox2_port_factory_event_notifier_builder(&battery_event_service, NULL), NULL, &battery_notifier);
    iox2_notifier_h position_notifier = NULL;
    iox2_port_factory_notifier_builder_create(
        iox2_port_factory_event_notifier_builder(&position_event_service, NULL), NULL, &position_notifier);

    while (iox2_node_wait(&node, CYCLE_TIME_SECS, CYCLE_TIME_NANOS) == IOX2_OK) {
        // publish battery state and notify
        iox2_sample_mut_h battery_sample = NULL;
        if (iox2_publisher_loan_slice_uninit(&battery_publisher, NULL, &battery_sample, 1) == IOX2_OK) {
            struct BatteryState* battery_payload = NULL;
            iox2_sample_mut_payload_mut(&battery_sample, (void**) &battery_payload, NULL);
            battery_payload->charge_percent = 87.0F; // NOLINT
            iox2_sample_mut_send(battery_sample, NULL);
            iox2_notifier_notify(&battery_notifier, NULL);
        }

        // publish position and notify
        iox2_sample_mut_h position_sample = NULL;
        if (iox2_publisher_loan_slice_uninit(&position_publisher, NULL, &position_sample, 1) == IOX2_OK) {
            struct Position* position_payload = NULL;
            iox2_sample_mut_payload_mut(&position_sample, (void**) &position_payload, NULL);
            position_payload->x = 1.0F; // NOLINT
            position_payload->y = 2.0F; // NOLINT
            iox2_sample_mut_send(position_sample, NULL);
            iox2_notifier_notify(&position_notifier, NULL);
        }
    }

    iox2_notifier_drop(position_notifier);
    iox2_notifier_drop(battery_notifier);
    iox2_publisher_drop(position_publisher);
    iox2_publisher_drop(battery_publisher);
    iox2_port_factory_event_drop(position_event_service);
drop_battery_event_service:
    iox2_port_factory_event_drop(battery_event_service);
drop_position_service:
    iox2_port_factory_pub_sub_drop(position_service);
drop_battery_service:
    iox2_port_factory_pub_sub_drop(battery_service);
drop_node:
    iox2_node_drop(node);
end:
    return ret_val;
}

In the dashboard application running on another host, we set up listeners to wake the thread when new data arrives, and subscribers to receive the data. A WaitSet is used to react to events from both listeners in a single thread.

RUST

use cross_host_communication::telemetry_data::{BatteryState, Position};
use iceoryx2::prelude::*;

let node = NodeBuilder::new()
    .name(&"Dashboard".try_into()?)
    .create::<ipc::Service>()?;

let battery_subscriber = node
    .service_builder(&"larry/battery".try_into()?)
    .publish_subscribe::<BatteryState>()
    .open_or_create()?
    .subscriber_builder()
    .create()?;
let battery_listener = node
    .service_builder(&"larry/battery".try_into()?)
    .event()
    .open_or_create()?
    .listener_builder()
    .create()?;

let position_subscriber = node
    .service_builder(&"larry/position".try_into()?)
    .publish_subscribe::<Position>()
    .open_or_create()?
    .subscriber_builder()
    .create()?;
let position_listener = node
    .service_builder(&"larry/position".try_into()?)
    .event()
    .open_or_create()?
    .listener_builder()
    .create()?;

let waitset = WaitSetBuilder::new().create::<ipc::Service>()?;
let battery_guard = waitset.attach_notification(&battery_listener)?;
let position_guard = waitset.attach_notification(&position_listener)?;

let on_event = |attachment_id: WaitSetAttachmentId<ipc::Service>| {
    if attachment_id.has_event_from(&battery_guard) {
        // drain every pending notification, otherwise the WaitSet wakes
        // again immediately and spins
        battery_listener.try_wait(|_| {}).unwrap();
        while let Ok(Some(sample)) = battery_subscriber.receive() {
            println!("battery: {}%", sample.charge_percent);
        }
    } else if attachment_id.has_event_from(&position_guard) {
        // drain every pending notification, otherwise the WaitSet wakes
        // again immediately and spins
        position_listener.try_wait(|_| {}).unwrap();
        while let Ok(Some(sample)) = position_subscriber.receive() {
            println!("position: ({}, {})", sample.x, sample.y);
        }
    }

    CallbackProgression::Continue
};

waitset.wait_and_process(on_event)?;

PYTHON

import iceoryx2 as iox2
from telemetry_data import BatteryState, Position

iox2.set_log_level_from_env_or(iox2.LogLevel.Info)
node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)

battery_subscriber = (
    node.service_builder(iox2.ServiceName.new("larry/battery"))
    .publish_subscribe(BatteryState)
    .open_or_create()
    .subscriber_builder()
    .create()
)
battery_listener = (
    node.service_builder(iox2.ServiceName.new("larry/battery"))
    .event()
    .open_or_create()
    .listener_builder()
    .create()
)

position_subscriber = (
    node.service_builder(iox2.ServiceName.new("larry/position"))
    .publish_subscribe(Position)
    .open_or_create()
    .subscriber_builder()
    .create()
)
position_listener = (
    node.service_builder(iox2.ServiceName.new("larry/position"))
    .event()
    .open_or_create()
    .listener_builder()
    .create()
)

waitset = iox2.WaitSetBuilder.new().create(iox2.ServiceType.Ipc)
battery_guard = waitset.attach_notification(battery_listener)
position_guard = waitset.attach_notification(position_listener)

try:
    while True:
        attachments, result = waitset.wait_and_process()
        if result in (
            iox2.WaitSetRunResult.TerminationRequest,
            iox2.WaitSetRunResult.Interrupt,
        ):
            break

        for attachment_id in attachments:
            if attachment_id.has_event_from(battery_guard):
                # drain every pending notification, otherwise the WaitSet wakes
                # again immediately and spins
                battery_listener.try_wait()
                sample = battery_subscriber.receive()
                while sample is not None:
                    print("battery:", sample.payload().contents.charge_percent, "%")
                    sample = battery_subscriber.receive()
            elif attachment_id.has_event_from(position_guard):
                # drain every pending notification, otherwise the WaitSet wakes
                # again immediately and spins
                position_listener.try_wait()
                sample = position_subscriber.receive()
                while sample is not None:
                    position = sample.payload().contents
                    print("position:", (position.x, position.y))
                    sample = position_subscriber.receive()

except iox2.WaitSetRunError:
    print("exit")

C++

auto main() -> int {
    using namespace iox2;
    set_log_level_from_env_or(LogLevel::Info);

    auto node = NodeBuilder().name(NodeName::create("Dashboard").value()).create<ServiceType::Ipc>().value();

    auto battery_subscriber = node.service_builder(ServiceName::create("larry/battery").value())
                                  .publish_subscribe<BatteryState>()
                                  .open_or_create()
                                  .value()
                                  .subscriber_builder()
                                  .create()
                                  .value();
    auto battery_listener = node.service_builder(ServiceName::create("larry/battery").value())
                                .event()
                                .open_or_create()
                                .value()
                                .listener_builder()
                                .create()
                                .value();

    auto position_subscriber = node.service_builder(ServiceName::create("larry/position").value())
                                   .publish_subscribe<Position>()
                                   .open_or_create()
                                   .value()
                                   .subscriber_builder()
                                   .create()
                                   .value();
    auto position_listener = node.service_builder(ServiceName::create("larry/position").value())
                                 .event()
                                 .open_or_create()
                                 .value()
                                 .listener_builder()
                                 .create()
                                 .value();

    auto waitset = WaitSetBuilder().create<ServiceType::Ipc>().value();
    auto battery_guard = waitset.attach_notification(battery_listener).value();
    auto position_guard = waitset.attach_notification(position_listener).value();

    auto on_event = [&](WaitSetAttachmentId<ServiceType::Ipc> attachment_id) -> CallbackProgression {
        if (attachment_id.has_event_from(battery_guard)) {
            // drain every pending notification, otherwise the WaitSet wakes
            // again immediately and spins
            battery_listener.try_wait([](auto) {}).value();
            auto battery = battery_subscriber.receive().value();
            while (battery.has_value()) {
                std::cout << "battery: " << battery->payload().charge_percent << "%" << std::endl;
                battery = battery_subscriber.receive().value();
            }
        } else if (attachment_id.has_event_from(position_guard)) {
            // drain every pending notification, otherwise the WaitSet wakes
            // again immediately and spins
            position_listener.try_wait([](auto) {}).value();
            auto position = position_subscriber.receive().value();
            while (position.has_value()) {
                std::cout << "position: (" << position->payload().x << ", " << position->payload().y << ")"
                          << std::endl;
                position = position_subscriber.receive().value();
            }
        }

        return CallbackProgression::Continue;
    };

    waitset.wait_and_process(on_event).value();

    return 0;
}

struct CallbackContext {
    iox2_waitset_guard_h_ref battery_guard;
    iox2_waitset_guard_h_ref position_guard;
    iox2_listener_h_ref battery_listener;
    iox2_listener_h_ref position_listener;
    iox2_subscriber_h_ref battery_subscriber;
    iox2_subscriber_h_ref position_subscriber;
};

// no-op callback: draining only needs to clear the queue, not inspect events
static void drain_event(const iox2_event_id_t* event_id, uint64_t count, void* context) {
    (void) event_id;
    (void) count;
    (void) context;
}

// called whenever a listener attached to the WaitSet has received an event
static iox2_callback_progression_e on_event(iox2_waitset_attachment_id_h attachment_id, void* context) {
    struct CallbackContext* ctx = (struct CallbackContext*) context;

    uint64_t number_of_notifications = 0;

    if (iox2_waitset_attachment_id_has_event_from(&attachment_id, ctx->battery_guard)) {
        // drain every pending notification, otherwise the WaitSet wakes
        // again immediately and spins
        iox2_listener_try_wait(ctx->battery_listener, &number_of_notifications, drain_event, NULL);

        iox2_sample_h battery_sample = NULL;
        while (iox2_subscriber_receive(ctx->battery_subscriber, NULL, &battery_sample) == IOX2_OK
               && battery_sample != NULL) {
            struct BatteryState* payload = NULL;
            iox2_sample_payload(&battery_sample, (const void**) &payload, NULL);
            printf("battery: %.1f%%\n", (double) payload->charge_percent);
            iox2_sample_drop(battery_sample);
            battery_sample = NULL;
        }
    } else if (iox2_waitset_attachment_id_has_event_from(&attachment_id, ctx->position_guard)) {
        // drain every pending notification, otherwise the WaitSet wakes
        // again immediately and spins
        iox2_listener_try_wait(ctx->position_listener, &number_of_notifications, drain_event, NULL);

        iox2_sample_h position_sample = NULL;
        while (iox2_subscriber_receive(ctx->position_subscriber, NULL, &position_sample) == IOX2_OK
               && position_sample != NULL) {
            struct Position* payload = NULL;
            iox2_sample_payload(&position_sample, (const void**) &payload, NULL);
            printf("position: (%.1f, %.1f)\n", (double) payload->x, (double) payload->y);
            iox2_sample_drop(position_sample);
            position_sample = NULL;
        }
    }

    iox2_waitset_attachment_id_drop(attachment_id);
    return iox2_callback_progression_e_CONTINUE;
}

// NOLINTBEGIN(readability-function-size)
int main(void) {
    iox2_set_log_level_from_env_or(iox2_log_level_e_INFO);
    int ret_val = 0;

    // create node
    iox2_node_builder_h node_builder = iox2_node_builder_new(NULL);
    iox2_node_h node = NULL;
    ret_val = iox2_node_builder_create(node_builder, NULL, iox2_service_type_e_IPC, &node);
    if (ret_val != IOX2_OK) {
        printf("Could not create node! Error: %d\n", ret_val);
        goto end;
    }

    // create battery publish-subscribe service
    iox2_service_name_h battery_name = NULL;
    const char* battery_name_value = "larry/battery";
    iox2_service_name_new(NULL, battery_name_value, strlen(battery_name_value), &battery_name);
    iox2_service_builder_pub_sub_h battery_builder =
        iox2_service_builder_pub_sub(iox2_node_service_builder(&node, NULL, iox2_cast_service_name_ptr(battery_name)));
    iox2_service_builder_pub_sub_set_payload_type_details(&battery_builder,
                                                          iox2_type_variant_e_FIXED_SIZE,
                                                          BATTERY_STATE_TYPE_NAME,
                                                          strlen(BATTERY_STATE_TYPE_NAME),
                                                          sizeof(struct BatteryState),
                                                          alignof(struct BatteryState));
    iox2_port_factory_pub_sub_h battery_service = NULL;
    ret_val = iox2_service_builder_pub_sub_open_or_create(battery_builder, NULL, &battery_service);
    if (ret_val != IOX2_OK) {
        printf("Could not create battery service! Error: %d\n", ret_val);
        goto drop_node;
    }

    // create position publish-subscribe service
    iox2_service_name_h position_name = NULL;
    const char* position_name_value = "larry/position";
    iox2_service_name_new(NULL, position_name_value, strlen(position_name_value), &position_name);
    iox2_service_builder_pub_sub_h position_builder =
        iox2_service_builder_pub_sub(iox2_node_service_builder(&node, NULL, iox2_cast_service_name_ptr(position_name)));
    iox2_service_builder_pub_sub_set_payload_type_details(&position_builder,
                                                          iox2_type_variant_e_FIXED_SIZE,
                                                          POSITION_TYPE_NAME,
                                                          strlen(POSITION_TYPE_NAME),
                                                          sizeof(struct Position),
                                                          alignof(struct Position));
    iox2_port_factory_pub_sub_h position_service = NULL;
    ret_val = iox2_service_builder_pub_sub_open_or_create(position_builder, NULL, &position_service);
    if (ret_val != IOX2_OK) {
        printf("Could not create position service! Error: %d\n", ret_val);
        goto drop_battery_service;
    }

    // create battery event service
    iox2_service_builder_event_h battery_event_builder =
        iox2_service_builder_event(iox2_node_service_builder(&node, NULL, iox2_cast_service_name_ptr(battery_name)));
    iox2_port_factory_event_h battery_event_service = NULL;
    ret_val = iox2_service_builder_event_open_or_create(battery_event_builder, NULL, &battery_event_service);
    if (ret_val != IOX2_OK) {
        printf("Could not create battery event service! Error: %d\n", ret_val);
        goto drop_position_service;
    }

    // create position event service
    iox2_service_builder_event_h position_event_builder =
        iox2_service_builder_event(iox2_node_service_builder(&node, NULL, iox2_cast_service_name_ptr(position_name)));
    iox2_port_factory_event_h position_event_service = NULL;
    ret_val = iox2_service_builder_event_open_or_create(position_event_builder, NULL, &position_event_service);
    if (ret_val != IOX2_OK) {
        printf("Could not create position event service! Error: %d\n", ret_val);
        goto drop_battery_event_service;
    }

    // create subscribers
    iox2_subscriber_h battery_subscriber = NULL;
    iox2_port_factory_subscriber_builder_create(
        iox2_port_factory_pub_sub_subscriber_builder(&battery_service, NULL), NULL, &battery_subscriber);
    iox2_subscriber_h position_subscriber = NULL;
    iox2_port_factory_subscriber_builder_create(
        iox2_port_factory_pub_sub_subscriber_builder(&position_service, NULL), NULL, &position_subscriber);

    // create listeners
    iox2_listener_h battery_listener = NULL;
    iox2_port_factory_listener_builder_create(
        iox2_port_factory_event_listener_builder(&battery_event_service, NULL), NULL, &battery_listener);
    iox2_listener_h position_listener = NULL;
    iox2_port_factory_listener_builder_create(
        iox2_port_factory_event_listener_builder(&position_event_service, NULL), NULL, &position_listener);

    // create waitset and attach both listeners
    iox2_waitset_builder_h waitset_builder = NULL;
    iox2_waitset_builder_new(NULL, &waitset_builder);
    iox2_waitset_h waitset = NULL;
    ret_val = iox2_waitset_builder_create(waitset_builder, iox2_service_type_e_IPC, NULL, &waitset);
    if (ret_val != IOX2_OK) {
        printf("Could not create waitset! Error: %d\n", ret_val);
        goto drop_listeners;
    }

    iox2_waitset_guard_h battery_guard = NULL;
    ret_val = iox2_waitset_attach_notification(
        &waitset, iox2_listener_get_file_descriptor(&battery_listener), NULL, &battery_guard);
    if (ret_val != IOX2_OK) {
        printf("Could not attach battery listener! Error: %d\n", ret_val);
        goto drop_waitset;
    }

    iox2_waitset_guard_h position_guard = NULL;
    ret_val = iox2_waitset_attach_notification(
        &waitset, iox2_listener_get_file_descriptor(&position_listener), NULL, &position_guard);
    if (ret_val != IOX2_OK) {
        printf("Could not attach position listener! Error: %d\n", ret_val);
        goto drop_battery_guard;
    }

    struct CallbackContext context;
    context.battery_guard = &battery_guard;
    context.position_guard = &position_guard;
    context.battery_listener = &battery_listener;
    context.position_listener = &position_listener;
    context.battery_subscriber = &battery_subscriber;
    context.position_subscriber = &position_subscriber;

    iox2_waitset_run_result_e result = iox2_waitset_run_result_e_STOP_REQUEST;
    ret_val = iox2_waitset_wait_and_process(&waitset, on_event, (void*) &context, &result);
    if (ret_val != IOX2_OK) {
        printf("Failure in WaitSet wait_and_process loop! Error: %d\n", ret_val);
    }

    iox2_waitset_guard_drop(position_guard);
drop_battery_guard:
    iox2_waitset_guard_drop(battery_guard);
drop_waitset:
    iox2_waitset_drop(waitset);
drop_listeners:
    iox2_listener_drop(position_listener);
    iox2_listener_drop(battery_listener);
    iox2_subscriber_drop(position_subscriber);
    iox2_subscriber_drop(battery_subscriber);
    iox2_port_factory_event_drop(position_event_service);
drop_battery_event_service:
    iox2_port_factory_event_drop(battery_event_service);
drop_position_service:
    iox2_port_factory_pub_sub_drop(position_service);
drop_battery_service:
    iox2_port_factory_pub_sub_drop(battery_service);
drop_node:
    iox2_node_drop(node);
end:
    return ret_val;
}
// NOLINTEND(readability-function-size)

Neither application is aware that the data will be propagated over the network. They are both set up as regular iceoryx2 applications.

Extending over the network¶

With the applications set up, we now need to spin up the gateway to extend the communication over the network. The gateway can either be run in a separate process or embedded into the application. First, let’s use the iceoryx2-cli to run the gateway in a separate process.

Installing the CLI¶

First install the iceoryx2-cli which is the entry point for all CLI commands:

cargo install iceoryx2-cli

Next, the CLI for a specific gateway backend must be installed. For this article, we use the Zenoh backend:

cargo install iceoryx2-integrations-zenoh-gateway-cli

The command iox2 gateway --list can be used to verify that the installation was successful. This command lists all installed gateway backends discovered by the CLI.

The gateway should be installed on both hosts.

Running a gateway process¶

The CLI can now be used to start a gateway process on both hosts:

iox2 gateway zenoh

Each gateway discovers iceoryx2 services, both on its own host and over the communication mechanism used by the chosen backend implementation. Larry’s gateway sees larry/battery and larry/position and forwards all samples published to them over the wire. The dashboard gateway receives these samples from the wire, and injects them into the iceoryx2 shared memory to be received by subscribers to matching services on the dashboard host. The two applications communicate over services, as if they were running on the same host.

By default the gateway polls for new services and pending samples every 100 ms, however since the event-based communication pattern is used, it is possible to configure the gateway to execute reactively.

The gateway running on Larry only has work to do when new samples are published on the services that need to be propagated:

iox2 gateway zenoh --listener "larry/battery" --listener "larry/position"

The gateway running on the dashboard host only has work to do when data arrives over the backend:

iox2 gateway zenoh --reactive-backend

Nothing here is application code. Both applications were written, compiled, and deployed with no knowledge that a gateway exists. Communicating across the hosts is purely a deployment decision.

Embedding the gateway¶

Note

The embeddable gateway is currently only available as a Rust library. C++, C, and Python applications must use the gateway from a separate process with the iox2 gateway CLI shown above; the remainder of this section applies to Rust.

Running the gateway as its own process is convenient, but it’s not the only option. The gateway is a library, and it spawns no threads of its own so you have full control over when and where to run it. It can be executed within a dedicated time slot of a time-triggered architecture, or on a separate thread that you control, pinned to a specific core so as not to interfere with application execution.

First add the generic gateway and your chosen backend as dependencies:

iceoryx2-gateway = { version = "X.Y.Z" }
iceoryx2-integrations-zenoh-gateway-backend = { version = "X.Y.Z" }

The gateway can then be instantiated in your application via the provided builder. The chosen backend implementation is passed as a generic on creation, here we again use the ZenohBackend, but any other implementation can be dropped in, even your own. Specifying polled() in the build returns a gateway that can be manually driven:

use core::time::Duration;
use iceoryx2::prelude::*;
use iceoryx2_gateway::Gateway;
use iceoryx2_integrations_zenoh_gateway_backend::ZenohBackend;

const POLL_INTERVAL: Duration = Duration::from_millis(100);

let mut gateway = Gateway::<ipc::Service, ZenohBackend<ipc::Service>>::new()
    .polled()
    .create()?;

while gateway.node().wait(POLL_INTERVAL).is_ok() {
    gateway.discover()?;
    gateway.propagate()?;
}

Execution is driven via two API calls. The discover() call reconciles which services exist locally and over the wire. The propagate() call moves the data in both directions: from the wire into iceoryx2 shared memory and vice versa. Calling the two on the polling gateway in a timed loop replicates the default behavior of the CLI.

Now let’s take a look at how to replicate the reactive configurations provided by the gateway. Recapping the details, both gateways must be set up to react to different things:

Larry’s gateway reacts to the services notified when a sample is published
The dashboard’s gateway reacts to data arriving over the network

Since Larry’s gateway does not react to network communication, it can be created in polling mode. Listeners are created for the services notified with each publish, and a WaitSet is set up to react to notifications from both of them in a single thread:

use iceoryx2::prelude::*;
use iceoryx2_gateway::Gateway;
use iceoryx2_integrations_zenoh_gateway_backend::ZenohBackend;

let mut gateway = Gateway::<ipc::Service, ZenohBackend<ipc::Service>>::new()
    .polled()
    .create()?;

// wake whenever Larry publishes locally, so freshly produced samples are
// pushed out promptly
let battery_listener = gateway
    .node()
    .service_builder(&"larry/battery".try_into()?)
    .event()
    .open_or_create()?
    .listener_builder()
    .create()?;
let position_listener = gateway
    .node()
    .service_builder(&"larry/position".try_into()?)
    .event()
    .open_or_create()?
    .listener_builder()
    .create()?;

let waitset = WaitSetBuilder::new().create::<ipc::Service>()?;
let _battery_guard = waitset.attach_notification(&battery_listener)?;
let _position_guard = waitset.attach_notification(&position_listener)?;

waitset.wait_and_process(|_| {
    let _ = battery_listener.try_wait(|_| {});
    let _ = position_listener.try_wait(|_| {});
    let _ = gateway.discover();
    let _ = gateway.propagate();

    CallbackProgression::Continue
})?;

The dashboard gateway only cares about data arriving over the network, so it is created in reactive mode, which provides a listener along with the created gateway. This listener is notified when new data is available from the backend specified on creation. No WaitSet is needed here, since there is only one listener to wait on:

use iceoryx2::prelude::*;
use iceoryx2_gateway::Gateway;
use iceoryx2_integrations_zenoh_gateway_backend::ZenohBackend;

// reactive mode hands back a listener that wakes when the backend has
// delivered something to ingest from the wire
let (mut gateway, listener) = Gateway::<ipc::Service, ZenohBackend<ipc::Service>>::new()
    .reactive()
    .create()?;

while listener.blocking_wait(|_| {}).is_ok() {
    gateway.discover()?;
    gateway.propagate()?;
}

Both approaches run the same discover() and propagate() calls; they differ only in what decides when those calls happen. Once again, the gateway has no background threads, so you are free to choose where and when these operations run, allowing you to plan for and minimize unpredictability.

Switching the mechanism¶

Let’s say that later in development you decide a different mechanism should carry the cross-host communication. Perhaps you want to reach the dashboard over the internet by integrating with existing MQTT infrastructure, or you need to bridge co-processors on a single board that each keep their own memory and can only communicate over a cross-chip API.

All that is required, is to switch out the chosen backend when creating the gateway. In our examples above, we have been using Zenoh, so the gateways were created with:

Gateway::<ipc::Service, ZenohBackend<ipc::Service>>

The first parameter is the local iceoryx2 service type; the second is the backend. The gateway itself knows nothing about Zenoh. It is written against the backend traits, and discover() and propagate() are the same calls regardless of what carries the bytes.

Switching over to a different mechanism is therefore only a change of the type specified on gateway creation:

Gateway::<ipc::Service, SomeOtherBackend<ipc::Service>>

The publishers, the subscribers, the embed loops, and the choice of polled or reactive mode are all untouched.

Over time, additional backend implementations will be provided upstream alongside iceoryx2, however if there isn’t an existing implementation for the mechanism you would like to use, it is possible to implement your own using the traits provided in iceoryx2-gateway-backend.

Cross-host communication¶

Application setup¶

Extending over the network¶

Installing the CLI¶

Running a gateway process¶

Embedding the gateway¶

Switching the mechanism¶

Further Reading¶