Trait SubscriptionTrait 

pub trait SubscriptionTrait {
    type Error;

    // Required methods
    fn take_serialized(
        &mut self,
        buf: &mut [u8],
    ) -> Result<Option<usize>, Self::Error>;
    fn deserialization_error(&self) -> Self::Error;

    // Provided methods
    fn has_data(&self) -> bool { ... }
    fn refused_sample_len(&self) -> Option<usize> { ... }
    fn take_serialized_with_attachment(
        &mut self,
        buf: &mut [u8],
        _att_buf: &mut [u8],
    ) -> Result<Option<(usize, usize)>, Self::Error> { ... }
    fn take_sequence(
        &mut self,
        buf: &mut [u8],
        per_msg_cap: usize,
        max_msgs: usize,
        out_lens: &mut [usize],
    ) -> Result<usize, Self::Error> { ... }
    fn take<M>(&mut self, buf: &mut [u8]) -> Result<Option<M>, Self::Error>
       where M: RosMessage { ... }
    fn process_raw_in_place(
        &mut self,
        f: impl FnOnce(&[u8]),
    ) -> Result<bool, Self::Error>
       where Self::Error: From<TransportError> { ... }
    fn supports_process_in_place(&self) -> bool { ... }
    fn process_raw_in_place_with_info(
        &mut self,
        f: impl FnOnce(&[u8], Option<MessageInfo>),
    ) -> Result<bool, Self::Error>
       where Self::Error: From<TransportError> { ... }
    fn take_serialized_with_info(
        &mut self,
        buf: &mut [u8],
    ) -> Result<Option<(usize, Option<MessageInfo>)>, Self::Error> { ... }
    fn register_waker(&self, _waker: &Waker) { ... }
    fn supports_event(&self, _kind: EventKind) -> bool { ... }
    unsafe fn register_event_callback(
        &mut self,
        _kind: EventKind,
        _deadline_ms: u32,
        _cb: unsafe extern "C" fn(EventKind, *const c_void, *mut c_void),
        _user_ctx: *mut c_void,
    ) -> Result<(), Self::Error> { ... }
    fn unsupported_event_error(&self) -> Self::Error { ... }
    fn actual_qos(&self) -> QoSProfile { ... }
    fn count_matched_publishers(&self) -> Result<usize, Self::Error>
       where Self::Error: From<TransportError> { ... }
}

Subscription trait for receiving messages.

Threading

&mut self on take_serialized — the executor takes exclusive ownership of the subscriber for the duration of a receive. A backend that wants to allow concurrent receives must split into per-thread sub-handles internally.

Buffer ownership

buf is caller-owned. The implementation copies the next ready message into buf and returns the byte count. The caller may re-use or drop buf immediately after the call.

Blocking

take_serialized is non-blocking: returns Ok(None) (or equivalent for backends that map empty into a zero-length read) when no message is ready. Use Session::drive_io to wait for data; never sleep inside take_serialized.

Required Associated Types

type Error

Error type for receive operations

Required Methods

fn take_serialized( &mut self, buf: &mut [u8], ) -> Result<Option<usize>, Self::Error>

Try to receive one message into buf.

Non-blocking. On success returns Ok(Some(len)) where len is the byte count written into buf[..len]. Returns Ok(None) if no message is ready. If buf is too small the backend may either truncate (and document it) or return an error (preferred).

fn deserialization_error(&self) -> Self::Error

Return a deserialization error (implementation specific)

Provided Methods

fn has_data(&self) -> bool

Check if data is available without consuming it.

Non-destructive — does not advance the receive cursor. Conservative default returns true (always assume data may be available); backends should override with a real check to avoid spurious receive attempts.

fn refused_sample_len(&self) -> Option<usize>

Issue 1612 — how many bytes the sample the most recent take REFUSED as too big for the caller’s buffer needed.

Read it right after a take returns BufferTooSmall; a later take may overwrite it. None means the backend does not know the size (the default, and the honest answer for one that never measured it), never “zero bytes” — a zero-byte sample fits every buffer, so no refusal can need zero.

A query beside the error rather than a payload on TransportError::BufferTooSmall: that variant is the same word a dozen unrelated paths return (arena exhaustion, a short reply buffer), so giving it a field would make every one of them invent a size. Only a take ever has one, and only a take is asked.

fn take_serialized_with_attachment( &mut self, buf: &mut [u8], _att_buf: &mut [u8], ) -> Result<Option<(usize, usize)>, Self::Error>

Phase 128.F.4 — receive with attachment bytes alongside the payload.

On success returns Ok(Some((payload_len, attachment_len))) with the payload written into buf[..payload_len] and the attachment (if any) written into att_buf[..attachment_len]. attachment_len == 0 means the incoming sample carried no attachment.

Default body falls back to take_serialized and reports a 0-length attachment. Backends with native attachment support override to populate att_buf. Cross-RMW bridges use the attachment to read the bridge_origin tag stamped by the sending side.

fn take_sequence( &mut self, buf: &mut [u8], per_msg_cap: usize, max_msgs: usize, out_lens: &mut [usize], ) -> Result<usize, Self::Error>

Phase 124.D.1 — burst-take.

Drain up to max_msgs queued samples into the contiguous buf block in one call, with the i-th sample at buf[i * per_msg_cap .. i * per_msg_cap + out_lens[i]]. Returns the number of messages actually delivered. Partial drains MUST report the count, not error out.

Default body loop-drives take_serialized so callers can commit to the batched API regardless of backend support. Concrete backends opt in by overriding with a native batch take (zenoh queue drain, dds_take(max_samples)).

fn take<M>(&mut self, buf: &mut [u8]) -> Result<Option<M>, Self::Error>

where M: RosMessage,

Try to receive a typed message (non-blocking)

fn process_raw_in_place( &mut self, f: impl FnOnce(&[u8]), ) -> Result<bool, Self::Error>

where Self::Error: From<TransportError>,

Process the received message in-place without copying.

Calls f with a reference to the raw CDR bytes in the subscriber’s internal receive buffer, avoiding a copy into a caller-provided buffer. While f executes the buffer is exclusively borrowed — any messages arriving from the transport during that time are dropped to prevent data races.

Returns Ok(true) if a message was available and f was called, Ok(false) if no message was available.

Default body: returns Err(MessageTooLarge) — the old default silently truncated anything larger than 1 KB into a stack buffer, which broke large messages with no diagnostic. Backends must override this with a real zero-copy path if they advertise support for process_raw_in_place; callers that hit the default should use take_serialized with a caller-sized buffer instead.

fn supports_process_in_place(&self) -> bool

Whether this backend implements the in-place dispatch methods (process_raw_in_place / process_raw_in_place_with_info) with a real zero-copy borrow.

The executor consults this at subscription registration to choose the in-place arena dispatch (borrow + deserialize from the backend slot, no arena buffer) over the buffered dispatch (copy into an arena buffer first). Backends that leave the in-place methods at their unsupported default return false (the default) and keep the buffered path. (RFC-0038, Phase 231 Wave 0.2.)

fn process_raw_in_place_with_info( &mut self, f: impl FnOnce(&[u8], Option<MessageInfo>), ) -> Result<bool, Self::Error>

where Self::Error: From<TransportError>,

In-place processing variant that also surfaces publisher metadata.

Same borrow contract as process_raw_in_place: f receives the raw CDR bytes plus the parsed MessageInfo — the co-located attachment (publisher GID / sequence / source timestamp), or None when no attachment was present — for the duration of the call; the slot is released after f returns. Ok(true) = a message was available and f was called; Ok(false) = none ready.

Default body: returns the unsupported error (mirrors process_raw_in_place). Backends that advertise in-place support override this with a real zero-copy path; callers that hit the default should use the buffered take_serialized_with_info path instead. (RFC-0038, Phase 231 Wave 0.1.)

fn take_serialized_with_info( &mut self, buf: &mut [u8], ) -> Result<Option<(usize, Option<MessageInfo>)>, Self::Error>

Try to receive raw data along with publisher metadata.

When available, MessageInfo contains the publisher’s GID (Global Identifier) and source timestamp, extracted from a transport-level attachment on the incoming message.

Returns Ok(Some((len, info))) if data is available, where:

• len is the number of bytes written to the buffer
• info is the parsed publisher metadata (if attachment was present)

Default: delegates to take_serialized with no info.

fn register_waker(&self, _waker: &Waker)

Register an async waker to be notified when data arrives.

Called from Future::poll() implementations to store the waker. The transport backend calls waker.wake() from its receive callback when new data is available, enabling event-driven async without busy-polling.

Default: no-op (backends that don’t support waking simply ignore this).

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

Phase 108 — true if the backend can generate this event for this subscriber. Default returns false; backends override per supported event kind.

Subscription-side event kinds: EventKind::LivelinessChanged, EventKind::RequestedDeadlineMissed, EventKind::MessageLost. Publisher kinds always return false here.

unsafe fn register_event_callback( &mut self, _kind: EventKind, _deadline_ms: u32, _cb: unsafe extern "C" fn(EventKind, *const c_void, *mut c_void), _user_ctx: *mut c_void, ) -> Result<(), Self::Error>

Phase 108 — register a callback fired when the named status event occurs. deadline_ms applies to EventKind::RequestedDeadlineMissed only; ignored otherwise. Default impl returns the backend’s “unsupported”-shaped error.

Safety

cb and user_ctx must remain valid for the entity’s lifetime. Caller (typically nros-node’s typed wrapper) is responsible for keeping the closure / context arena alive.

fn unsupported_event_error(&self) -> Self::Error

Phase 108 — backend’s error variant for “this event kind is not supported.” Default reuses deserialization_error() for backends that don’t have a distinct Unsupported mapping.

fn actual_qos(&self) -> QoSProfile

The profile this entity is ACTUALLY running — what the backend GRANTED, not what the call site requested.

Upstream’s rmw_subscription_get_actual_qos. The two differ whenever a backend serves something other than what it was asked for, and the difference is exactly what answers “why is nothing arriving”: a RELIABLE reader does not match a BEST_EFFORT writer, and a clamped history depth drops samples the caller believed were kept.

Per-field, and an unreportable field is an ABSENCE. A backend that determines four policies and not the fifth returns the four and leaves the fifth at its Unknown variant. Never echo the request into a field you did not read back — that is the inverse of this method’s meaning, and it is what the six C-ABI slots did until this phase.

The default is QoSProfile::QOS_PROFILE_UNKNOWN — every field an absence — which is the honest answer for a backend with no read-back at all (XRCE has none; zenoh-pico’s QoS is per-message flags). It is NOT the request, and it is NOT QOS_PROFILE_DEFAULT.

fn count_matched_publishers(&self) -> Result<usize, Self::Error>

where Self::Error: From<TransportError>,

How many publishers are MATCHED to this subscription — upstream rmw_subscription_count_matched_publishers, the vtable’s subscription_count_matched_publishers. Issue 1643.

The subscription half of Publisher::count_matched_subscriptions, which states the contract: matched, not merely discovered; Err(Unsupported) by default, never a fabricated 0.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementations on Foreign Types

impl Subscription for CffiSubscription

fn actual_qos(&self) -> QoSProfile

The profile the backend GRANTED. See CffiPublisher::actual_qos.

fn count_matched_publishers(&self) -> Result<usize, TransportError>

Upstream rmw_subscription_count_matched_publishers, through the vtable slot. See CffiPublisher::count_matched_subscriptions (Publisher::count_matched_subscriptions).

type Error = TransportError

fn supports_process_in_place(&self) -> bool

fn process_raw_in_place( &mut self, f: impl FnOnce(&[u8]), ) -> Result<bool, <CffiSubscription as Subscription>::Error>

fn has_data(&self) -> bool

fn take_serialized( &mut self, buf: &mut [u8], ) -> Result<Option<usize>, TransportError>

fn refused_sample_len(&self) -> Option<usize>

fn take_serialized_with_info( &mut self, buf: &mut [u8], ) -> Result<Option<(usize, Option<MessageInfo>)>, TransportError>

fn take_sequence( &mut self, buf: &mut [u8], per_msg_cap: usize, max_msgs: usize, out_lens: &mut [usize], ) -> Result<usize, TransportError>

fn deserialization_error(&self) -> TransportError

fn unsupported_event_error(&self) -> TransportError

unsafe fn register_event_callback( &mut self, kind: EventKind, deadline_ms: u32, cb: unsafe extern "C" fn(EventKind, *const c_void, *mut c_void), user_ctx: *mut c_void, ) -> Result<(), TransportError>

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

Implementors