up_rust/payload/
codec.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 std::io::Read;
15
16use bytes::Bytes;
17
18use crate::PayloadEncoding;
19#[cfg(feature = "protobuf-support")]
20use crate::ProtobufMappable;
21
22use super::UWireError;
23
24/// Byte layout requested by a payload codec.
25#[derive(Clone, Copy, Debug, Eq, PartialEq)]
26pub struct PayloadLayout {
27    len: usize,
28    align: usize,
29}
30
31impl PayloadLayout {
32    /// Creates a payload layout.
33    ///
34    /// # Errors
35    ///
36    /// Returns an error when `align` is zero or not a power of two.
37    pub fn new(len: usize, align: usize) -> Result<Self, UWireError> {
38        if align == 0 {
39            return Err(UWireError::invalid_payload(
40                "payload alignment must be non-zero",
41            ));
42        }
43        if !align.is_power_of_two() {
44            return Err(UWireError::invalid_payload(format!(
45                "payload alignment {align} is not a power of two"
46            )));
47        }
48        Ok(Self { len, align })
49    }
50
51    /// Returns the payload length in bytes.
52    #[must_use]
53    pub fn len(self) -> usize {
54        self.len
55    }
56
57    /// Returns the required payload alignment in bytes.
58    #[must_use]
59    pub fn align(self) -> usize {
60        self.align
61    }
62
63    /// Returns whether the payload has zero bytes.
64    #[must_use]
65    pub fn is_empty(self) -> bool {
66        self.len == 0
67    }
68}
69
70/// Compile-time identity for an application payload codec.
71///
72/// ```rust
73/// use up_rust::{payload::codec::PayloadFormat, PayloadEncoding, UPayloadFormat};
74///
75/// struct JsonTelemetry;
76///
77/// impl PayloadFormat for JsonTelemetry {
78///     fn name() -> &'static str {
79///         "json-telemetry-v1"
80///     }
81///
82///     fn encoding() -> PayloadEncoding {
83///         PayloadEncoding::JSON
84///     }
85/// }
86/// ```
87pub trait PayloadFormat {
88    /// Stable codec name for logs, diagnostics, and configuration.
89    fn name() -> &'static str;
90
91    /// Payload encoding metadata written into frames that use this codec.
92    fn encoding() -> PayloadEncoding;
93}
94
95/// Payload-layer codec identity used by typed frame helpers.
96pub trait PayloadCodec {
97    /// Stable codec name for logs, diagnostics, and configuration.
98    fn codec_name() -> &'static str;
99
100    /// Payload encoding metadata written into frames that use this codec.
101    fn payload_encoding() -> PayloadEncoding;
102
103    /// Verifies frame encoding metadata against this codec.
104    ///
105    /// # Errors
106    ///
107    /// Returns an error if the frame is missing payload encoding metadata or if
108    /// the metadata is incompatible with this codec.
109    fn verify_encoding(actual: Option<&PayloadEncoding>) -> Result<(), UWireError> {
110        let expected = Self::payload_encoding();
111        let actual = actual.ok_or(UWireError::MissingEncoding)?;
112        if !actual.is_compatible_with(&expected) {
113            return Err(UWireError::UnsupportedEncoding {
114                expected: Box::new(expected),
115                actual: Box::new(actual.clone()),
116            });
117        }
118        Ok(())
119    }
120}
121
122impl<F> PayloadCodec for F
123where
124    F: PayloadFormat,
125{
126    fn codec_name() -> &'static str {
127        <F as PayloadFormat>::name()
128    }
129
130    fn payload_encoding() -> PayloadEncoding {
131        <F as PayloadFormat>::encoding()
132    }
133}
134
135/// Encodes a typed value with a [`PayloadCodec`].
136#[diagnostic::on_unimplemented(
137    message = "the payload codec `{Self}` cannot encode `{T}`",
138    label = "no `EncodePayload<{T}>` implementation",
139    note = "wire implementers provide `payload_layout` and `encode_payload`; see the wire-implementer walkthrough"
140)]
141pub trait EncodePayload<T: ?Sized>: PayloadCodec {
142    /// Returns the exact payload layout required to encode `value`.
143    ///
144    /// This is a measurement operation, not necessarily a constant-time size
145    /// lookup. A variable-length codec may serialize or traverse `value` here
146    /// and perform the work again in [`Self::encode_payload`]. Benchmarks and
147    /// callers that care about this cost should label the probe and write phases
148    /// separately. The returned length and alignment are the complete contract
149    /// for the destination passed to `encode_payload`.
150    ///
151    /// # Errors
152    ///
153    /// Returns an error if the value cannot be measured for this codec.
154    fn payload_layout(value: &T) -> Result<PayloadLayout, UWireError>;
155
156    /// Encodes `value` into `dst`.
157    ///
158    /// # Errors
159    ///
160    /// Returns an error if `dst` is too small or serialization fails.
161    fn encode_payload(value: &T, dst: &mut [u8]) -> Result<(), UWireError>;
162
163    /// Encodes `value` into owned bytes.
164    ///
165    /// # Errors
166    ///
167    /// Returns an error if serialization fails.
168    fn encode_payload_owned(value: &T) -> Result<Bytes, UWireError> {
169        let layout = Self::payload_layout(value)?;
170        let mut bytes = vec![0_u8; layout.len()];
171        Self::encode_payload(value, &mut bytes)?;
172        Ok(Bytes::from(bytes))
173    }
174}
175
176/// Decodes a typed value from contiguous payload bytes.
177pub trait DecodePayload<'a, T>: PayloadCodec {
178    /// Decodes `T` from payload bytes.
179    ///
180    /// # Errors
181    ///
182    /// Returns an error if `src` is malformed for this codec.
183    fn decode_payload(src: &'a [u8]) -> Result<T, UWireError>;
184}
185
186/// Decodes a typed value from an ordered payload byte stream.
187pub trait ReadDecodePayload<T>: PayloadCodec {
188    /// Decodes `T` from `reader`, which must yield exactly `payload_len` bytes.
189    ///
190    /// Implementations must treat `payload_len` as a finite allocation and read
191    /// bound before reserving payload-sized storage. They must consume exactly
192    /// that many bytes, distinguish an early EOF from malformed payload data,
193    /// and check for an additional byte so overlong input is rejected. Reader
194    /// errors and malformed input must return [`UWireError`], not panic. A codec
195    /// may enforce a lower implementation-specific maximum, but must report that
196    /// limit rather than allocating first and rejecting later.
197    ///
198    /// # Errors
199    ///
200    /// Returns an error if the reader fails, yields an unexpected byte count, or
201    /// contains malformed payload bytes for this codec.
202    fn decode_payload_from_reader<R: Read>(reader: R, payload_len: usize) -> Result<T, UWireError>;
203}
204
205/// Marker trait for byte-oriented payload codecs.
206pub trait BytePayloadCodec: PayloadCodec {}
207
208/// Built-in raw byte payload codec.
209#[derive(Debug)]
210pub struct RawBytes;
211
212impl RawBytes {
213    /// Returns the raw-byte payload encoding metadata.
214    #[must_use]
215    pub fn encoding() -> PayloadEncoding {
216        <Self as PayloadFormat>::encoding()
217    }
218}
219
220impl PayloadFormat for RawBytes {
221    fn name() -> &'static str {
222        "raw-bytes"
223    }
224
225    fn encoding() -> PayloadEncoding {
226        PayloadEncoding::RAW
227    }
228}
229
230impl BytePayloadCodec for RawBytes {}
231
232impl EncodePayload<[u8]> for RawBytes {
233    fn payload_layout(value: &[u8]) -> Result<PayloadLayout, UWireError> {
234        PayloadLayout::new(value.len(), 1)
235    }
236
237    fn encode_payload(value: &[u8], dst: &mut [u8]) -> Result<(), UWireError> {
238        let actual = dst.len();
239        let out = dst
240            .get_mut(..value.len())
241            .ok_or_else(|| UWireError::buffer_too_small(value.len(), actual))?;
242        out.copy_from_slice(value);
243        Ok(())
244    }
245}
246
247impl<'a> DecodePayload<'a, &'a [u8]> for RawBytes {
248    fn decode_payload(src: &'a [u8]) -> Result<&'a [u8], UWireError> {
249        Ok(src)
250    }
251}
252
253impl DecodePayload<'_, Vec<u8>> for RawBytes {
254    fn decode_payload(src: &[u8]) -> Result<Vec<u8>, UWireError> {
255        Ok(src.to_vec())
256    }
257}
258
259impl DecodePayload<'_, Bytes> for RawBytes {
260    fn decode_payload(src: &[u8]) -> Result<Bytes, UWireError> {
261        Ok(Bytes::copy_from_slice(src))
262    }
263}
264
265impl ReadDecodePayload<Vec<u8>> for RawBytes {
266    fn decode_payload_from_reader<R: Read>(
267        reader: R,
268        payload_len: usize,
269    ) -> Result<Vec<u8>, UWireError> {
270        read_exact_payload(reader, payload_len)
271    }
272}
273
274impl ReadDecodePayload<Bytes> for RawBytes {
275    fn decode_payload_from_reader<R: Read>(
276        reader: R,
277        payload_len: usize,
278    ) -> Result<Bytes, UWireError> {
279        read_exact_payload(reader, payload_len).map(Bytes::from)
280    }
281}
282
283/// Protocol Buffers application payload codec.
284///
285/// `ProtobufPayload` serializes and deserializes only application payload bytes.
286/// It does not wrap a complete uProtocol frame and does not serialize frame
287/// metadata.
288#[derive(Debug)]
289pub struct ProtobufPayload;
290
291impl ProtobufPayload {
292    /// Returns the generic Protocol Buffers payload encoding metadata.
293    #[must_use]
294    pub fn encoding() -> PayloadEncoding {
295        <Self as PayloadFormat>::encoding()
296    }
297}
298
299impl PayloadFormat for ProtobufPayload {
300    fn name() -> &'static str {
301        "protobuf"
302    }
303
304    fn encoding() -> PayloadEncoding {
305        PayloadEncoding::PROTOBUF
306    }
307}
308
309#[cfg(feature = "protobuf-support")]
310impl<T> EncodePayload<T> for ProtobufPayload
311where
312    T: ProtobufMappable,
313{
314    fn payload_layout(value: &T) -> Result<PayloadLayout, UWireError> {
315        PayloadLayout::new(Self::encode_payload_owned(value)?.len(), 1)
316    }
317
318    fn encode_payload(value: &T, dst: &mut [u8]) -> Result<(), UWireError> {
319        copy_encoded_payload(Self::encode_payload_owned(value)?, dst)
320    }
321
322    fn encode_payload_owned(value: &T) -> Result<Bytes, UWireError> {
323        value
324            .write_to_protobuf_bytes()
325            .map(Bytes::from)
326            .map_err(|error| UWireError::serialization_error(error.to_string()))
327    }
328}
329
330#[cfg(feature = "protobuf-support")]
331impl<'a, T> DecodePayload<'a, T> for ProtobufPayload
332where
333    T: ProtobufMappable,
334{
335    fn decode_payload(src: &'a [u8]) -> Result<T, UWireError> {
336        T::parse_from_protobuf_bytes(src)
337            .map_err(|error| UWireError::invalid_payload(error.to_string()))
338    }
339}
340
341#[cfg(feature = "protobuf-support")]
342impl<T> ReadDecodePayload<T> for ProtobufPayload
343where
344    T: ProtobufMappable,
345{
346    fn decode_payload_from_reader<R: Read>(reader: R, payload_len: usize) -> Result<T, UWireError> {
347        let bytes = read_exact_payload(reader, payload_len)?;
348        T::parse_from_protobuf_bytes(&bytes)
349            .map_err(|error| UWireError::invalid_payload(error.to_string()))
350    }
351}
352
353/// Protocol Buffers `google.protobuf.Any` application payload codec.
354#[derive(Debug)]
355pub struct ProtobufAnyPayload;
356
357impl ProtobufAnyPayload {
358    /// Returns the protobuf-Any payload encoding metadata.
359    #[must_use]
360    pub fn encoding() -> PayloadEncoding {
361        <Self as PayloadFormat>::encoding()
362    }
363}
364
365impl PayloadFormat for ProtobufAnyPayload {
366    fn name() -> &'static str {
367        "protobuf-any"
368    }
369
370    fn encoding() -> PayloadEncoding {
371        PayloadEncoding::PROTOBUF_WRAPPED_IN_ANY
372    }
373}
374
375#[cfg(feature = "protobuf-support")]
376impl<T> EncodePayload<T> for ProtobufAnyPayload
377where
378    T: ProtobufMappable,
379{
380    fn payload_layout(value: &T) -> Result<PayloadLayout, UWireError> {
381        PayloadLayout::new(Self::encode_payload_owned(value)?.len(), 1)
382    }
383
384    fn encode_payload(value: &T, dst: &mut [u8]) -> Result<(), UWireError> {
385        copy_encoded_payload(Self::encode_payload_owned(value)?, dst)
386    }
387
388    fn encode_payload_owned(value: &T) -> Result<Bytes, UWireError> {
389        value
390            .write_to_packed_protobuf_bytes()
391            .map(Bytes::from)
392            .map_err(|error| UWireError::serialization_error(error.to_string()))
393    }
394}
395
396#[cfg(feature = "protobuf-support")]
397impl<'a, T> DecodePayload<'a, T> for ProtobufAnyPayload
398where
399    T: ProtobufMappable,
400{
401    fn decode_payload(src: &'a [u8]) -> Result<T, UWireError> {
402        T::parse_from_packed_protobuf_bytes(src)
403            .map_err(|error| UWireError::invalid_payload(error.to_string()))
404    }
405}
406
407#[cfg(feature = "protobuf-support")]
408impl<T> ReadDecodePayload<T> for ProtobufAnyPayload
409where
410    T: ProtobufMappable,
411{
412    fn decode_payload_from_reader<R: Read>(reader: R, payload_len: usize) -> Result<T, UWireError> {
413        let bytes = read_exact_payload(reader, payload_len)?;
414        T::parse_from_packed_protobuf_bytes(&bytes)
415            .map_err(|error| UWireError::invalid_payload(error.to_string()))
416    }
417}
418
419#[cfg(feature = "protobuf-support")]
420fn copy_encoded_payload(bytes: Bytes, dst: &mut [u8]) -> Result<(), UWireError> {
421    let actual = dst.len();
422    let out = dst
423        .get_mut(..bytes.len())
424        .ok_or_else(|| UWireError::buffer_too_small(bytes.len(), actual))?;
425    out.copy_from_slice(&bytes);
426    Ok(())
427}
428
429fn read_exact_payload<R: Read>(mut reader: R, payload_len: usize) -> Result<Vec<u8>, UWireError> {
430    let mut bytes = Vec::with_capacity(payload_len);
431    reader
432        .read_to_end(&mut bytes)
433        .map_err(|error| UWireError::invalid_payload(error.to_string()))?;
434    if bytes.len() != payload_len {
435        return Err(UWireError::invalid_payload(format!(
436            "payload reader yielded {} bytes but payload_len returned {payload_len} bytes",
437            bytes.len()
438        )));
439    }
440    Ok(bytes)
441}
442
443#[cfg(test)]
444mod tests {
445    #[cfg(feature = "zero-copy-transport")]
446    use std::io::Chain;
447    #[cfg(any(feature = "protobuf-support", feature = "zero-copy-transport"))]
448    use std::io::Cursor;
449
450    #[cfg(feature = "protobuf-support")]
451    use protobuf::well_known_types::wrappers::StringValue;
452
453    #[cfg(feature = "zero-copy-transport")]
454    use crate::{UFrameMetadata, UFrameView, UMessageBuilder, UUri};
455
456    use super::*;
457
458    #[cfg(feature = "zero-copy-transport")]
459    fn topic() -> UUri {
460        UUri::try_from_parts("vehicle", 0x4210, 0x01, 0x9000).expect("topic")
461    }
462
463    #[cfg(feature = "protobuf-support")]
464    fn message(value: &str) -> StringValue {
465        StringValue {
466            value: value.to_string(),
467            ..Default::default()
468        }
469    }
470
471    #[test]
472    fn raw_bytes_owned_encode_decode_round_trips() {
473        let payload = b"raw payload".as_slice();
474
475        let encoded = RawBytes::encode_payload_owned(payload).expect("encode raw bytes");
476        let decoded: Vec<u8> = RawBytes::decode_payload(&encoded).expect("decode raw bytes");
477
478        assert_eq!(encoded.as_ref(), payload);
479        assert_eq!(decoded, payload);
480        assert_eq!(RawBytes::encoding(), PayloadEncoding::RAW);
481    }
482
483    #[test]
484    fn raw_bytes_rejects_too_small_output_buffer() {
485        let error = RawBytes::encode_payload(b"payload".as_slice(), &mut [0_u8; 3]).unwrap_err();
486
487        assert_eq!(
488            error,
489            UWireError::BufferTooSmall {
490                expected: 7,
491                actual: 3,
492            }
493        );
494    }
495
496    #[cfg(feature = "protobuf-support")]
497    #[test]
498    fn protobuf_payload_encode_decode_round_trips() {
499        let input = message("protobuf payload");
500
501        let encoded = ProtobufPayload::encode_payload_owned(&input).expect("encode protobuf");
502        let decoded: StringValue =
503            ProtobufPayload::decode_payload(&encoded).expect("decode protobuf");
504
505        assert_eq!(decoded.value, input.value);
506        assert_eq!(ProtobufPayload::encoding(), PayloadEncoding::PROTOBUF);
507    }
508
509    #[cfg(feature = "protobuf-support")]
510    #[test]
511    fn protobuf_any_payload_encode_decode_round_trips() {
512        let input = message("protobuf any payload");
513
514        let encoded = ProtobufAnyPayload::encode_payload_owned(&input).expect("encode any");
515        let decoded: StringValue =
516            ProtobufAnyPayload::decode_payload(&encoded).expect("decode any");
517
518        assert_eq!(decoded.value, input.value);
519        assert_eq!(
520            ProtobufAnyPayload::encoding(),
521            PayloadEncoding::PROTOBUF_WRAPPED_IN_ANY
522        );
523    }
524
525    #[cfg(feature = "protobuf-support")]
526    #[test]
527    fn protobuf_payload_reader_decode_round_trips() {
528        let input = message("protobuf reader payload");
529        let encoded = ProtobufPayload::encode_payload_owned(&input).expect("encode protobuf");
530
531        let decoded: StringValue = ProtobufPayload::decode_payload_from_reader(
532            Cursor::new(encoded.as_ref()),
533            encoded.len(),
534        )
535        .expect("decode protobuf from reader");
536
537        assert_eq!(decoded.value, input.value);
538    }
539
540    #[cfg(feature = "zero-copy-transport")]
541    #[test]
542    fn segmented_frame_view_decodes_from_reader_without_contiguous_payload() {
543        let message = UMessageBuilder::publish(topic()).build().expect("message");
544        let metadata = crate::frame::metadata::try_project_attributes_to_frame_metadata(
545            message.attributes(),
546            Some(PayloadEncoding::RAW),
547        )
548        .expect("metadata");
549        let frame = SegmentedFrame {
550            metadata,
551            first: b"seg".to_vec(),
552            second: b"mented".to_vec(),
553        };
554
555        let decoded: Vec<u8> = frame
556            .decode_payload_from_reader_as::<RawBytes, _>()
557            .expect("decode segmented payload");
558
559        assert_eq!(decoded, b"segmented".as_slice());
560        assert!(frame.try_contiguous_payload().is_none());
561    }
562
563    #[cfg(feature = "zero-copy-transport")]
564    struct SegmentedFrame {
565        metadata: UFrameMetadata,
566        first: Vec<u8>,
567        second: Vec<u8>,
568    }
569
570    #[cfg(feature = "zero-copy-transport")]
571    impl UFrameView for SegmentedFrame {
572        type PayloadReader<'a>
573            = Chain<Cursor<&'a [u8]>, Cursor<&'a [u8]>>
574        where
575            Self: 'a;
576        type PayloadSlices<'a>
577            = std::array::IntoIter<&'a [u8], 2>
578        where
579            Self: 'a;
580
581        fn metadata(&self) -> &UFrameMetadata {
582            &self.metadata
583        }
584
585        fn payload_len(&self) -> usize {
586            self.first.len() + self.second.len()
587        }
588
589        fn has_payload(&self) -> bool {
590            true
591        }
592
593        fn payload_reader(&self) -> Self::PayloadReader<'_> {
594            Cursor::new(self.first.as_slice()).chain(Cursor::new(self.second.as_slice()))
595        }
596
597        fn payload_slices(&self) -> Self::PayloadSlices<'_> {
598            [self.first.as_slice(), self.second.as_slice()].into_iter()
599        }
600    }
601}
up_rust/payload/codec.rs

up_rust/payload/
codec.rs
