Subscribers can't hear `ros2_client` publishers

[onkoe]

Hi again! I was making tests for our nodes and couldn't get ros2_client subscribers to hear ros2_client publishers. When listening on the topic using ros2 topic echo /topic_name, I was able to see output.

As an example, I made a test for ros2-client in a newly created tests/topic.rs:

//! Ensures that two `ros2_client` nodes can speak to each other.

use std::{pin::pin, time::Duration};
use futures::{future::select, StreamExt};

use ros2_client::{
  Context, MessageTypeName, Name, NodeName, NodeOptions, Publisher, Subscription,
  DEFAULT_PUBLISHER_QOS,
};

#[tokio::test]
async fn topic_communication() {
  select(pin!(make_publisher()), pin!(make_subscriber())).await;
}

async fn make_subscriber() {
  // make context
  let ctx = Context::new().unwrap();

  // node, topic, subscriber
  let mut node = ctx
    .new_node(
      NodeName::new("/", "topic_test_sub_node").unwrap(),
      NodeOptions::new(),
    )
    .unwrap();
  let topic = node
    .create_topic(
      &Name::new("/", "pub_sub_topic").unwrap(),
      MessageTypeName::new("std_msgs", "String"),
      &DEFAULT_PUBLISHER_QOS.clone(),
    )
    .unwrap();
  let subscriber: Subscription<String> = node.create_subscription(&topic, None).unwrap();

  // spin node in background
  tokio::task::spawn(node.spinner().unwrap().spin());

  // start listening.
  //
  // if we don't get a message within five seconds, fail the test.
  let (msg, _msg_info) = tokio::time::timeout(
    Duration::from_secs(5),
    Box::pin(subscriber.async_stream()).next(),
  )
  .await
  .expect("Test timed out - publisher never sent anything!")
  .expect("we should've got a message.")
  .expect("message should've sent correctly!");

  assert_eq!(msg, "hello subscriber!");
}

async fn make_publisher() {
  let ctx = Context::new().unwrap();

  let mut node = ctx
    .new_node(
      NodeName::new("/", "topic_test_pub_node").unwrap(),
      NodeOptions::new(),
    )
    .unwrap();

  let topic = node
    .create_topic(
      &Name::new("/", "pub_sub_topic").unwrap(),
      MessageTypeName::new("std_msgs", "String"),
      &DEFAULT_PUBLISHER_QOS.clone(),
    )
    .unwrap();

  let publisher: Publisher<String> = node.create_publisher(&topic, None).unwrap();

  tokio::task::spawn(node.spinner().unwrap().spin());

  // send messages every 0.25 seconds
  loop {
    publisher
      .async_publish("hello subscriber!".into())
      .await
      .unwrap();

    tokio::time::sleep(Duration::from_millis(250)).await;
  }
}

Output from the test:

test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

     Running tests/topic.rs (target/debug/deps/topic-f3f388d49a16d225)

running 1 test
test topic_communication ... FAILED

failures:

---- topic_communication stdout ----
thread 'topic_communication' panicked at tests/topic.rs:47:4:
Test timed out - publisher never sent anything!: Elapsed(())
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace


failures:
    topic_communication

test result: FAILED. 0 passed; 1 failed; 0 ignored; 0 measured; 0 filtered out; finished in 5.06s

error: test failed, to rerun pass `--test topic`

Output from ros2 topic echo /pub_sub_topic:

$ ros2 topic echo /pub_sub_topic
data: hello subscriber!
---
data: hello subscriber!
---
data: hello subscriber!
---
data: hello subscriber!
---
data: hello subscriber!
---
data: hello subscriber!
---
data: hello subscriber!
---

Please note that this test uses tokio (as mentioned in #26), though the same problem seems to appear when doing things synchronously, too.

In addition, you can force the test to pass by using ros2 topic pub:

$ ros2 topic pub /pub_sub_topic std_msgs/String {data: "hello subscriber!"}
publisher: beginning loop
publishing #1: std_msgs.msg.String(data='hello subscriber!')

publishing #2: std_msgs.msg.String(data='hello subscriber!')

publishing #3: std_msgs.msg.String(data='hello subscriber!')

publishing #4: std_msgs.msg.String(data='hello subscriber!')

Running tests/topic.rs (target/debug/deps/topic-f3f388d49a16d225)

running 1 test
test topic_communication ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 2.58s

[jhelovuo]

Thank you for reporting and writing an example code.

Is it ok for you if we include this test case in ros2-client sources with some modification?
(I renamed this test case to pub_and_sub , added attribution to you. You grant Apache-2.0 license to the test case.)

However, your test code seems very good: It actually works on my computer.

$ cargo test -- pub_and_sub
...
running 1 test
test pub_and_sub ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 4.53s

Using ros2-client 0.8.0.

I cannot advise you much further on this, since I cannot reproduce the problem. Some ideas:

• Listen also to the Node's .status_receiver() and dump the output. Both Publisher and Subscriber should report matching the other. DDS does not talk to the other party unless Discovery does its job and there is a match in topic name, type name, and QoS policies.
• Can you run the case with logging enabled at e.g. warn or info levels? Can you see any indication of errors there?
• The next step to diagnose would be to trace RTPS traffic with Wireshark.

[onkoe]

Interesting! I noticed that this problem isn't present on our robot's Jetson Orin Nano, but it affects both macOS and Fedora, on different computers. I'll try it out in a fresh environment to find some kind of conf. issue :D

I'm happy to PR the test. Did you want me to rename the file to pub_and_sub.rs?

[jhelovuo]

Ok, then I'll consider this issue done. If you find out that ros2-client is still misbehaving, then you are welcome to reopen the issue.

I will rename and merge your test case to master.

Some advice on testing:

• First thing to check is firewalls, but this is unlikely, since other DDSs work.
• ros2-client (RustDDS) currently requires a real IP-multicast-capable network interface to communicate, even within the same computer. Local loopback is not sufficient, because it should not support multicast.
• Most often "no communication" issues are caused by DDS/RTPS Discovery process failing somehow.

[onkoe]

@jhelovuo We experienced this in our CI: https://github.com/Sooner-Rover-Team/auto_ros2/actions/runs/13572151553/job/37939639153?pr=42

It fixed itself in the next push: https://github.com/Sooner-Rover-Team/auto_ros2/actions/runs/13577154841/job/37955885293?pr=42

I think something is flaky in ros2-client, though I can't yet confirm that. I'll continue watching for any causes as we polish up our codebase. :)

[jhelovuo]

Good observation, thank you for notifying us. Unfortunately, I cannot imemdiately tell what is wrong.

From your test code, it seems that you are creating a ROS2 Service, which of course consists of a server and a client. That seems just fine. Also your QoS settings seem to be ok. At the DDS/RTPS level, this breaks down into two Topics (request and response), both of which have a a Writer and Reader, so a total of four network endpoints. All of these have to match with their counterparts for the exchange to work.

If you find means to reproduce this, please open an issue so that we can investigate and fix.

[onkoe]

PR'd to RustDDS in hopes of fixing this. Not enough time to test right now, but I promise to soon! :)