Struct Subscription 

pub struct Subscription<M, const RX_BUF: usize = { crate::config::DEFAULT_RX_BUF_SIZE }> { /* private fields */ }

Typed subscription handle with internal receive buffer.

ADOPT-BOUNDED against rclrs’s Subscription, whose callback is set on it AFTER construction (set_callback, set_async_callback, …). This handle is POLL-style and has no callback; a callback subscription takes its closure at DECLARATION (create_subscription(topic, callback)), because with no allocator the entity set is fixed when register runs. There is no setter to call later.

Two methods, both byte-oriented at the wire:

• take / recv — pull bytes from the backend, CDR-decode into M: RosMessage, hand back ownership of the typed message.
• take_serialized — copy bytes into the subscription’s internal buffer and return the length, leaving CDR decoding to the caller.

No typed borrow() exists. Borrow lives exclusively on RawSubscription. RecvView is &[u8] semantics; CDR decoding into a typed M requires owning the bytes (or running the decoder in place), which the borrow contract doesn’t fit. See docs/design/0010-zero-copy-raw-api.md decision D7.

Implementations

impl<M: RosMessage, const RX_BUF: usize> Subscription<M, RX_BUF>

pub fn take(&mut self) -> Result<Option<M>, NodeError>

Try to receive a typed message (non-blocking). Take one message if the middleware has one — phase-379 W3.

Named take after rcl (rcl_take) and rclcpp (Subscription::take), which spell the non-blocking receive that way — as does our own C surface. NOT after rclrs: its subscription API is callback/worker-driven and has no public polling counterpart at all (its take_* methods are private), which is why the ledger records this as an extension against the Rust reference rather than a match. take was Rust channel vocabulary that reads as a different contract to a ROS 2 user: both are non-blocking and both report emptiness without failing, so nothing asked for the other word. Renamed as a clean break, no shim.

pub fn take_serialized(&mut self) -> Result<Option<usize>, NodeError>

Try to receive raw CDR-encoded data (non-blocking).

pub fn buffer(&self) -> &[u8]

Get the receive buffer (valid after take_serialized).

pub fn supports_event(&self, kind: EventKind) -> bool

true if the active backend can fire the named event for this subscriber.

pub fn on_liveliness_changed<F>(&mut self, cb: F) -> Result<(), NodeError>

where F: FnMut(LivelinessChangedStatus) + Send + 'static,

Register a callback for LivelinessChanged. Fires when a tracked publisher’s liveliness state changes.

pub fn on_requested_deadline_missed<F>( &mut self, deadline: Duration, cb: F, ) -> Result<(), NodeError>

where F: FnMut(CountStatus) + Send + 'static,

Register a callback for RequestedDeadlineMissed. Fires when an expected sample doesn’t arrive within deadline.

pub fn on_message_lost<F>(&mut self, cb: F) -> Result<(), NodeError>

where F: FnMut(CountStatus) + Send + 'static,

Register a callback for MessageLost. Fires when the backend drops a sample (overflow, etc.).

pub fn has_data(&self) -> bool

Check if data is available without consuming it.

pub fn process_in_place( &mut self, f: impl FnOnce(&M), ) -> Result<bool, NodeError>

Process the received message in-place without copying.

pub async fn recv(&mut self) -> Result<M, NodeError>

Async: wait for the next message (no futures dependency needed).

Requires a background task running executor.spin_async() to drive I/O. Returns Ok(msg) on the next received message, or Err if the transport reports an error.

When the stream feature is enabled, prefer StreamExt::next() / TryStreamExt::try_next() for combinator support.

Example

let mut sub = node.create_subscription::<Int32>("/topic")?;
loop {
    let msg = sub.recv().await?;
    /* handle msg */
}

pub fn wait_next( &mut self, executor: &mut Executor<'_>, timeout: Duration, ) -> Result<Option<M>, NodeError>

Sync: wait for the next message, spinning the executor.

Returns Ok(Some(msg)) if a message arrives within timeout_ms, or Ok(None) on timeout. Unlike Promise::wait(), timeout is not an error — the caller typically retries in a loop.

Example

while let Some(msg) = sub.wait_next(&mut executor, core::time::Duration::from_millis(1000))? {
    /* handle msg */
}

pub fn qos(&self) -> QoSProfile

The QoS profile this subscription is ACTUALLY running.

rclrs’s SubscriptionState::qos, and the answer is a step better than its own: rclrs returns the profile the subscription was CREATED with, because it never reads one back. This returns what the backend GRANTED — rmw_subscription_get_actual_qos, in other words, which is the same question with the answer measured instead of assumed. The two differ whenever a backend serves something other than it was asked (zenoh clamps depth to its receive ring; DDS negotiates), and that difference is what answers “why is nothing arriving”.

A policy the backend could not report reads back as its Unknown variant, never as the request — upstream’s contract for this call. A backend with no read-back at all (XRCE) answers QoSProfile::QOS_PROFILE_UNKNOWN, every field an absence.

Free: read once at create and retained on the handle, so this never re-enters the transport.

pub fn topic_name(&self) -> &str

The topic this subscription was created on — rclrs’s SubscriptionState::topic_name. See EmbeddedPublisher::topic_name for the contract and for why the accessor is here rather than on the RMW trait. phase-444.

pub fn get_publisher_count( &self, executor: &mut Executor<'_>, ) -> Result<usize, NodeError>

How many publishers are on this subscription’s topic, RIGHT NOW — rclcpp’s Subscription::get_publisher_count, rcl’s rclc_subscription_get_publisher_count. phase-444.

The subscription half of EmbeddedPublisher::get_subscription_count, which states the order (the matched count — upstream’s rmw_subscription_count_matched_publishers — where the backend fills it, else the topic-wide graph count), why the executor is an argument, and why Unsupported is not 0.

Trait Implementations

impl<M, const RX_BUF: usize> Drop for Subscription<M, RX_BUF>

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.