Request-Response

The request-response service enables a direct, unicast exchange of data between two nodes. Unlike pub-sub, which broadcasts to many subscribers, request-response expects a direct response for every request sent.

This pattern is especially useful when a node wants to request another node’s ability to handle a specific task e.g., checking battery levels, resource availability, or configuration setting.

Mobile Robots Battery Status Tutorial Jump to a more involved tutorial that uses the request-response pattern.

The Request-Response service

Python

from hyveos_sdk import Connection
import asyncio


async with Connection() as connection:
    req_resp = connection.get_request_response_service()

    # continue with usage of req_resp
    # TODO

Rust

use hyveos_sdk::Connection;


#[tokio::main]
async fn main() {
    let connection = Connection::new().await.unwrap();
    let mut req_resp = connection.req_resp();

    // continue with usage of req_resp
    // todo!()
}

Typescript/JS

import { Client } from 'hyveos-sdk'
import { Connection } from 'hyveos-web'


async function main() {
    const transport = new Connection('http://localhost:8080')
    const client = new Client(transport)
    const reqResp = client.reqresp

    // continue with usage of reqResp
    // TODO
}

Example 1. Obtain the Request-Response service handler. Note the async environment.

Sending Requests

To query a specific node for data or ask it to perform an action, send a request. Typically, it will hold the target peer’s ID and an optional topic to which that peer is registered. Topics can and should mainly be used to logically group requests.

The receiving peer must be subscribed to the topic specified in the request in order to receive the request.

Python

requesting_node.py

from hyveos_sdk import Connection
import asyncio


async with Connection() as connection:
    req_resp = connection.get_request_response_service()

    peer_id = "0123456789ABCDEF"
    topic = "robot_status"
    message_data = "give_status"

    response = await req_resp.send_request(peer_id, message_data, topic)

    response_data = loads(response.data.data)
    print(f"Received response: {response_data}")

asyncio.run(main())

Rust

requesting_node.rs

use hyveos_sdk::Connection;
use futures::StreamExt as _;


#[tokio::main]
async fn main() {
    let connection = Connection::new().await.unwrap();
    let mut req_resp = connection.req_resp();

    let peer_id = String::from("0123456789ABCDEF");
    let topic = Some("robot_status");
    let message_data = String::from("give_status");

    let response = req_resp.send_request(peer_id, message_data, topic).await.unwrap();

    let data = Result::from(response).unwrap();
    println!("Received response: {}", String::from_utf8(data).unwrap());
}

Typescript/JS

requesting_node.ts

import { Client } from 'hyveos-sdk'
import { Connection } from 'hyveos-web'


async function main() {
    const transport = new Connection('http://localhost:8080')
    const client = new Client(transport)
    const reqResp = client.reqresp

    const peerId = "0123456789ABCDEF"
    const topic = "robot_status"
    const messageData = "give_status"

    const response = await reqResp.request(peerId, messageData, topic)

    const decodedResponse = new TextDecoder().decode(response)
    console.log('Received response:', decodedResponse)
}

Example 2. Simple request of status information through request/response without error handling. You might want to include try/catch in actual production code.

Actual peer_id

The actual peer id can be found with Discovery.
Here, we used an exemplary "0123456789ABCDEF".

Intricacies for Sending a Request

• If the request specified None as topic, the receiving peer must also be subscribed to None. Otherwise, the receiving peer will never handle the request.
• Make sure whatever data you’re sending is serializable into a String or some binary format recognized on both sides. You can use the handle to the request/response service that has JSON- or CBOR-encoded requests and responses.
• Since requests are asynchronous, you won’t get the response in the same call to send. Instead, you’ll have to listen for it explicitly (usually in a separate part of your logic).

Proper Error Handling

send_request() will always return a Result/Promise wrapper, the return value being either Response or Error.

Receiving & Responding to Requests

The remote node will have to react to the sent request.

Workflow

1. Receive the message through an incoming stream at the remote node
2. Process the message
3. Respond, with a handler (currently in Rust and Typescript, but not Python yet)

Intricacies in Receiving and Responding to Requests

• Topic Mismatch
  If the responder subscribes to a topic (say, robot_status), the remote node will receive requests explicitly sent to robot_status. If the requester sends to a different topic, you won’t see their request. If you subscribe with None, you only see requests that also have None as their topic.
• Concurrency
  Your node might be handling multiple inbound requests concurrently. Make sure your application logic can handle concurrency, especially if responses require unique resources.
• Error Handling
  Instead of forcibly sending a data payload when something goes wrong, use respond_with_error() or the typed JSON/CBOR versions.

Receive a Request

Messages will be received through a stream of inbound receive messages.

Python

receiving_node.py

from hyveos_sdk import Connection
import asyncio


async def main():
    async with Connection() as connection:
        req_resp = connection.get_request_response_service()

        topic = "robot_status"

        async with req_resp.receive(topic) as requests:
            # Handle stream of incoming requests -> RecvRequest
            # TODO
            pass

asyncio.run(main())

Rust

receiving_node.rs

use hyveos_sdk::Connection;
use futures::TryStreamExt as _;


#[tokio::main]
async fn main() {
    let connection = Connection::new().await.unwrap();
    let mut req_resp = connection.req_resp();

    let topic = Some("robot_status");
    let mut request_stream = req_resp.recv(topic).await.unwrap();

    // Handle stream of incoming requests -> InboundRequest
    // todo!()
}

Typescript/JS

receiving_node.ts

import { Client } from 'hyveos-sdk'
import { Connection } from 'hyveos-web'


async function main() {
    const transport = new Connection('http://localhost:8080')
    const client = new Client(transport)
    const reqResp = client.reqresp

    const topic = "robot_status"

    const requestStream = reqResp.recv(topic)

    // Handle stream of incoming requests -> IncomingRequestHandle
    // TODO
}

Example 3. Receiving the request of giving the status information. Example 4 below will explain the processing of inbound requests.

Process Inbound Request

In the Pyhton SDK, you explicitly receive a seq number upon receive(). seq is then given in respond() for matching the request to the eventual response.

The Rust and Typescript SDKs have a InboundRequestHandle object for takes care of this.

Each inbound message in Rust comes in a (request, handle) pair. request is of type InboundRequest, handle is of type InboundRequestHandle.

Incoming Request Objects

In Rust, handlers for Request Processing exist.

InboundRequest

pub struct InboundRequest<T> {
    pub peer_id: PeerId,
    pub topic: Option<String>,
    pub data: T,
}

InboundRequestHandle

pub struct InboundRequestHandle<'a> {
    id: u64,
    service: Service,
    phantom: PhantomData<&'a ()>,
}

Retrieve the the fields of request when handling the inbound request.

Python

receiving_node.py

            # continuing Example 3
            async for request in requests:
                print(f"Request from {request.peer.peer_id}")
                seq = request.seq
                received = request.msg
                print(f"Data: {received}")

                # respond with seq
                # TODO

Rust

receiving_node.rs

    // continuing Example 3
    while let Some(Ok(request, handle)) = request_stream.try_next().await.unwrap() {
        println!("Request from {} on topic {:?}", request.peer_id, request.topic);

        let received = String::from_utf8(request.data).unwrap_or_default();
        println!("Data: {}", received);

        // For responding, use the handler
        // todo!()

Typescript/JS

receiving_node.ts

    // continuing Example 3
    for await (const request of requestStream) {
        console.log(`Request from ${request.peerId} on topic ${request.topic}`)

        const received = new TextDecoder().decode(request.data)
        console.log(`Data: ${received}`)

        // For responding, use the handler
        // TODO

Example 4. Retrieving fields from the request. In Rust and Typescript, this means leveraging InboundRequest.

Responding to a request

Python

receiving_node.py

                # Continuing from Example 4
                if received is "give_status":
                    await req_resp.respond(seq, "All systems operational.")
                else:
                    await req_resp.respond(seq, None, "Unknown request") # Error

Rust

receiving_node.rs

        // Continuing from Example 4
        if received == "give_status" {
            handle.respond("All systems operational.".as_bytes()).await.unwrap();
        } else {
            handle.respond_with_error("Unknown request").await.unwrap();
        }
    }
}

Typescript/JS

receiving_node.ts

        // Continuing from Example 4
        if (received === "give_status") {
            await reqResp.respond(new TextEncoder().encode('All systems operational.'))
        } else {
            await request.error('Unknown request')
        }
    }
}

Example 5. Use the InboundRequestHandle for responding. In Rust and Typescript, you can respond with an error as well. Note that in Python in case of an error, you can leave the data argument to None and give an error message in the last argument—here: "Unknown request".

Pyhton responding incorporates explicitly giving the seq number in the argument of respond().

JSON and CBOR Request/Response Services

While raw bytes give you flexibility, you’ll often want structured data. hyveOS offers typed request/response services for JSON and CBOR.

The same code examples apply analogously.

Typed Services: JSON & CBOR

The Request/Response JSON and CBOR services.

JSON

let mut req_resp_service = connection.req_resp_json();

CBOR

let mut req_resp_service = connection.req_resp_cbor();

Example 6. Obtaining the JSON and CBOR request/response services.

All methods send_request(), recv() and respond() are in called the same way for these services. The JSON and CBOR services are just convenient data format wrapper services.