Request-Response

With emergency brakes and some basic fault handling in place, it’s time for something a bit more sophisticated.

Imagine the user wants to send our robot Larry to a specific position: “Go to point B.” As Larry starts driving, the user expects status updates - position, speed, progress. At some point the user may even cancel the trip, or Larry himself may have to abort if he encounters, say, a canyon.

Client App Server Larry Autopilot time Request: "Go to point B" Response: "Starting drive…" Response: "At (10, 5), speed (1, 0)" Response: "At (20, 5), speed (1, 0)" Response: "At (30, 5), speed (0, 0)" Response: "Arrived at destination!" Cancel signal (drop pending response)

Request-Response interactions between App and Larry Autopilot

This is where the request-response (stream) pattern in iceoryx2 comes into play. It behaves like regular non-blocking request-response, but with one extra trick: the server can send a stream of responses instead of just one. Either side can end the stream at any time if it’s no longer interested.

Client

The client is our user-facing app. The request can be a simple 2D position:

#[derive(Debug, ZeroCopySend)]
#[repr(C)]
pub struct Position {
    pub position: [f32; 2],
}

import ctypes


class Position(ctypes.Structure):
    _fields_ = [
        ("position", ctypes.c_float * 2),
    ]

struct Position {
    float position[2];
};

struct Position {
    float position[2];
};

The server responds with Larry’s current state:

#[derive(Debug, ZeroCopySend)]
#[repr(C)]
pub struct State {
    pub position: [f32; 2],
    pub speed: [f32; 2],
}

class State(ctypes.Structure):
    _fields_ = [
        ("position", ctypes.c_float * 2),
        ("speed", ctypes.c_float * 2),
    ]

struct State {
    float position[2];
    float speed[2];
};

struct State {
    float position[2];
    float speed[2];
};

First we create a node:

use iceoryx2::prelude::*;

let node = NodeBuilder::new().create::<ipc::Service>()?;

import iceoryx2 as iox2

node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)

auto node = NodeBuilder().create<ServiceType::Ipc>().value();

iox2_node_builder_h node_builder = iox2_node_builder_new(NULL);
iox2_node_h node = NULL;
if (iox2_node_builder_create(node_builder, NULL, iox2_service_type_e_IPC, &node) != IOX2_OK) {
    printf("Could not create node!\n");
    exit(-1);
}

Then the service, specifying request and response types:

let service = node
    .service_builder(&"autopilot".try_into()?)
    .request_response::<Position, State>()
    .open_or_create()?;

service = (
    node.service_builder(iox2.ServiceName.new("autopilot"))
    .request_response(Position, State)
    .open_or_create()
)

auto service = node.service_builder(ServiceName::create("autopilot").value())
                   .request_response<Position, State>()
                   .open_or_create()
                   .value();

const char* service_name_value = "autopilot";
iox2_service_name_h service_name = NULL;
if (iox2_service_name_new(NULL, service_name_value, strlen(service_name_value), &service_name) != IOX2_OK) {
    printf("Unable to create service name!\n");
    exit(-1);
}

iox2_service_name_ptr service_name_ptr = iox2_cast_service_name_ptr(service_name);
iox2_service_builder_h service_builder = iox2_node_service_builder(&node, NULL, service_name_ptr);
iox2_service_builder_request_response_h service_builder_request_response =
    iox2_service_builder_request_response(service_builder);

const char* request_type_name = "Position";
if (iox2_service_builder_request_response_set_request_payload_type_details(&service_builder_request_response,
                                                                           iox2_type_variant_e_FIXED_SIZE,
                                                                           request_type_name,
                                                                           strlen(request_type_name),
                                                                           sizeof(struct Position),
                                                                           alignof(struct Position))
    != IOX2_OK) {
    printf("Unable to set request type details\n");
    exit(-1);
}

const char* response_type_name = "State";
if (iox2_service_builder_request_response_set_response_payload_type_details(&service_builder_request_response,
                                                                            iox2_type_variant_e_FIXED_SIZE,
                                                                            response_type_name,
                                                                            strlen(response_type_name),
                                                                            sizeof(struct State),
                                                                            alignof(struct State))
    != IOX2_OK) {
    printf("Unable to set response type details\n");
    exit(-1);
}

iox2_port_factory_request_response_h service = NULL;
if (iox2_service_builder_request_response_open_or_create(service_builder_request_response, NULL, &service)
    != IOX2_OK) {
    printf("Unable to create service!\n");
    exit(-1);
}

And create the client port:

let client = service.client_builder().create()?;

client = service.client_builder().create()

auto client = service.client_builder().create().value();

iox2_port_factory_client_builder_h client_builder =
    iox2_port_factory_request_response_client_builder(&service, NULL);
iox2_client_h client = NULL;
if (iox2_port_factory_client_builder_create(client_builder, NULL, &client) != IOX2_OK) {
    printf("Unable to create client!\n");
    exit(-1);
}

Now we can send a request. The returned pending_response object represents the stream of updates for this request. Dropping it automatically signals the server to stop sending further responses.

let request = client.loan_uninit()?;
let request = request.write_payload(Position {
    position: [123.456, 789.1],
});
let pending_response = request.send()?;

request = client.loan_uninit()
position = Position()
position.position[0] = 123.456
position.position[1] = 789.1
request = request.write_payload(position)
pending_response = request.send()

auto request = client.loan_uninit().value();
auto initialized_request = request.write_payload(Position { { 123.456F, 789.1F } }); // NOLINT

auto pending_response = send(std::move(initialized_request)).value();

iox2_request_mut_h request = NULL;
if (iox2_client_loan_slice_uninit(&client, NULL, &request, 1) != IOX2_OK) {
    printf("Failed to loan request\n");
    exit(-1);
}

struct Position* payload = NULL;
iox2_request_mut_payload_mut(&request, (void**) &payload, NULL);
payload->position[0] = 123.456F; // NOLINT
payload->position[1] = 789.1F;   // NOLINT

iox2_pending_response_h pending_response = NULL;
if (iox2_request_mut_send(request, NULL, &pending_response) != IOX2_OK) {
    printf("Failed to send request\n");
    exit(-1);
}

In the event loop we poll every 100 ms and handle three cases:

  1. User pressed “stop” → drop pending_response to cancel.

  2. Larry disconnected before reaching the goal → show “obstacle detected.”

  3. Normal update → display current position.

while node.wait(Duration::from_millis(100)).is_ok() {
    while let Some(response) = pending_response.receive()? {
        show_larry_position_in_app(&response);

        if !pending_response.is_connected() {
            if is_at_destination(&response) {
                show_larry_arrived_popup_in_app();
            } else {
                show_larry_encountered_obstacle_in_app();
            }
        }
    }

    if user_has_pressed_stop_button() {
        // dropping the pending response cancels the stream
        drop(pending_response);
        break;
    }
}

try:
    while True:
        node.wait(iox2.Duration.from_millis(100))

        while True:
            response = pending_response.receive()
            if response is None:
                break

            show_larry_position_in_app(response)

            if not pending_response.is_connected():
                if is_at_destination(response):
                    show_larry_arrived_popup_in_app()
                else:
                    show_larry_encountered_obstacle_in_app()

        if user_has_pressed_stop_button():
            break

except iox2.NodeWaitFailure:
    print("exit")

while (node.wait(iox2::bb::Duration::from_millis(100)).has_value()) {
    auto response = pending_response.receive().value();
    while (response.has_value()) {
        show_larry_position_in_app(response.value());

        if (!pending_response.is_connected()) {
            if (is_at_destination(response.value())) {
                show_larry_arrived_popup_in_app();
            } else {
                show_larry_encountered_obstacle_in_app();
            }
        }

        response = pending_response.receive().value();
    }

    if (user_has_pressed_stop_button()) {
        // dropping the pending response cancels the stream
        break;
    }
}

while (iox2_node_wait(&node, 0, 100000000) == IOX2_OK) {
    iox2_response_h response = NULL;
    while (true) {
        response = NULL;
        if (iox2_pending_response_receive(&pending_response, NULL, &response) != IOX2_OK) {
            printf("Failed to receive response\n");
            exit(-1);
        }
        if (response == NULL) {
            break;
        }

        const struct State* state = NULL;
        iox2_response_payload(&response, (const void**) &state, NULL);
        show_larry_position_in_app(state);

        if (!iox2_pending_response_is_connected(&pending_response)) {
            if (is_at_destination(state)) {
                show_larry_arrived_popup_in_app();
            } else {
                show_larry_encountered_obstacle_in_app();
            }
        }

        iox2_response_drop(response);
    }

    if (user_has_pressed_stop_button()) {
        // dropping the pending response cancels the stream
        break;
    }
}

// release every handle once the loop exits
iox2_pending_response_drop(pending_response);
iox2_client_drop(client);
iox2_port_factory_request_response_drop(service);
iox2_service_name_drop(service_name);
iox2_node_drop(node);

Server

On the server side, the setup looks similar:

use iceoryx2::prelude::*;

let node = NodeBuilder::new().create::<ipc::Service>()?;
let service = node
    .service_builder(&"autopilot".try_into()?)
    .request_response::<Position, State>()
    .open_or_create()?;

let server = service.server_builder().create()?;

import iceoryx2 as iox2

node = iox2.NodeBuilder.new().create(iox2.ServiceType.Ipc)

service = (
    node.service_builder(iox2.ServiceName.new("autopilot"))
    .request_response(Position, State)
    .open_or_create()
)

server = service.server_builder().create()

auto node = NodeBuilder().create<ServiceType::Ipc>().value();

auto service = node.service_builder(ServiceName::create("autopilot").value())
                   .request_response<Position, State>()
                   .open_or_create()
                   .value();

auto server = service.server_builder().create().value();

iox2_node_builder_h node_builder = iox2_node_builder_new(NULL);
iox2_node_h node = NULL;
if (iox2_node_builder_create(node_builder, NULL, iox2_service_type_e_IPC, &node) != IOX2_OK) {
    printf("Could not create node!\n");
    exit(-1);
}

const char* service_name_value = "autopilot";
iox2_service_name_h service_name = NULL;
if (iox2_service_name_new(NULL, service_name_value, strlen(service_name_value), &service_name) != IOX2_OK) {
    printf("Unable to create service name!\n");
    exit(-1);
}

iox2_service_name_ptr service_name_ptr = iox2_cast_service_name_ptr(service_name);
iox2_service_builder_h service_builder = iox2_node_service_builder(&node, NULL, service_name_ptr);
iox2_service_builder_request_response_h service_builder_request_response =
    iox2_service_builder_request_response(service_builder);

const char* request_type_name = "Position";
if (iox2_service_builder_request_response_set_request_payload_type_details(&service_builder_request_response,
                                                                           iox2_type_variant_e_FIXED_SIZE,
                                                                           request_type_name,
                                                                           strlen(request_type_name),
                                                                           sizeof(struct Position),
                                                                           alignof(struct Position))
    != IOX2_OK) {
    printf("Unable to set request type details\n");
    exit(-1);
}

const char* response_type_name = "State";
if (iox2_service_builder_request_response_set_response_payload_type_details(&service_builder_request_response,
                                                                            iox2_type_variant_e_FIXED_SIZE,
                                                                            response_type_name,
                                                                            strlen(response_type_name),
                                                                            sizeof(struct State),
                                                                            alignof(struct State))
    != IOX2_OK) {
    printf("Unable to set response type details\n");
    exit(-1);
}

iox2_port_factory_request_response_h service = NULL;
if (iox2_service_builder_request_response_open_or_create(service_builder_request_response, NULL, &service)
    != IOX2_OK) {
    printf("Unable to create service!\n");
    exit(-1);
}

iox2_port_factory_server_builder_h server_builder =
    iox2_port_factory_request_response_server_builder(&service, NULL);
iox2_server_h server = NULL;
if (iox2_port_factory_server_builder_create(server_builder, NULL, &server) != IOX2_OK) {
    printf("Unable to create server!\n");
    exit(-1);
}

The server checks every 100 ms for new requests. (You could combine this with event messaging to block until a request arrives, but for the purpose of this tutorial, we keep it simple.) When a request is received, Larry starts driving and periodically sends status updates.

If an obstacle is detected, or the client disconnects, the active request is dropped and Larry stops.

let mut active_request = None;
while node.wait(Duration::from_millis(100)).is_ok() {
    if active_request.is_none() {
        if let Some(new_request) = server.receive()? {
            drive_to_position(&new_request);
            active_request = Some(new_request);
        }
    }

    if let Some(current_request) = active_request.take() {
        if !current_request.is_connected() {
            stop_driving();
        } else if !arrived_at_destination() {
            let response = current_request.loan_uninit()?;
            let response = response.write_payload(get_current_state());
            response.send()?;
            active_request = Some(current_request);
        }
    }
}

active_request = None
try:
    while True:
        node.wait(iox2.Duration.from_millis(100))

        if active_request is None:
            active_request = server.receive()
            if active_request is not None:
                drive_to_position(active_request)

        if active_request is not None:
            if not active_request.is_connected():
                stop_driving()
                active_request = None
            elif arrived_at_destination():
                active_request = None
            else:
                response = active_request.loan_uninit()
                response = response.write_payload(get_current_state())
                response.send()

except iox2.NodeWaitFailure:
    print("exit")

auto active_request = server.receive().value();
while (node.wait(iox2::bb::Duration::from_millis(100)).has_value()) {
    if (!active_request.has_value()) {
        active_request = server.receive().value();
        if (active_request.has_value()) {
            drive_to_position(active_request.value());
        }
    }

    if (active_request.has_value()) {
        if (!active_request->is_connected()) {
            stop_driving();
            active_request.reset();
        } else if (arrived_at_destination()) {
            active_request.reset();
        } else {
            auto response = active_request->loan_uninit().value();
            auto initialized_response = response.write_payload(get_current_state());
            send(std::move(initialized_response)).value();
        }
    }
}

iox2_active_request_h active_request = NULL;
while (iox2_node_wait(&node, 0, 100000000) == IOX2_OK) {
    if (active_request == NULL) {
        if (iox2_server_receive(&server, NULL, &active_request) != IOX2_OK) {
            printf("Failed to receive request\n");
            exit(-1);
        }

        if (active_request != NULL) {
            const struct Position* position = NULL;
            iox2_active_request_payload(&active_request, (const void**) &position, NULL);
            drive_to_position(position);
        }
    }

    if (active_request != NULL) {
        if (!iox2_active_request_is_connected(&active_request)) {
            stop_driving();
            iox2_active_request_drop(active_request);
            active_request = NULL;
        } else if (arrived_at_destination()) {
            iox2_active_request_drop(active_request);
            active_request = NULL;
        } else {
            iox2_response_mut_h response = NULL;
            if (iox2_active_request_loan_slice_uninit(&active_request, NULL, &response, 1) != IOX2_OK) {
                printf("Failed to loan response\n");
                exit(-1);
            }

            struct State* state = NULL;
            iox2_response_mut_payload_mut(&response, (void**) &state, NULL);
            get_current_state(state);

            if (iox2_response_mut_send(response) != IOX2_OK) {
                printf("Failed to send response\n");
                exit(-1);
            }
        }
    }
}

// release every handle once the loop exits
iox2_server_drop(server);
iox2_port_factory_request_response_drop(service);
iox2_service_name_drop(service_name);
iox2_node_drop(node);

This way, the client can receive a continuous stream of updates, either side can gracefully stop when necessary, and Larry doesn’t drive blindly into canyons (in theory).