up_rust/umessage.rs
1/********************************************************************************
2 * Copyright (c) 2023 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
16#[cfg(all(feature = "communication-api", feature = "protobuf-support"))]
17pub(crate) use protobuf_support::deserialize_protobuf_bytes;
18pub use umessagebuilder::*;
19
20use crate::{
21 SerializationError, UAttributes, UAttributesError, UCode, UMessageType, UPayloadFormat,
22 UPriority, UUri, UUID,
23};
24
25mod umessagebuilder;
26
27pub(crate) type Payload = Bytes;
28
29#[derive(Debug)]
30/// Errors produced when building or converting a [`UMessage`].
31pub enum UMessageError {
32 /// The message attributes failed validation.
33 AttributesValidationError(UAttributesError),
34 /// The payload could not be (de)serialized.
35 DataSerializationError(SerializationError),
36 /// The payload is inconsistent with the message (wrong format or presence).
37 PayloadError(String),
38}
39
40impl std::fmt::Display for UMessageError {
41 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42 match self {
43 Self::AttributesValidationError(e) => f.write_fmt(format_args!(
44 "Builder state is not consistent with message type: {e}"
45 )),
46 Self::DataSerializationError(e) => {
47 f.write_fmt(format_args!("Failed to serialize payload: {e}"))
48 }
49 Self::PayloadError(e) => f.write_fmt(format_args!("UMessage payload error: {e}")),
50 }
51 }
52}
53
54impl std::error::Error for UMessageError {}
55
56impl From<UAttributesError> for UMessageError {
57 fn from(value: UAttributesError) -> Self {
58 Self::AttributesValidationError(value)
59 }
60}
61
62impl From<SerializationError> for UMessageError {
63 fn from(value: SerializationError) -> Self {
64 Self::DataSerializationError(value)
65 }
66}
67
68#[cfg(feature = "protobuf-support")]
69impl From<protobuf::Error> for UMessageError {
70 fn from(value: protobuf::Error) -> Self {
71 Self::DataSerializationError(value.into())
72 }
73}
74
75impl From<String> for UMessageError {
76 fn from(value: String) -> Self {
77 Self::PayloadError(value)
78 }
79}
80
81impl From<&str> for UMessageError {
82 fn from(value: &str) -> Self {
83 Self::from(value.to_string())
84 }
85}
86
87#[derive(Debug, Clone, PartialEq)]
88#[repr(C)]
89/// One uProtocol message: attributes plus optional payload bytes.
90pub struct UMessage {
91 attributes: UAttributes,
92 payload: Option<Payload>,
93}
94
95impl UMessage {
96 /// Returns this message with an assumed payload encoding stamped on its
97 /// attributes when the message has payload bytes and no encoding
98 /// declaration yet.
99 ///
100 /// This is for transports whose wire profile fixes payload encoding by
101 /// topic convention rather than carrying it per message. Existing encoding
102 /// declarations are preserved unchanged.
103 #[must_use]
104 pub fn with_assumed_payload_encoding(mut self, encoding: &crate::PayloadEncoding) -> Self {
105 if self.payload.is_none() {
106 return self;
107 }
108
109 let attributes = &mut self.attributes;
110 let has_legacy = !matches!(
111 attributes.payload_format,
112 None | Some(crate::UPayloadFormat::Unspecified)
113 );
114 let has_open = attributes.open_payload_encoding_parts() != (None, None, None);
115 if has_legacy || has_open {
116 return self;
117 }
118
119 if let Some(format) = encoding.to_legacy_format() {
120 attributes.payload_format = Some(format);
121 } else {
122 attributes.payload_encoding_registry_id = encoding.registry_id();
123 attributes.payload_encoding = encoding.literal_id().map(str::to_owned);
124 attributes.payload_content_type = encoding.content_type().map(str::to_owned);
125 }
126 self
127 }
128
129 // This convenience constructor is used internally only, e.g. by the UMessageBuilder for creating
130 // the final message after validation.
131 // Client code can create UMessages using the UMessageBuilder only.
132 pub(crate) fn new(
133 attributes: UAttributes,
134 payload: Option<Bytes>,
135 ) -> Result<Self, UMessageError> {
136 Ok(UMessage {
137 attributes,
138 payload,
139 })
140 }
141
142 /// Get this message's attributes.
143 ///
144 /// This function simply delegates to [`UAttributes::type_`].
145 ///
146 /// # Example
147 ///
148 /// ```rust
149 /// use up_rust::{UMessageBuilder, UMessageType, UUri};
150 ///
151 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
152 /// let topic = UUri::try_from("//my-vehicle/D45/23/A001")?;
153 /// let msg = UMessageBuilder::publish(topic).build()?;
154 /// assert_eq!(msg.type_(), UMessageType::Publish);
155 /// # Ok(())
156 /// # }
157 /// ```
158 #[must_use]
159 pub fn attributes(&self) -> &UAttributes {
160 &self.attributes
161 }
162
163 /// Gets this message's type.
164 #[must_use]
165 pub fn type_(&self) -> UMessageType {
166 self.attributes().type_()
167 }
168
169 /// Gets this message's identifier.
170 ///
171 /// This function simply delegates to [`UAttributes::id`].
172 ///
173 /// # Example
174 ///
175 /// ```rust
176 /// use up_rust::{UMessageBuilder, UMessageType, UUri, UUID};
177 ///
178 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
179 /// let topic = UUri::try_from("//my-vehicle/D45/23/A001")?;
180 /// let msg_id = UUID::build();
181 /// let msg = UMessageBuilder::publish(topic).with_message_id(msg_id.clone()).build()?;
182 /// assert_eq!(msg.id(), &msg_id);
183 /// # Ok(())
184 /// # }
185 /// ```
186 #[must_use]
187 pub fn id(&self) -> &UUID {
188 self.attributes().id()
189 }
190
191 /// Gets this message's source address.
192 ///
193 /// This function simply delegates to [`UAttributes::source`].
194 ///
195 /// # Example
196 ///
197 /// ```rust
198 /// use up_rust::{UMessageBuilder, UMessageType, UUri};
199 ///
200 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
201 /// let topic = UUri::try_from("//my-vehicle/D45/23/A001")?;
202 /// let msg = UMessageBuilder::publish(topic.clone()).build()?;
203 /// assert_eq!(msg.source(), &topic);
204 /// # Ok(())
205 /// # }
206 /// ```
207 #[must_use]
208 pub fn source(&self) -> &UUri {
209 self.attributes().source()
210 }
211
212 /// Gets this message's sink address.
213 ///
214 /// This function simply delegates to [`UAttributes::sink`].
215 ///
216 /// # Example
217 ///
218 /// ```rust
219 /// use up_rust::{UMessageBuilder, UMessageType, UUri};
220 ///
221 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
222 /// let origin = UUri::try_from("//my-vehicle/D45/23/A001")?;
223 /// let dest = UUri::try_from("//other-vehicle/D10/3/0")?;
224 /// let msg = UMessageBuilder::notification(origin, dest.clone()).build()?;
225 /// assert!(msg.sink().is_some_and(|sink| sink == &dest));
226 /// # Ok(())
227 /// # }
228 /// ```
229 #[must_use]
230 pub fn sink(&self) -> Option<&UUri> {
231 self.attributes().sink()
232 }
233
234 /// Gets this message's sink address.
235 ///
236 /// This function simply delegates to [`UAttributes::sink_unchecked`].
237 ///
238 /// # Panics
239 ///
240 /// if the property has no value.
241 ///
242 /// # Example
243 ///
244 /// ```rust
245 /// use up_rust::{UMessageBuilder, UMessageType, UUri};
246 ///
247 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
248 /// let origin = UUri::try_from("//my-vehicle/D45/23/A001")?;
249 /// let dest = UUri::try_from("//other-vehicle/D10/3/0")?;
250 /// let msg = UMessageBuilder::notification(origin, dest.clone()).build()?;
251 /// assert_eq!(msg.sink_unchecked(), &dest);
252 /// # Ok(())
253 /// # }
254 /// ```
255 #[must_use]
256 pub fn sink_unchecked(&self) -> &UUri {
257 self.attributes().sink_unchecked()
258 }
259
260 /// Gets this message's priority.
261 ///
262 /// This function simply delegates to [`UAttributes::priority`].
263 ///
264 /// # Example
265 ///
266 /// ```rust
267 /// use up_rust::{UMessageBuilder, UMessageType, UPriority, UUri};
268 ///
269 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
270 /// let topic = UUri::try_from("//my-vehicle/D45/23/A001")?;
271 /// let msg = UMessageBuilder::publish(topic).with_priority(UPriority::CS3).build()?;
272 /// assert!(msg.priority().is_some_and(|prio| prio == UPriority::CS3));
273 /// # Ok(())
274 /// # }
275 /// ```
276 #[must_use]
277 pub fn priority(&self) -> Option<UPriority> {
278 self.attributes().priority()
279 }
280
281 /// Gets this message's priority.
282 ///
283 /// This function simply delegates to [`UAttributes::priority_unchecked`].
284 ///
285 /// # Panics
286 ///
287 /// if the property has no value.
288 ///
289 /// # Example
290 ///
291 /// ```rust
292 /// use up_rust::{UMessageBuilder, UMessageType, UPriority, UUri};
293 ///
294 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
295 /// let topic = UUri::try_from("//my-vehicle/D45/23/A001")?;
296 /// let msg = UMessageBuilder::publish(topic).with_priority(UPriority::CS3).build()?;
297 /// assert_eq!(msg.priority_unchecked(), UPriority::CS3);
298 /// # Ok(())
299 /// # }
300 /// ```
301 #[must_use]
302 pub fn priority_unchecked(&self) -> UPriority {
303 self.attributes().priority_unchecked()
304 }
305
306 /// Gets this message's commstatus.
307 ///
308 /// This function simply delegates to [`UAttributes::commstatus`].
309 ///
310 /// # Example
311 ///
312 /// ```rust
313 /// use up_rust::{UCode, UMessageBuilder, UMessageType, UUID, UUri};
314 ///
315 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
316 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
317 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
318 /// let msg = UMessageBuilder::response(reply_to, UUID::build(), invoked_method)
319 /// .with_comm_status(UCode::Ok)
320 /// .build()?;
321 /// assert!(msg.commstatus().is_some_and(|status| status == UCode::Ok));
322 /// # Ok(())
323 /// # }
324 /// ```
325 #[must_use]
326 pub fn commstatus(&self) -> Option<UCode> {
327 self.attributes().commstatus()
328 }
329
330 /// Gets this message's commstatus.
331 ///
332 /// This function simply delegates to [`UAttributes::commstatus_unchecked`].
333 ///
334 /// # Panics
335 ///
336 /// if the property has no value.
337 ///
338 /// # Example
339 ///
340 /// ```rust
341 /// use up_rust::{UCode, UMessageBuilder, UMessageType, UUID, UUri};
342 ///
343 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
344 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
345 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
346 /// let msg = UMessageBuilder::response(reply_to, UUID::build(), invoked_method)
347 /// .with_comm_status(UCode::Internal)
348 /// .build()?;
349 /// assert_eq!(msg.commstatus_unchecked(), UCode::Internal);
350 /// # Ok(())
351 /// # }
352 /// ```
353 #[must_use]
354 pub fn commstatus_unchecked(&self) -> UCode {
355 self.attributes().commstatus_unchecked()
356 }
357
358 /// Gets this message's time-to-live.
359 ///
360 /// This function simply delegates to [`UAttributes::ttl`].
361 ///
362 /// # Returns
363 ///
364 /// the time-to-live in milliseconds.
365 ///
366 /// # Example
367 ///
368 /// ```rust
369 /// use up_rust::{UMessageBuilder, UUri};
370 ///
371 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
372 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
373 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
374 /// let msg = UMessageBuilder::request(invoked_method, reply_to, 5000)
375 /// .build()?;
376 /// assert!(msg.ttl().is_some_and(|ttl| ttl == 5000));
377 /// # Ok(())
378 /// # }
379 /// ```
380 #[must_use]
381 pub fn ttl(&self) -> Option<u32> {
382 self.attributes().ttl()
383 }
384
385 /// Gets this message's time-to-live.
386 ///
387 /// This function simply delegates to [`UAttributes::ttl_unchecked`].
388 ///
389 /// # Returns
390 ///
391 /// the time-to-live in milliseconds.
392 ///
393 /// # Panics
394 ///
395 /// if the property has no value.
396 ///
397 /// # Example
398 ///
399 /// ```rust
400 /// use up_rust::{UMessageBuilder, UUri};
401 ///
402 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
403 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
404 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
405 /// let msg = UMessageBuilder::request(invoked_method, reply_to, 5000)
406 /// .build()?;
407 /// assert!(msg.ttl_unchecked() == 5000);
408 /// # Ok(())
409 /// # }
410 /// ```
411 #[must_use]
412 pub fn ttl_unchecked(&self) -> u32 {
413 self.attributes().ttl_unchecked()
414 }
415
416 /// Gets this message's permission level.
417 ///
418 /// This function simply delegates to [`UAttributes::permission_level`].
419 ///
420 /// # Example
421 ///
422 /// ```rust
423 /// use up_rust::{UMessageBuilder, UUri};
424 ///
425 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
426 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
427 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
428 /// let msg = UMessageBuilder::request(invoked_method, reply_to, 5000)
429 /// .with_permission_level(3)
430 /// .build()?;
431 /// assert!(msg.permission_level().is_some_and(|pl| pl == 3));
432 /// # Ok(())
433 /// # }
434 /// ```
435 #[must_use]
436 pub fn permission_level(&self) -> Option<u32> {
437 self.attributes().permission_level()
438 }
439
440 /// Gets this message's token.
441 ///
442 /// This function simply delegates to [`UAttributes::token`].
443 ///
444 /// # Example
445 ///
446 /// ```rust
447 /// use up_rust::{UMessageBuilder, UUri};
448 ///
449 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
450 /// let token = "my_token";
451 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
452 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
453 /// let msg = UMessageBuilder::request(invoked_method, reply_to, 5000)
454 /// .with_token(token)
455 /// .build()?;
456 /// assert!(msg.token().is_some_and(|t| t == token));
457 /// # Ok(())
458 /// # }
459 /// ```
460 #[must_use]
461 pub fn token(&self) -> Option<&str> {
462 self.attributes().token()
463 }
464
465 /// Gets this message's traceparent.
466 ///
467 /// This function simply delegates to [`UAttributes::traceparent`].
468 ///
469 /// # Example
470 ///
471 /// ```rust
472 /// use up_rust::{UMessageBuilder, UUri};
473 ///
474 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
475 /// let traceparent = "my_traceparent";
476 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
477 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
478 /// let msg = UMessageBuilder::request(invoked_method, reply_to, 5000)
479 /// .with_traceparent(traceparent)
480 /// .build()?;
481 /// assert!(msg.traceparent().is_some_and(|tp| tp == traceparent));
482 /// # Ok(())
483 /// # }
484 /// ```
485 #[must_use]
486 pub fn traceparent(&self) -> Option<&str> {
487 self.attributes().traceparent()
488 }
489
490 /// Gets this message's request identifier.
491 ///
492 /// This function simply delegates to [`UAttributes::request_id`].
493 ///
494 /// # Example
495 ///
496 /// ```rust
497 /// use up_rust::{UCode, UMessageBuilder, UUID, UUri};
498 ///
499 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
500 /// let request_id = UUID::build();
501 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
502 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
503 /// let msg = UMessageBuilder::response(reply_to, request_id.clone(), invoked_method)
504 /// .build()?;
505 /// assert!(msg.request_id().is_some_and(|id| id == &request_id));
506 /// # Ok(())
507 /// # }
508 /// ```
509 #[must_use]
510 pub fn request_id(&self) -> Option<&UUID> {
511 self.attributes().request_id()
512 }
513
514 /// Gets this message's request identifier.
515 ///
516 /// This function simply delegates to [`UAttributes::request_id_unchecked`].
517 ///
518 /// # Panics
519 ///
520 /// if the property has no value.
521 ///
522 /// # Example
523 ///
524 /// ```rust
525 /// use up_rust::{UCode, UMessageBuilder, UUID, UUri};
526 ///
527 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
528 /// let request_id = UUID::build();
529 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
530 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
531 /// let msg = UMessageBuilder::response(reply_to, request_id.clone(), invoked_method)
532 /// .build()?;
533 /// assert_eq!(msg.request_id_unchecked(), &request_id);
534 /// # Ok(())
535 /// # }
536 /// ```
537 #[must_use]
538 pub fn request_id_unchecked(&self) -> &UUID {
539 self.attributes().request_id_unchecked()
540 }
541
542 /// Gets this message's payload format.
543 ///
544 /// This function simply delegates to [`UAttributes::payload_format`].
545 ///
546 /// # Example
547 ///
548 /// ```rust
549 /// use up_rust::{UMessageBuilder, UPayloadFormat, UUri};
550 ///
551 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
552 /// let topic = UUri::try_from("//my-vehicle/D45/23/A001")?;
553 /// let msg = UMessageBuilder::publish(topic)
554 /// .build_with_payload("hello".as_bytes(), UPayloadFormat::Text)?;
555 /// assert!(msg.payload_format().is_some_and(|format| format == UPayloadFormat::Text));
556 /// # Ok(())
557 /// # }
558 /// ```
559 #[must_use]
560 pub fn payload_format(&self) -> Option<UPayloadFormat> {
561 self.attributes().payload_format()
562 }
563
564 /// Gets this message's payload format.
565 ///
566 /// This function simply delegates to [`UAttributes::payload_format_unchecked`].
567 ///
568 /// # Panics
569 ///
570 /// if the property has no value.
571 ///
572 /// # Example
573 ///
574 /// ```rust
575 /// use up_rust::{UMessageBuilder, UPayloadFormat, UUri};
576 ///
577 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
578 /// let topic = UUri::try_from("//my-vehicle/D45/23/A001")?;
579 /// let msg = UMessageBuilder::publish(topic)
580 /// .build_with_payload("hello".as_bytes(), UPayloadFormat::Text)?;
581 /// assert_eq!(msg.payload_format_unchecked(), UPayloadFormat::Text);
582 /// # Ok(())
583 /// # }
584 /// ```
585 #[must_use]
586 pub fn payload_format_unchecked(&self) -> UPayloadFormat {
587 self.attributes().payload_format_unchecked()
588 }
589
590 #[must_use]
591 /// Returns the payload bytes, if the message carries any.
592 pub fn payload(&self) -> Option<Bytes> {
593 self.payload.clone()
594 }
595
596 /// Checks if this is a Publish message.
597 ///
598 /// This function simply delegates to [`UAttributes::is_publish`].
599 ///
600 /// # Examples
601 ///
602 /// ```rust
603 /// use up_rust::{UMessageBuilder, UUri};
604 ///
605 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
606 /// let topic = UUri::try_from("//my-vehicle/D45/23/A001")?;
607 /// let msg = UMessageBuilder::publish(topic).build()?;
608 /// assert!(msg.is_publish());
609 /// # Ok(())
610 /// # }
611 /// ```
612 #[must_use]
613 pub fn is_publish(&self) -> bool {
614 self.attributes().is_publish()
615 }
616
617 /// Checks if this is an RPC Request message.
618 ///
619 /// This function simply delegates to [`UAttributes::is_request`].
620 ///
621 /// # Examples
622 ///
623 /// ```rust
624 /// use up_rust::{UMessageBuilder, UUri};
625 ///
626 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
627 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
628 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
629 /// let msg = UMessageBuilder::request(invoked_method, reply_to, 5000)
630 /// .build()?;
631 /// assert!(msg.is_request());
632 /// # Ok(())
633 /// # }
634 /// ```
635 #[must_use]
636 pub fn is_request(&self) -> bool {
637 self.attributes().is_request()
638 }
639
640 /// Checks if this is an RPC Response message.
641 ///
642 /// This function simply delegates to [`UAttributes::is_response`].
643 ///
644 /// # Examples
645 ///
646 /// ```rust
647 /// use up_rust::{UCode, UMessageBuilder, UUID, UUri};
648 ///
649 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
650 /// let invoked_method = UUri::try_from("//my-vehicle/D45/2/101")?;
651 /// let reply_to = UUri::try_from("//other-vehicle/D10/3/0")?;
652 /// let msg = UMessageBuilder::response(reply_to, UUID::build(), invoked_method)
653 /// .build()?;
654 /// assert!(msg.is_response());
655 /// # Ok(())
656 /// # }
657 /// ```
658 #[must_use]
659 pub fn is_response(&self) -> bool {
660 self.attributes().is_response()
661 }
662
663 /// Checks if this is a Notification message.
664 ///
665 /// This function simply delegates to [`UAttributes::is_notification`].
666 ///
667 /// # Examples
668 ///
669 /// ```rust
670 /// use up_rust::{UMessageBuilder, UUri};
671 ///
672 /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
673 /// let origin = UUri::try_from("//my-vehicle/D45/23/A001")?;
674 /// let dest = UUri::try_from("//other-vehicle/D10/3/0")?;
675 /// let msg = UMessageBuilder::notification(origin, dest).build()?;
676 /// assert!(msg.is_notification());
677 /// # Ok(())
678 /// # }
679 /// ```
680 #[must_use]
681 pub fn is_notification(&self) -> bool {
682 self.attributes().is_notification()
683 }
684
685 /// Checks if this message should be considered expired.
686 ///
687 /// This function simply delegates to [`UAttributes::check_expired`].
688 pub fn check_expired(&self) -> Result<(), UAttributesError> {
689 self.attributes().check_expired()
690 }
691
692 /// Checks if this message should be considered expired for a given reference time.
693 ///
694 /// This function simply delegates to [`UAttributes::check_expired_for_reference`].
695 ///
696 /// # Arguments
697 ///
698 /// * `reference_time` - The reference time as milliseconds since UNIX epoch. The check will
699 /// be performed in relation to this point in time.
700 pub fn check_expired_for_reference(
701 &self,
702 reference_time: u128,
703 ) -> Result<(), UAttributesError> {
704 self.attributes()
705 .check_expired_for_reference(reference_time)
706 }
707}
708
709#[cfg(feature = "protobuf-support")]
710mod protobuf_support {
711 use super::*;
712 use crate::ProtobufMappable;
713
714 impl UMessage {
715 /// Deserializes this message's protobuf payload into a type.
716 ///
717 /// # Type Parameters
718 ///
719 /// * `T`: The target type of the data to be unpacked.
720 ///
721 /// # Errors
722 ///
723 /// Returns an error if the message payload format is neither [UPayloadFormat::Protobuf] nor
724 /// [UPayloadFormat::ProtobufWrappedInAny] or if the bytes in the
725 /// payload cannot be deserialized into the target type.
726 #[cfg(feature = "protobuf-support")]
727 pub fn extract_protobuf<T: ProtobufMappable>(&self) -> Result<T, UMessageError> {
728 if let Some(payload) = self.payload.as_ref() {
729 let payload_format = self.payload_format().unwrap_or(UPayloadFormat::Unspecified);
730 deserialize_protobuf_bytes(payload, &payload_format)
731 } else {
732 Err(UMessageError::PayloadError(
733 "Message has no payload".to_string(),
734 ))
735 }
736 }
737 }
738
739 /// Deserializes a protobuf message from a byte array.
740 ///
741 /// # Type Parameters
742 ///
743 /// * `T`: The target type of the data to be unpacked.
744 ///
745 /// # Arguments
746 ///
747 /// * `payload` - The payload data.
748 /// * `payload_format` - The format/encoding of the data. Must be one of
749 /// - `UPayloadFormat::UPAYLOAD_FORMAT_PROTOBUF`
750 /// - `UPayloadFormat::UPAYLOAD_FORMAT_PROTOBUF_WRAPPED_IN_ANY`
751 ///
752 /// # Errors
753 ///
754 /// Returns an error if the payload format is unsupported or if the data can not be deserialized
755 /// into the target type based on the given format.
756 #[cfg(feature = "protobuf-support")]
757 pub(crate) fn deserialize_protobuf_bytes<T: ProtobufMappable>(
758 payload: &[u8],
759 payload_format: &UPayloadFormat,
760 ) -> Result<T, UMessageError> {
761 match payload_format {
762 UPayloadFormat::Protobuf => {
763 T::parse_from_protobuf_bytes(payload).map_err(UMessageError::DataSerializationError)
764 }
765 UPayloadFormat::ProtobufWrappedInAny => T::parse_from_packed_protobuf_bytes(payload)
766 .map_err(UMessageError::DataSerializationError),
767 UPayloadFormat::Unspecified
768 | UPayloadFormat::Json
769 | UPayloadFormat::Raw
770 | UPayloadFormat::Shm
771 | UPayloadFormat::Someip
772 | UPayloadFormat::SomeipTlv
773 | UPayloadFormat::Text => {
774 let detail_msg = payload_format.to_media_type().map_or_else(
775 || format!("Unknown payload format: {}", *payload_format as i32),
776 |mt| format!("Invalid/unsupported payload format: {mt}"),
777 );
778 Err(UMessageError::from(detail_msg))
779 }
780 }
781 }
782
783 #[cfg(test)]
784 mod protobuf_support_test {
785 use super::*;
786 use crate::UUri;
787
788 use protobuf::well_known_types::{
789 any::Any,
790 duration::Duration,
791 wrappers::{DoubleValue, StringValue},
792 };
793 use protobuf::Message;
794 use test_case::test_case;
795
796 #[test]
797 fn test_from_protobuf_error() {
798 let protobuf_error = protobuf::Error::from(std::io::Error::last_os_error());
799 let message_error = UMessageError::from(protobuf_error);
800 assert!(matches!(
801 message_error,
802 UMessageError::DataSerializationError(_)
803 ));
804 }
805
806 #[test]
807 fn test_deserialize_protobuf_bytes_succeeds() {
808 let mut data = StringValue::new();
809 data.value = "hello world".to_string();
810
811 let result = deserialize_protobuf_bytes::<StringValue>(
812 &data
813 .write_to_bytes()
814 .expect("Failed to write protobuf bytes"),
815 &UPayloadFormat::Protobuf,
816 );
817 assert!(result.is_ok_and(|v| v.value == *"hello world"));
818
819 let any = Any::pack(&data).expect("Failed to pack Any");
820 let buf: Bytes = any
821 .write_to_bytes()
822 .expect("Failed to write protobuf bytes")
823 .into();
824
825 let result = deserialize_protobuf_bytes::<StringValue>(
826 &buf,
827 &UPayloadFormat::ProtobufWrappedInAny,
828 );
829 assert!(result.is_ok_and(|v| v.value == *"hello world"));
830 }
831
832 #[test]
833 fn test_deserialize_protobuf_bytes_fails_for_payload_type_mismatch() {
834 let mut data = StringValue::new();
835 data.value = "hello world".to_string();
836 let any = Any::pack(&data).unwrap();
837 let buf: Bytes = any.write_to_bytes().unwrap().into();
838 let result = deserialize_protobuf_bytes::<DoubleValue>(
839 &buf,
840 &UPayloadFormat::ProtobufWrappedInAny,
841 );
842 assert!(result.is_err_and(|e| matches!(e, UMessageError::DataSerializationError(_))));
843 }
844
845 #[test_case(UPayloadFormat::Json; "JSON format")]
846 #[test_case(UPayloadFormat::Raw; "RAW format")]
847 #[test_case(UPayloadFormat::Shm; "SHM format")]
848 #[test_case(UPayloadFormat::Someip; "SOMEIP format")]
849 #[test_case(UPayloadFormat::SomeipTlv; "SOMEIP TLV format")]
850 #[test_case(UPayloadFormat::Text; "TEXT format")]
851 #[test_case(UPayloadFormat::Unspecified; "UNSPECIFIED format")]
852 fn test_deserialize_protobuf_bytes_fails_for_(format: UPayloadFormat) {
853 let result = deserialize_protobuf_bytes::<StringValue>("hello".as_bytes(), &format);
854 assert!(result.is_err_and(|e| matches!(e, UMessageError::PayloadError(_))));
855 }
856
857 #[test]
858 fn test_deserialize_protobuf_bytes_fails_for_invalid_encoding() {
859 // GIVEN a protobuf Any message with an embedded Duration message, but with invalid encoding
860 // (i.e. the value field does not contain a valid protobuf encoding of a Duration message)
861 let any = Any {
862 type_url: "type.googleapis.com/google.protobuf.Duration".to_string(),
863 value: vec![0x0A],
864 ..Default::default()
865 };
866 let buf = any.write_to_bytes().unwrap();
867
868 // WHEN deserializing the bytes into a Duration message using the ProtobufWrappedInAny format
869 let result = deserialize_protobuf_bytes::<Duration>(
870 buf.as_slice(),
871 &UPayloadFormat::ProtobufWrappedInAny,
872 );
873 // THEN the deserialization fails with a DataSerializationError
874 assert!(result.is_err_and(|e| matches!(e, UMessageError::DataSerializationError(_))))
875 }
876
877 #[test]
878 fn extract_payload_succeeds() {
879 let payload = StringValue {
880 value: "hello".to_string(),
881 ..Default::default()
882 };
883 let topic = UUri::try_from_parts("local", 0xabcd, 0x01, 0x9000)
884 .expect("failed to create topic");
885 let msg = UMessageBuilder::publish(topic)
886 .build_with_protobuf_payload(&payload)
887 .expect("failed to create message");
888 assert!(msg
889 .extract_protobuf::<StringValue>()
890 .is_ok_and(|v| v.value == *"hello"));
891 }
892
893 #[test]
894 fn extract_payload_fails_for_no_payload() {
895 let topic = UUri::try_from_parts("local", 0xabcd, 0x01, 0x9000)
896 .expect("failed to create topic");
897 let msg = UMessageBuilder::publish(topic)
898 .build()
899 .expect("failed to create message");
900 assert!(msg
901 .extract_protobuf::<StringValue>()
902 .is_err_and(|e| matches!(e, UMessageError::PayloadError(_))));
903 }
904 }
905}
906
907#[cfg(feature = "up-core-api")]
908mod core_types_support {
909 use protobuf::{well_known_types::any::Any, Message};
910
911 use super::*;
912 use crate::up_core_api::uattributes::UAttributes as UAttributesProto;
913 use crate::up_core_api::umessage::UMessage as UMessageProto;
914 use crate::ProtobufMappable;
915
916 impl From<&UMessage> for UMessageProto {
917 fn from(value: &UMessage) -> Self {
918 let attributes = UAttributesProto::from(&value.attributes);
919 UMessageProto {
920 attributes: Some(attributes).into(),
921 payload: value.payload.clone(),
922 ..Default::default()
923 }
924 }
925 }
926
927 impl TryFrom<&UMessageProto> for UMessage {
928 type Error = UMessageError;
929 fn try_from(value: &UMessageProto) -> Result<Self, Self::Error> {
930 let mut attributes = value.attributes.as_ref().map_or_else(
931 || {
932 Err(UAttributesError::validation_error(
933 "UMessageProto has no attributes",
934 ))
935 },
936 UAttributes::try_from,
937 )?;
938 // The uattributes.proto file in up-spec v1.6.0.alpha.7 does not declare the payload_format field
939 // as optional. Consequently, a UMessage protobuf that does not have a payload still ALWAYS has
940 // the UNSPECIFIED payload_format when being deserialized.
941 // When mapping such a message to the (internal) UMessage struct, we therefore need to also consider
942 // whether the message actually has payload or not, in order to set the payload_format field to the
943 // proper value, i.e. Some(Unspecified), if the message has payload, or None otherwise.
944 // This is relevant, because the presence of a payload_format value in the UMessage struct is used to
945 // determine whether the message has payload or not, e.g. when serializing the message back to
946 // protobuf or when extracting the payload as a protobuf message.
947 //
948 // This should no longer be necessary once the payload_format field in uattributes.proto is declared as
949 // optional in the UP specification and the protobuf definitions are updated accordingly.
950 //
951 if value.payload.is_none() {
952 attributes.payload_format = None;
953 }
954 UMessage::new(attributes, value.payload.clone())
955 }
956 }
957
958 impl ProtobufMappable for UMessage {
959 fn parse_from_protobuf_bytes(proto: &[u8]) -> Result<Self, SerializationError> {
960 let proto = UMessageProto::parse_from_bytes(proto)?;
961 UMessage::try_from(&proto).map_err(|e| SerializationError::new(e.to_string()))
962 }
963
964 fn parse_from_packed_protobuf_bytes(proto: &[u8]) -> Result<Self, SerializationError> {
965 Any::parse_from_bytes(proto)
966 .map_err(|err| crate::SerializationError::new(err.to_string()))
967 .and_then(|any| match any.unpack::<UMessageProto>() {
968 Ok(Some(umessage_proto)) => UMessage::try_from(&umessage_proto)
969 .map_err(|e| crate::SerializationError::new(e.to_string())),
970 Ok(None) => Err(crate::SerializationError::new(
971 "Protobuf Any does not contain UMessage".to_string(),
972 )),
973 Err(e) => Err(crate::SerializationError::new(format!(
974 "Protobuf Any unpack error: {e}"
975 ))),
976 })
977 }
978
979 fn write_to_protobuf_bytes(&self) -> Result<Vec<u8>, SerializationError> {
980 Ok(UMessageProto::from(self).write_to_bytes()?)
981 }
982
983 fn write_to_packed_protobuf_bytes(&self) -> Result<Vec<u8>, SerializationError> {
984 Any::pack(&UMessageProto::from(self))
985 .map_err(|e| {
986 crate::SerializationError::new(format!("Failed to pack UMessage: {e}"))
987 })
988 .and_then(|any| any.write_to_protobuf_bytes())
989 }
990 }
991
992 #[cfg(test)]
993 mod test {
994 use protobuf::{Enum, EnumOrUnknown};
995
996 use super::*;
997 use crate::up_core_api::uattributes::UAttributes as UAttributesProto;
998 use crate::up_core_api::umessage::UMessage as UMessageProto;
999
1000 #[test]
1001 fn test_try_from_umessage_proto_fails_for_missing_attributes() {
1002 let proto = UMessageProto {
1003 attributes: None.into(),
1004 ..Default::default()
1005 };
1006 let result = UMessage::try_from(&proto);
1007 assert!(result.is_err_and(|e| matches!(e, UMessageError::AttributesValidationError(_))));
1008 }
1009
1010 #[test_case::test_case(None => None; "for no payload")]
1011 #[test_case::test_case(Some("payload") => Some(UPayloadFormat::Unspecified); "for some payload")]
1012 fn test_try_from_handles_payload_format(payload: Option<&str>) -> Option<UPayloadFormat> {
1013 let valid_attribs_proto = UAttributesProto {
1014 type_: EnumOrUnknown::from_i32(
1015 crate::up_core_api::uattributes::UMessageType::UMESSAGE_TYPE_PUBLISH.value(),
1016 ),
1017 id: Some(crate::up_core_api::uuid::UUID {
1018 msb: 0x0000000000017000_u64,
1019 lsb: 0x8010101010101a1a_u64,
1020 ..Default::default()
1021 })
1022 .into(),
1023 source: Some(crate::up_core_api::uri::UUri {
1024 authority_name: "source".to_string(),
1025 ue_id: 0x0001,
1026 ue_version_major: 0x01,
1027 resource_id: 0x0001,
1028 ..Default::default()
1029 })
1030 .into(),
1031 priority: EnumOrUnknown::from_i32(
1032 crate::up_core_api::uattributes::UPriority::UPRIORITY_UNSPECIFIED.value(),
1033 ),
1034 payload_format: EnumOrUnknown::from_i32(
1035 crate::up_core_api::uattributes::UPayloadFormat::UPAYLOAD_FORMAT_UNSPECIFIED
1036 .value(),
1037 ),
1038 ..Default::default()
1039 };
1040 // GIVEN a UMessageProto with valid attributes (including the UNSPECIFIED payload format) but no payload
1041 let proto = UMessageProto {
1042 attributes: Some(valid_attribs_proto).into(),
1043 payload: payload.map(|p| p.as_bytes().to_vec().into()),
1044 ..Default::default()
1045 };
1046 // WHEN converting the UMessageProto to a UMessage
1047 UMessage::try_from(&proto)
1048 .expect("failed to convert UMessageProto to UMessage")
1049 .payload_format()
1050 }
1051 }
1052}
1053
1054#[cfg(test)]
1055mod test {
1056
1057 use super::*;
1058
1059 #[test]
1060 fn test_from_attributes_error() {
1061 let attributes_error = UAttributesError::validation_error("failed to validate");
1062 let message_error = UMessageError::from(attributes_error);
1063 assert!(matches!(
1064 message_error,
1065 UMessageError::AttributesValidationError(UAttributesError::ValidationError(_))
1066 ));
1067 }
1068
1069 #[test]
1070 fn test_from_error_msg() {
1071 let message_error = UMessageError::from("an error occurred");
1072 assert!(matches!(message_error, UMessageError::PayloadError(_)));
1073 }
1074}