Service client not Send

[kristoferB]

If i reimplement the client example with tokio, i get the following error:

future cannot be sent between threads safely
within `r2r::WrappedClient<r2r::example_interfaces::srv::AddTwoInts::Service>`, the trait `std::marker::Send` is not implemented for `*mut rcl::rcl_client_impl_t`rustc
client_tokio.rs(1, 1): required by a bound in this
spawn.rs(127, 21): required by this bound in `tokio::spawn`
client_tokio.rs(24, 20): future is not `Send` as this value is used across an await
client_tokio.rs(24, 64): `client` is later dropped here
client_tokio.rs(24, 20): consider moving this into a `let` binding to create a shorter lived borrow
fn spawn<impl Future<Output = ()>>(impl Future<Output = ()>) -> JoinHandle<<impl Future<Output = ()> as Future>::Output>

My code:

use futures::stream::StreamExt;
use r2r;
use r2r::example_interfaces::srv::AddTwoInts;


#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let ctx = r2r::Context::create()?;
    let mut node = r2r::Node::create(ctx, "testnode", "")?;

    let mut service = node.create_service::<AddTwoInts::Service>("/add_two_ints")?;
    let client = node.create_client::<AddTwoInts::Service>("/add_two_ints")?;

    println!("waiting for service...");
    while !node.service_available(&client).unwrap() {
        std::thread::sleep(std::time::Duration::from_millis(1000));
    }
    println!("found it");

    tokio::spawn(async move {
        loop {
            let req = service.next().await.unwrap();
            let resp = AddTwoInts::Response {
                sum: req.message.a + req.message.b,
            };
            req.respond(resp);
        }
    });

    tokio::spawn(async move {
        let req = AddTwoInts::Request { a: 10 * 2, b: 3 };
        let resp = client.request(&req).unwrap().await.unwrap();
        println!("{}", resp.sum);
    });

    let handle = std::thread::spawn(move || loop {
        node.spin_once(std::time::Duration::from_millis(100));
    });

    
    handle.join().unwrap();

    Ok(())
}

But this works:

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let ctx = r2r::Context::create()?;
    let mut node = r2r::Node::create(ctx, "testnode", "")?;

    let mut service = node.create_service::<AddTwoInts::Service>("/add_two_ints")?;
    let client = node.create_client::<AddTwoInts::Service>("/add_two_ints")?;

    println!("waiting for service...");
    while !node.service_available(&client).unwrap() {
        std::thread::sleep(std::time::Duration::from_millis(1000));
    }
    println!("found it");

    tokio::spawn(async move {
        loop {
            let req = service.next().await.unwrap();
            let resp = AddTwoInts::Response {
                sum: req.message.a + req.message.b,
            };
            req.respond(resp);
        }
    });

    let handle = std::thread::spawn(move || loop {
        node.spin_once(std::time::Duration::from_millis(100));
    });

    let req = AddTwoInts::Request { a: 10 * 2, b: 3 };
    let resp = client.request(&req).unwrap().await.unwrap();
    println!("{}", resp.sum);

    
    handle.join().unwrap();

    Ok(())
}

Even if I create the client inside the spawn with access to the node via Arc, i get the same error. Maybe it is possible to just make it safe?

[m-dahl]

Not an r2r issue, closing.

[kristoferB]

So the solution was:

tokio::spawn(async move {
        let req = AddTwoInts::Request { a: 10 * 2, b: 3 };
        let resp = client.request(&req).unwrap();
        let resp = resp.await.unwrap();
        println!("{}", resp.sum);
    });