up_rust/
owned_frame.rs

1/********************************************************************************
2 * Copyright (c) 2026 Contributors to the Eclipse Foundation
3 *
4 * See the NOTICE file(s) distributed with this work for additional
5 * information regarding copyright ownership.
6 *
7 * This program and the accompanying materials are made available under the
8 * terms of the Apache License Version 2.0 which is available at
9 * https://www.apache.org/licenses/LICENSE-2.0
10 *
11 * SPDX-License-Identifier: Apache-2.0
12 ********************************************************************************/
13
14use bytes::Bytes;
15
16use crate::{UFrameMetadata, UFrameMetadataError};
17
18/// Owned, serialization-neutral native uProtocol frame.
19///
20/// `UOwnedFrame` carries Phase 01 frame metadata plus optional owned payload
21/// bytes. It is available only with the `owned-frame-transport` feature and is
22/// additive to the ordinary `UTransport`/`UMessage` compatibility path.
23#[derive(Clone, Debug, PartialEq)]
24pub struct UOwnedFrame<S = Validated> {
25    metadata: UFrameMetadata,
26    payload: Option<Bytes>,
27    _state: core::marker::PhantomData<S>,
28}
29
30pub use crate::validation_state::{Unvalidated, Validated};
31
32impl UOwnedFrame<Validated> {
33    /// Creates a frame from metadata and optional payload bytes after validation.
34    ///
35    /// # Errors
36    ///
37    /// Returns an error if metadata validation fails or payload presence does
38    /// not match metadata payload encoding presence.
39    pub fn new(
40        metadata: UFrameMetadata,
41        payload: Option<Bytes>,
42    ) -> Result<Self, UFrameMetadataError> {
43        let frame = UOwnedFrame::<Validated> {
44            metadata,
45            payload,
46            _state: core::marker::PhantomData,
47        };
48        frame.check()?;
49        Ok(frame)
50    }
51
52    /// Creates a frame without validation.
53    #[must_use]
54    pub fn new_unchecked(
55        metadata: UFrameMetadata,
56        payload: Option<Bytes>,
57    ) -> UOwnedFrame<Unvalidated> {
58        UOwnedFrame {
59            metadata,
60            payload,
61            _state: core::marker::PhantomData,
62        }
63    }
64
65    /// Creates a payload-bearing frame after validation.
66    ///
67    /// # Errors
68    ///
69    /// Returns an error if metadata validation fails or the metadata has no
70    /// payload encoding.
71    pub fn with_payload(
72        metadata: UFrameMetadata,
73        payload: impl Into<Bytes>,
74    ) -> Result<Self, UFrameMetadataError> {
75        Self::new(metadata, Some(payload.into()))
76    }
77
78    /// Creates a payload-bearing frame without validation (test support:
79    /// negative tests must be able to construct invalid frames).
80    #[must_use]
81    #[cfg(test)]
82    pub(crate) fn with_payload_unchecked(
83        metadata: UFrameMetadata,
84        payload: impl Into<Bytes>,
85    ) -> UOwnedFrame<Unvalidated> {
86        Self::new_unchecked(metadata, Some(payload.into()))
87    }
88
89    /// Creates a no-payload frame after validation.
90    ///
91    /// # Errors
92    ///
93    /// Returns an error if metadata validation fails or the metadata still has
94    /// a payload encoding.
95    pub fn without_payload(metadata: UFrameMetadata) -> Result<Self, UFrameMetadataError> {
96        Self::new(metadata, None)
97    }
98}
99
100impl UOwnedFrame<Unvalidated> {
101    /// Validates the frame, transitioning it to the [`Validated`] state.
102    ///
103    /// This is the only way from [`Unvalidated`] to [`Validated`]; the
104    /// state transition IS the validation, and every transport API
105    /// accepts only the validated state.
106    ///
107    /// # Errors
108    ///
109    /// Returns an error if metadata is invalid, if payload bytes are
110    /// present without encoding metadata, or if encoding metadata is
111    /// present without payload bytes.
112    pub fn validate(self) -> Result<UOwnedFrame<Validated>, UFrameMetadataError> {
113        self.check()?;
114        Ok(UOwnedFrame {
115            metadata: self.metadata,
116            payload: self.payload,
117            _state: core::marker::PhantomData,
118        })
119    }
120}
121
122impl UOwnedFrame<Validated> {
123    /// Projects this frame into a `UMessage` — the compatibility bridge back
124    /// to the `UTransport` family.
125    ///
126    /// # Errors
127    ///
128    /// Returns an error if the frame metadata cannot be represented as
129    /// message attributes.
130    pub fn to_umessage(&self) -> Result<crate::UMessage, crate::UFrameMetadataError> {
131        crate::frame::metadata::try_project_frame_to_umessage(
132            self.metadata().clone(),
133            self.payload().cloned(),
134        )
135    }
136}
137
138impl<S> UOwnedFrame<S> {
139    /// Validates metadata and payload presence consistency.
140    ///
141    /// # Errors
142    ///
143    /// Returns an error if metadata is invalid, if payload bytes are present
144    /// without encoding metadata, or if encoding metadata is present without
145    /// payload bytes.
146    pub(crate) fn check(&self) -> Result<(), UFrameMetadataError> {
147        self.metadata.validate()?;
148        match (
149            self.payload.is_some(),
150            self.metadata.payload_encoding().is_some(),
151        ) {
152            (true, true) | (false, false) => Ok(()),
153            (true, false) => Err(UFrameMetadataError::PayloadWithoutEncoding),
154            (false, true) => Err(UFrameMetadataError::EncodingWithoutPayload),
155        }
156    }
157
158    /// Returns the frame metadata.
159    #[must_use]
160    pub fn metadata(&self) -> &UFrameMetadata {
161        &self.metadata
162    }
163
164    /// Returns the payload bytes when the frame carries a payload.
165    #[must_use]
166    pub fn payload(&self) -> Option<&Bytes> {
167        self.payload.as_ref()
168    }
169
170    /// Returns the payload bytes, or an empty slice when payload is absent.
171    ///
172    /// Use [`Self::payload`] when distinguishing absent payload from present
173    /// empty payload matters.
174    #[must_use]
175    pub fn payload_bytes(&self) -> &[u8] {
176        self.payload.as_deref().unwrap_or_default()
177    }
178
179    /// Returns whether the frame carries payload bytes, including present empty payloads.
180    #[must_use]
181    pub fn has_payload(&self) -> bool {
182        self.payload.is_some()
183    }
184
185    /// Consumes the frame and returns its metadata.
186    #[must_use]
187    pub fn into_metadata(self) -> UFrameMetadata {
188        self.metadata
189    }
190
191    /// Consumes the frame and returns its optional payload bytes.
192    #[must_use]
193    pub fn into_payload(self) -> Option<Bytes> {
194        self.payload
195    }
196
197    /// Consumes the frame and returns metadata plus optional payload bytes.
198    #[must_use]
199    pub fn into_parts(self) -> (UFrameMetadata, Option<Bytes>) {
200        (self.metadata, self.payload)
201    }
202}
203
204#[cfg(test)]
205mod tests {
206    use super::*;
207    use crate::{PayloadEncoding, UMessageBuilder, UUri};
208
209    fn topic() -> UUri {
210        UUri::try_from_parts("vehicle", 0x4210, 0x01, 0x9000).expect("failed to create topic")
211    }
212
213    fn metadata_without_encoding() -> UFrameMetadata {
214        let message = UMessageBuilder::publish(topic()).build().expect("message");
215        crate::frame::metadata::try_project_attributes_to_frame_metadata(message.attributes(), None)
216            .expect("metadata")
217    }
218
219    fn metadata_with_encoding() -> UFrameMetadata {
220        let message = UMessageBuilder::publish(topic()).build().expect("message");
221        crate::frame::metadata::try_project_attributes_to_frame_metadata(
222            message.attributes(),
223            Some(PayloadEncoding::RAW),
224        )
225        .expect("metadata")
226    }
227
228    #[test]
229    fn owned_frame_rejects_payload_without_encoding() {
230        let error =
231            UOwnedFrame::with_payload(metadata_without_encoding(), Bytes::from_static(b"payload"))
232                .unwrap_err();
233
234        assert_eq!(error, UFrameMetadataError::PayloadWithoutEncoding);
235    }
236
237    #[test]
238    fn owned_frame_rejects_encoding_without_payload() {
239        let error = UOwnedFrame::without_payload(metadata_with_encoding()).unwrap_err();
240
241        assert_eq!(error, UFrameMetadataError::EncodingWithoutPayload);
242    }
243
244    #[test]
245    fn owned_frame_accepts_absent_payload_without_encoding() {
246        let frame = UOwnedFrame::without_payload(metadata_without_encoding()).unwrap();
247
248        assert!(!frame.has_payload());
249        assert_eq!(frame.payload_bytes(), &[] as &[u8]);
250        assert!(frame.metadata().payload_encoding().is_none());
251    }
252
253    #[test]
254    fn owned_frame_accepts_present_empty_payload_with_standard_encoding() {
255        let frame = UOwnedFrame::with_payload(metadata_with_encoding(), Bytes::new()).unwrap();
256
257        assert!(frame.has_payload());
258        assert_eq!(frame.payload(), Some(&Bytes::new()));
259        assert_eq!(frame.payload_bytes(), &[] as &[u8]);
260        assert_eq!(
261            frame.metadata().payload_encoding(),
262            Some(&PayloadEncoding::RAW)
263        );
264    }
265}
266
267#[cfg(any(test, feature = "test-util"))]
268type OwnedListenerRegistration = (
269    crate::UUri,
270    Option<crate::UUri>,
271    std::sync::Arc<dyn crate::UOwnedListener>,
272);
273
274/// In-memory owned-frame transport for unit tests, examples, and guide
275/// doctests.
276///
277/// *Role: test/proof support — a complete [`UOwnedTransportImpl`](crate::UOwnedTransportImpl)
278/// loopback. Frames sent through it are dispatched to every registered
279/// [`UOwnedListener`](crate::UOwnedListener) whose filters match.*
280#[cfg(any(test, feature = "test-util"))]
281#[derive(Default)]
282pub struct InMemoryOwnedTransport {
283    listeners: std::sync::RwLock<Vec<OwnedListenerRegistration>>,
284}
285
286#[cfg(any(test, feature = "test-util"))]
287impl core::fmt::Debug for InMemoryOwnedTransport {
288    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
289        f.debug_struct("InMemoryOwnedTransport")
290            .finish_non_exhaustive()
291    }
292}
293
294#[cfg(any(test, feature = "test-util"))]
295#[async_trait::async_trait]
296impl crate::UOwnedTransportImpl for InMemoryOwnedTransport {
297    async fn send_validated_owned(&self, frame: crate::UOwnedFrame) -> Result<(), crate::UStatus> {
298        let matching: Vec<_> = {
299            let metadata = frame.metadata();
300            let source = metadata.source();
301            let sink = metadata.sink();
302            self.listeners
303                .read()
304                .expect("lock")
305                .iter()
306                .filter(|(source_filter, sink_filter, _)| {
307                    source_filter.matches(source)
308                        && match (sink_filter, sink) {
309                            (Some(pattern), Some(candidate)) => pattern.matches(candidate),
310                            (None, None) => true,
311                            _ => false,
312                        }
313                })
314                .map(|(_, _, listener)| listener.clone())
315                .collect()
316        };
317        for listener in matching {
318            listener.on_receive_owned(frame.clone()).await;
319        }
320        Ok(())
321    }
322
323    async fn register_validated_owned_listener(
324        &self,
325        source_filter: &crate::UUri,
326        sink_filter: Option<&crate::UUri>,
327        listener: std::sync::Arc<dyn crate::UOwnedListener>,
328    ) -> Result<(), crate::UStatus> {
329        self.listeners.write().expect("lock").push((
330            source_filter.to_owned(),
331            sink_filter.map(ToOwned::to_owned),
332            listener,
333        ));
334        Ok(())
335    }
336
337    async fn unregister_validated_owned_listener(
338        &self,
339        source_filter: &crate::UUri,
340        sink_filter: Option<&crate::UUri>,
341        listener: std::sync::Arc<dyn crate::UOwnedListener>,
342    ) -> Result<(), crate::UStatus> {
343        let target = std::sync::Arc::as_ptr(&listener) as *const () as usize;
344        let mut listeners = self.listeners.write().expect("lock");
345        let before = listeners.len();
346        listeners.retain(|(sf, kf, l)| {
347            !(sf == source_filter
348                && kf.as_ref() == sink_filter
349                && std::sync::Arc::as_ptr(l) as *const () as usize == target)
350        });
351        if listeners.len() < before {
352            Ok(())
353        } else {
354            Err(crate::UStatus::fail_with_code(
355                crate::UCode::NotFound,
356                "no such owned listener",
357            ))
358        }
359    }
360}
361
362#[cfg(test)]
363mod in_memory_owned_transport_tests {
364    use std::sync::Arc;
365
366    use super::*;
367    use crate::{UMessageBuilder, UOwnedListener, UOwnedTransport, UUri};
368
369    struct Capture(std::sync::Mutex<Vec<UOwnedFrame>>);
370
371    #[async_trait::async_trait]
372    impl UOwnedListener for Capture {
373        async fn on_receive_owned(&self, frame: UOwnedFrame) {
374            self.0.lock().expect("capture lock poisoned").push(frame);
375        }
376    }
377
378    fn frame(topic: &UUri) -> UOwnedFrame {
379        let message = UMessageBuilder::publish(topic.clone())
380            .build()
381            .expect("failed to build message");
382        let metadata = crate::frame::metadata::try_project_attributes_to_frame_metadata(
383            message.attributes(),
384            Some(crate::PayloadEncoding::RAW),
385        )
386        .expect("failed to project metadata");
387        UOwnedFrame::with_payload(metadata, b"x".to_vec()).expect("failed to build frame")
388    }
389
390    async fn registered(transport: &InMemoryOwnedTransport, topic: &UUri) -> Arc<Capture> {
391        let capture = Arc::new(Capture(std::sync::Mutex::new(Vec::new())));
392        transport
393            .register_owned_listener(topic, None, capture.clone())
394            .await
395            .expect("listener registration succeeds");
396        capture
397    }
398
399    #[tokio::test]
400    async fn delivers_frames_matching_the_registered_filter() {
401        let transport = InMemoryOwnedTransport::default();
402        let topic = UUri::try_from_parts("demo", 0x1_0001, 1, 0x8001).expect("valid topic URI");
403        let capture = registered(&transport, &topic).await;
404
405        transport
406            .send_owned(frame(&topic))
407            .await
408            .expect("matching send succeeds");
409
410        assert_eq!(capture.0.lock().expect("capture lock").len(), 1);
411    }
412
413    #[tokio::test]
414    async fn does_not_deliver_frames_for_other_topics() {
415        let transport = InMemoryOwnedTransport::default();
416        let topic = UUri::try_from_parts("demo", 0x1_0001, 1, 0x8001).expect("valid topic URI");
417        let other = UUri::try_from_parts("demo", 0x1_0001, 1, 0x8002).expect("valid other URI");
418        let capture = registered(&transport, &topic).await;
419
420        transport
421            .send_owned(frame(&other))
422            .await
423            .expect("non-matching send succeeds");
424
425        assert!(capture.0.lock().expect("capture lock").is_empty());
426    }
427
428    #[tokio::test]
429    async fn unregistering_an_unknown_listener_returns_not_found() {
430        let transport = InMemoryOwnedTransport::default();
431        let topic = UUri::try_from_parts("demo", 0x1_0001, 1, 0x8001).expect("valid topic URI");
432        let capture = registered(&transport, &topic).await;
433
434        transport
435            .unregister_owned_listener(&topic, None, capture.clone())
436            .await
437            .expect("first unregister succeeds");
438        let error = transport
439            .unregister_owned_listener(&topic, None, capture)
440            .await
441            .expect_err("second unregister must fail");
442
443        assert_eq!(error.code(), crate::UCode::NotFound);
444    }
445}
up_rust/owned_frame.rs

up_rust/
owned_frame.rs
