up_rust/
cloudevents.rs

1// SPDX-FileCopyrightText: 2024 Contributors to the Eclipse Foundation
2//
3// See the NOTICE file(s) distributed with this work for additional
4// information regarding copyright ownership.
5//
6// Licensed under the Apache License, Version 2.0 (the "License");
7// you may not use this file except in compliance with the License.
8// You may obtain a copy of the License at
9//
10//     https://www.apache.org/licenses/LICENSE-2.0
11//
12// Unless required by applicable law or agreed to in writing, software
13// distributed under the License is distributed on an "AS IS" BASIS,
14// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15// See the License for the specific language governing permissions and
16// limitations under the License.
17//
18// SPDX-License-Identifier: Apache-2.0
19
20// [impl->dsn~cloudevents-umessage-mapping~2]
21
22use crate::{
23    UAttributes, UAttributesError, UAttributesValidators, UCode, UMessage, UMessageError,
24    UMessageType, UPayloadFormat, UPriority, UUri, UUID,
25};
26use bytes::Bytes;
27use protobuf::well_known_types::any::Any;
28
29pub use cloudevents::{cloud_event::CloudEventAttributeValue, CloudEvent};
30
31include!(concat!(env!("OUT_DIR"), "/cloudevents/mod.rs"));
32
33// The _official_ content type to use for CloudEvents serialized using the
34// protobuf format.
35/// MIME content type for protobuf-encoded CloudEvents.
36pub const CONTENT_TYPE_CLOUDEVENTS_PROTOBUF: &str = "application/cloudevents+protobuf";
37
38const CLOUDEVENTS_SPEC_VERSION: &str = "1.0";
39
40const EXTENSION_NAME_COMMSTATUS: &str = "commstatus";
41const EXTENSION_NAME_PERMISSION_LEVEL: &str = "plevel";
42const EXTENSION_NAME_PFORMAT: &str = "pformat";
43const EXTENSION_NAME_PRIORITY: &str = "priority";
44const EXTENSION_NAME_REQUEST_ID: &str = "reqid";
45const EXTENSION_NAME_SINK: &str = "sink";
46const EXTENSION_NAME_TOKEN: &str = "token";
47const EXTENSION_NAME_TRACEPARENT: &str = "traceparent";
48const EXTENSION_NAME_TTL: &str = "ttl";
49
50impl CloudEvent {
51    fn get_id(&self) -> Result<UUID, UAttributesError> {
52        self.id
53            .parse::<UUID>()
54            .map_err(|e| UAttributesError::parsing_error(e.to_string()))
55    }
56
57    fn set_id(&mut self, id: &UUID) {
58        self.id = id.to_hyphenated_string();
59    }
60
61    fn get_type(&self) -> Result<UMessageType, UAttributesError> {
62        UMessageType::try_from_cloudevent_type(self.type_.clone())
63    }
64
65    fn set_type(&mut self, type_: UMessageType) {
66        self.type_ = type_.to_cloudevent_type();
67    }
68
69    fn get_source(&self) -> Result<UUri, UAttributesError> {
70        self.source
71            .parse::<UUri>()
72            .map_err(|e| UAttributesError::parsing_error(e.to_string()))
73    }
74
75    fn set_source<T: Into<String>>(&mut self, uri: T) {
76        self.source = uri.into();
77    }
78
79    fn get_sink(&self) -> Result<Option<UUri>, UAttributesError> {
80        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_SINK) {
81            if extension_value.has_ce_uri_ref() {
82                return extension_value
83                    .ce_uri_ref()
84                    .parse::<UUri>()
85                    .map(Option::Some)
86                    .map_err(|e| UAttributesError::parsing_error(e.to_string()));
87            } else {
88                return Err(UAttributesError::parsing_error(format!(
89                    "expected URI reference for {} extension attribute but found {:?}",
90                    EXTENSION_NAME_SINK, extension_value
91                )));
92            }
93        }
94        Ok(None)
95    }
96
97    fn set_sink<T: Into<String>>(&mut self, uri: T) {
98        let mut val = CloudEventAttributeValue::new();
99        val.set_ce_uri_ref(uri.into());
100        self.attributes.insert(EXTENSION_NAME_SINK.to_string(), val);
101    }
102
103    fn get_priority(&self) -> Result<UPriority, UAttributesError> {
104        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_PRIORITY) {
105            if extension_value.has_ce_string() {
106                return UPriority::try_from_priority_code(extension_value.ce_string());
107            } else {
108                return Err(UAttributesError::parsing_error(format!(
109                    "expected String value for {} extension attribute but found {:?}",
110                    EXTENSION_NAME_PRIORITY, extension_value
111                )));
112            }
113        }
114        Ok(crate::uattributes::UPRIORITY_DEFAULT)
115    }
116
117    fn set_priority(&mut self, priority: UPriority) {
118        let mut val = CloudEventAttributeValue::new();
119        val.set_ce_string(priority.to_priority_code().to_owned());
120        self.attributes
121            .insert(EXTENSION_NAME_PRIORITY.to_string(), val);
122    }
123
124    fn get_ttl(&self) -> Result<Option<u32>, UAttributesError> {
125        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_TTL) {
126            if extension_value.has_ce_integer() {
127                return u32::try_from(extension_value.ce_integer())
128                    .map(Option::Some)
129                    .map_err(|_e| UAttributesError::parsing_error(format!("expected unsigned Integer value for {} extension attribute but found {:?}", EXTENSION_NAME_TTL, extension_value)));
130            } else {
131                return Err(UAttributesError::parsing_error(format!(
132                    "expected Integer value for {} extension attribute but found {:?}",
133                    EXTENSION_NAME_TTL, extension_value
134                )));
135            }
136        }
137        Ok(None)
138    }
139
140    fn set_ttl(&mut self, ttl: u32) -> Result<(), UAttributesError> {
141        let v = i32::try_from(ttl).map_err(|e| UAttributesError::parsing_error(e.to_string()))?;
142        let mut val = CloudEventAttributeValue::new();
143        val.set_ce_integer(v);
144        self.attributes.insert(EXTENSION_NAME_TTL.to_string(), val);
145        Ok(())
146    }
147
148    fn get_token(&self) -> Result<Option<String>, UAttributesError> {
149        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_TOKEN) {
150            if extension_value.has_ce_string() {
151                return Ok(Some(extension_value.ce_string().to_string()));
152            } else {
153                return Err(UAttributesError::parsing_error(format!(
154                    "expected String value for {} extension attribute but found {:?}",
155                    EXTENSION_NAME_TOKEN, extension_value
156                )));
157            }
158        }
159        Ok(None)
160    }
161
162    fn set_token<T: Into<String>>(&mut self, token: T) {
163        let mut val = CloudEventAttributeValue::new();
164        val.set_ce_string(token.into());
165        self.attributes
166            .insert(EXTENSION_NAME_TOKEN.to_string(), val);
167    }
168
169    fn get_permission_level(&self) -> Result<Option<u32>, UAttributesError> {
170        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_PERMISSION_LEVEL) {
171            if extension_value.has_ce_integer() {
172                return u32::try_from(extension_value.ce_integer())
173                    .map(Option::Some)
174                    .map_err(|_e| UAttributesError::parsing_error(format!("expected unsigned Integer value for {} extension attribute but found {:?}", EXTENSION_NAME_PERMISSION_LEVEL, extension_value)));
175            } else {
176                return Err(UAttributesError::parsing_error(format!(
177                    "expected Integer value for {} extension attribute but found {:?}",
178                    EXTENSION_NAME_PERMISSION_LEVEL, extension_value
179                )));
180            }
181        }
182        Ok(None)
183    }
184
185    fn set_permission_level(&mut self, level: u32) -> Result<(), UAttributesError> {
186        let v = i32::try_from(level).map_err(|e| UAttributesError::parsing_error(e.to_string()))?;
187        let mut val = CloudEventAttributeValue::new();
188        val.set_ce_integer(v);
189        self.attributes
190            .insert(EXTENSION_NAME_PERMISSION_LEVEL.to_string(), val);
191        Ok(())
192    }
193
194    fn get_request_id(&self) -> Result<Option<UUID>, UAttributesError> {
195        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_REQUEST_ID) {
196            if extension_value.has_ce_string() {
197                return extension_value
198                    .ce_string()
199                    .parse::<UUID>()
200                    .map(Option::Some)
201                    .map_err(|e| UAttributesError::parsing_error(e.to_string()));
202            } else {
203                return Err(UAttributesError::parsing_error(format!(
204                    "expected String value for {} extension attribute but found {:?}",
205                    EXTENSION_NAME_REQUEST_ID, extension_value
206                )));
207            }
208        }
209        Ok(None)
210    }
211
212    fn set_request_id(&mut self, id: &UUID) {
213        let mut val = CloudEventAttributeValue::new();
214        val.set_ce_string(id.to_hyphenated_string());
215        self.attributes
216            .insert(EXTENSION_NAME_REQUEST_ID.to_string(), val);
217    }
218
219    fn get_commstatus(&self) -> Result<Option<UCode>, UAttributesError> {
220        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_COMMSTATUS) {
221            if extension_value.has_ce_integer() {
222                return UCode::try_from_i32(extension_value.ce_integer())
223                    .map(Option::Some)
224                    .map_err(|_e| {
225                        UAttributesError::parsing_error(format!(
226                            "unsupported commstatus code {:?} in {} extension attribute",
227                            extension_value.ce_integer(),
228                            EXTENSION_NAME_COMMSTATUS
229                        ))
230                    });
231            } else {
232                return Err(UAttributesError::parsing_error(format!(
233                    "expected Integer value for {} extension attribute but found {:?}",
234                    EXTENSION_NAME_COMMSTATUS, extension_value
235                )));
236            }
237        }
238        Ok(None)
239    }
240
241    fn set_commstatus(&mut self, status: UCode) {
242        if status != UCode::Ok {
243            let mut val = CloudEventAttributeValue::new();
244            val.set_ce_integer(status.value());
245            self.attributes
246                .insert(EXTENSION_NAME_COMMSTATUS.to_string(), val);
247        }
248    }
249
250    fn get_traceparent(&self) -> Result<Option<String>, UAttributesError> {
251        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_TRACEPARENT) {
252            if extension_value.has_ce_string() {
253                return Ok(Some(extension_value.ce_string().to_string()));
254            } else {
255                return Err(UAttributesError::parsing_error(format!(
256                    "expected String value for {} extension attribute but found {:?}",
257                    EXTENSION_NAME_TRACEPARENT, extension_value
258                )));
259            }
260        }
261        Ok(None)
262    }
263
264    fn set_traceparent<T: Into<String>>(&mut self, traceparent: T) {
265        let mut val = CloudEventAttributeValue::new();
266        val.set_ce_string(traceparent.into());
267        self.attributes
268            .insert(EXTENSION_NAME_TRACEPARENT.to_string(), val);
269    }
270
271    fn get_payload_format(&self) -> Result<UPayloadFormat, UAttributesError> {
272        if let Some(extension_value) = self.attributes.get(EXTENSION_NAME_PFORMAT) {
273            if extension_value.has_ce_integer() {
274                return UPayloadFormat::try_from_i32(extension_value.ce_integer()).map_err(|_e| {
275                    UAttributesError::parsing_error(format!(
276                        "unsupported payload format {:?} in {} extension attribute",
277                        extension_value.ce_integer(),
278                        EXTENSION_NAME_PFORMAT
279                    ))
280                });
281            } else {
282                return Err(UAttributesError::parsing_error(format!(
283                    "expected Integer value for {} extension attribute but found {:?}",
284                    EXTENSION_NAME_PFORMAT, extension_value
285                )));
286            }
287        }
288        Ok(UPayloadFormat::Unspecified)
289    }
290
291    fn set_payload_format(&mut self, format: UPayloadFormat) {
292        if format != UPayloadFormat::Unspecified {
293            let mut val = CloudEventAttributeValue::new();
294            val.set_ce_integer(format.as_i32());
295            self.attributes
296                .insert(EXTENSION_NAME_PFORMAT.to_string(), val);
297        }
298    }
299}
300
301impl TryFrom<UMessage> for CloudEvent {
302    type Error = UMessageError;
303
304    // Converts a uProtocol message into a CloudEvent using the
305    // [Protobuf Event Format](https://github.com/cloudevents/spec/blob/v1.0.2/cloudevents/formats/protobuf-format.md).
306    //
307    // # Arguments
308    //
309    // * `message` - The message to create the event from.
310    //               Note that the message is not validated against the uProtocol specification before processing.
311    //
312    // # Returns
313    //
314    // Returns a CloudEvent protobuf with all information from the uProtocol message mapped as defined by the
315    // [uProtocol specification]().
316    //
317    // # Errors
318    //
319    // Returns an error if the given message does not contain the necessary information for creating a CloudEvent.
320    fn try_from(message: UMessage) -> Result<Self, Self::Error> {
321        // [impl->dsn~up-cloudevents-open-identity-unrepresentable~1]
322        // This mapping revision defines carriage for `payload_format`
323        // (registry ids 1..=8) only. A message carrying the open
324        // payload-encoding identity fields has no CloudEvents
325        // representation here and MUST fail the mapping explicitly: the
326        // identity is never dropped, substituted into `pformat`, or
327        // emitted as an unlabeled payload.
328        {
329            let attribs = message.attributes();
330            if attribs.open_payload_encoding_parts() != (None, None, None) {
331                return Err(UMessageError::PayloadError(
332                    "open payload-encoding identity fields are not representable in the CloudEvents mapping"
333                        .to_string(),
334                ));
335            }
336        }
337        let mut event = CloudEvent::new();
338        event.spec_version = CLOUDEVENTS_SPEC_VERSION.into();
339        event.set_id(message.id());
340        event.set_type(message.type_());
341        event.set_source(message.source());
342        if let Some(sink) = message.sink() {
343            event.set_sink(sink);
344        }
345        if let Some(priority) = message.priority() {
346            event.set_priority(priority);
347        }
348        if let Some(ttl) = message.ttl() {
349            event.set_ttl(ttl)?;
350        }
351        if let Some(token) = message.token() {
352            event.set_token(token);
353        }
354        if let Some(plevel) = message.permission_level() {
355            event.set_permission_level(plevel)?;
356        }
357        if let Some(reqid) = message.request_id() {
358            event.set_request_id(reqid);
359        }
360        if let Some(commstatus) = message.commstatus() {
361            event.set_commstatus(commstatus);
362        }
363        if let Some(traceparent) = message.traceparent() {
364            event.set_traceparent(traceparent);
365        }
366        let payload_format = message
367            .payload_format()
368            .unwrap_or(UPayloadFormat::Unspecified);
369        if let Some(payload) = message.payload() {
370            event.set_payload_format(payload_format);
371            match payload_format {
372                UPayloadFormat::Protobuf | UPayloadFormat::ProtobufWrappedInAny => {
373                    let data = Any {
374                        value: payload.to_vec(),
375                        ..Default::default()
376                    };
377                    event.set_proto_data(data);
378                }
379                UPayloadFormat::Text | UPayloadFormat::Json => {
380                    let data = String::from_utf8(payload.to_vec())
381                        .map(|v| v.to_string())
382                        .map_err(|_e| {
383                            UMessageError::PayloadError(
384                                "failed to transform payload to string".to_string(),
385                            )
386                        })?;
387                    event.set_text_data(data);
388                }
389                UPayloadFormat::Unspecified
390                | UPayloadFormat::Raw
391                | UPayloadFormat::Shm
392                | UPayloadFormat::Someip
393                | UPayloadFormat::SomeipTlv => {
394                    event.set_binary_data(payload.to_vec());
395                }
396            }
397        }
398        Ok(event)
399    }
400}
401
402impl TryFrom<CloudEvent> for UMessage {
403    type Error = UMessageError;
404
405    // Converts a CloudEvent to a uProtocol message.
406    //
407    // # Arguments
408    //
409    // * `event` - The CloudEvent to create the message from.
410    //
411    // # Errors
412    //
413    // Returns an error if the given event does not contain the necessary information for creating a uProtocol message.
414    // Also returns an error if the resulting message is not a valid uProtocol message.
415    fn try_from(event: CloudEvent) -> Result<Self, Self::Error> {
416        if !CLOUDEVENTS_SPEC_VERSION.eq(&event.spec_version) {
417            let msg = format!("expected spec version 1.0 but found {}", event.spec_version);
418            return Err(UMessageError::AttributesValidationError(
419                UAttributesError::ValidationError(msg),
420            ));
421        }
422
423        let attributes = UAttributes {
424            commstatus: event.get_commstatus()?,
425            id: event.get_id()?,
426            type_: event.get_type()?,
427            source: event.get_source()?,
428            sink: event.get_sink()?,
429            priority: Some(event.get_priority()?),
430            ttl: event.get_ttl()?,
431            permission_level: event.get_permission_level()?,
432            reqid: event.get_request_id()?,
433            token: event.get_token()?,
434            traceparent: event.get_traceparent()?,
435            payload_format: Some(event.get_payload_format()?),
436            payload_encoding_registry_id: None,
437            payload_encoding: None,
438            payload_content_type: None,
439        };
440        UAttributesValidators::validator_for_attributes(&attributes).validate(&attributes)?;
441
442        let payload = if event.has_binary_data() {
443            Some(Bytes::copy_from_slice(event.binary_data()))
444        } else if event.has_text_data() {
445            Some(event.text_data().to_owned().into())
446        } else if event.has_proto_data() {
447            Some(event.proto_data().value.to_vec().into())
448        } else {
449            None
450        };
451
452        UMessage::new(attributes, payload)
453    }
454}
455
456#[cfg(test)]
457mod tests {
458    use std::str::FromStr;
459
460    use cloudevents::CloudEvent;
461    use protobuf::{well_known_types::wrappers::StringValue, Message};
462
463    use crate::UMessageBuilder;
464
465    use super::*;
466
467    const MESSAGE_ID: &str = "00000000-0001-7000-8010-101010101a1a";
468    const TOPIC: &str = "//my-vehicle/A81B/1/A9BA";
469    const METHOD: &str = "//my-vehicle/A000/2/1";
470    const REPLY_TO: &str = "//my-vehicle/A81B/1/0";
471    const DESTINATION: &str = "//my-vehicle/A000/2/0";
472    const PERMISSION_LEVEL: u32 = 5;
473    const PRIORITY: UPriority = UPriority::CS4;
474    const TTL: u32 = 15_000;
475    const TRACEPARENT: &str = "traceparent";
476    const DATA: [u8; 4] = [0x00, 0x01, 0x02, 0x03];
477
478    //
479    // tests asserting conversion of UMessage -> CloudEvent
480    // [utest->dsn~cloudevents-umessage-mapping~2]
481    //
482
483    fn assert_standard_cloudevent_attributes(
484        event: &CloudEvent,
485        message_type: &str,
486        source: &str,
487        sink: Option<String>,
488    ) {
489        assert_eq!(event.spec_version, CLOUDEVENTS_SPEC_VERSION);
490        assert_eq!(event.type_, message_type);
491        assert_eq!(event.id, MESSAGE_ID);
492        assert_eq!(event.source.as_str(), source);
493        assert_eq!(
494            event
495                .attributes
496                .get(EXTENSION_NAME_SINK)
497                .map(|v| v.ce_uri_ref().to_owned()),
498            sink
499        );
500        assert_eq!(
501            event
502                .attributes
503                .get(EXTENSION_NAME_PRIORITY)
504                .map(|v| v.ce_string()),
505            Some(PRIORITY.to_priority_code())
506        );
507        assert_eq!(
508            event
509                .attributes
510                .get(EXTENSION_NAME_TTL)
511                .map(|v| v.ce_integer() as u32),
512            Some(TTL),
513            "unexpected TTL"
514        );
515        assert_eq!(
516            event
517                .attributes
518                .get(EXTENSION_NAME_TRACEPARENT)
519                .map(|v| v.ce_string()),
520            Some(TRACEPARENT)
521        );
522    }
523
524    #[test]
525    fn test_try_from_publish_message_succeeds() {
526        let message_id = MESSAGE_ID
527            .parse::<UUID>()
528            .expect("failed to parse message ID");
529        let message =
530            UMessageBuilder::publish(UUri::from_str(TOPIC).expect("failed to create topic URI"))
531                .with_message_id(message_id)
532                .with_priority(PRIORITY)
533                .with_ttl(TTL)
534                .with_traceparent(TRACEPARENT)
535                .build_with_payload("test".as_bytes(), UPayloadFormat::Text)
536                .expect("failed to create message");
537
538        let event =
539            CloudEvent::try_from(message).expect("failed to create CloudEvent from UMessage");
540        assert_standard_cloudevent_attributes(&event, "up-pub.v1", TOPIC, None);
541        assert_eq!(
542            event
543                .attributes
544                .get(EXTENSION_NAME_PFORMAT)
545                .map(|v| v.ce_integer()),
546            Some(UPayloadFormat::Text.as_i32())
547        );
548        assert_eq!(event.text_data(), "test");
549    }
550
551    #[test]
552    fn test_try_from_notification_message_succeeds() {
553        let message_id = MESSAGE_ID
554            .parse::<UUID>()
555            .expect("failed to parse message ID");
556        let message = UMessageBuilder::notification(
557            UUri::from_str(TOPIC).expect("failed to create source URI"),
558            UUri::from_str(DESTINATION).expect("failed to create sink URI"),
559        )
560        .with_message_id(message_id)
561        .with_priority(PRIORITY)
562        .with_ttl(TTL)
563        .with_traceparent(TRACEPARENT)
564        .build_with_payload("{\"count\": 5}".as_bytes(), UPayloadFormat::Json)
565        .expect("failed to create message");
566
567        let event =
568            CloudEvent::try_from(message).expect("failed to create CloudEvent from UMessage");
569        assert_standard_cloudevent_attributes(
570            &event,
571            "up-not.v1",
572            TOPIC,
573            Some(DESTINATION.to_string()),
574        );
575        assert_eq!(
576            event
577                .attributes
578                .get(EXTENSION_NAME_PFORMAT)
579                .map(|v| v.ce_integer()),
580            Some(UPayloadFormat::Json.as_i32())
581        );
582        assert_eq!(event.text_data(), "{\"count\": 5}");
583    }
584
585    #[test]
586    fn test_try_from_request_message_succeeds() {
587        let payload = b"Hello";
588        let message_id = MESSAGE_ID
589            .parse::<UUID>()
590            .expect("failed to parse message ID");
591        let token = "my-token";
592        let message = UMessageBuilder::request(
593            UUri::from_str(METHOD).expect("failed to create sink URI"),
594            UUri::from_str(REPLY_TO).expect("failed to create source URI"),
595            TTL,
596        )
597        .with_message_id(message_id)
598        .with_priority(PRIORITY)
599        .with_permission_level(PERMISSION_LEVEL)
600        .with_traceparent(TRACEPARENT)
601        .with_token(token)
602        .build_with_payload(payload.as_slice(), UPayloadFormat::Raw)
603        .expect("failed to create message");
604        let event =
605            CloudEvent::try_from(message).expect("failed to create CloudEvent from UMessage");
606        assert_standard_cloudevent_attributes(
607            &event,
608            "up-req.v1",
609            REPLY_TO,
610            Some(METHOD.to_string()),
611        );
612        assert_eq!(
613            event
614                .attributes
615                .get(EXTENSION_NAME_TOKEN)
616                .map(|v| v.ce_string()),
617            Some(token)
618        );
619        assert_eq!(
620            event
621                .attributes
622                .get(EXTENSION_NAME_PERMISSION_LEVEL)
623                .map(|v| v.ce_integer()),
624            Some(PERMISSION_LEVEL as i32)
625        );
626        assert_eq!(
627            event
628                .attributes
629                .get(EXTENSION_NAME_PFORMAT)
630                .map(|v| v.ce_integer()),
631            Some(UPayloadFormat::Raw.as_i32())
632        );
633        assert!(!event.has_proto_data());
634        assert!(!event.has_text_data());
635        assert_eq!(event.binary_data(), payload);
636    }
637
638    #[test]
639    fn test_try_from_response_message_succeeds() {
640        let mut payload = StringValue::new();
641        payload.value = "Hello".into();
642
643        let message_id = MESSAGE_ID
644            .parse::<UUID>()
645            .expect("failed to parse message ID");
646        let request_id = UUID::build();
647
648        let message = UMessageBuilder::response(
649            UUri::from_str(REPLY_TO).expect("failed to create sink URI"),
650            request_id.clone(),
651            UUri::from_str(METHOD).expect("failed to create source URI"),
652        )
653        .with_message_id(message_id)
654        .with_ttl(TTL)
655        .with_priority(PRIORITY)
656        .with_comm_status(UCode::Ok)
657        .with_traceparent(TRACEPARENT)
658        .build_with_protobuf_payload(&payload)
659        .expect("failed to create message");
660
661        let event =
662            CloudEvent::try_from(message).expect("failed to create CloudEvent from UMessage");
663        assert_standard_cloudevent_attributes(
664            &event,
665            "up-res.v1",
666            METHOD,
667            Some(REPLY_TO.to_string()),
668        );
669        assert_eq!(
670            event
671                .attributes
672                .get(EXTENSION_NAME_COMMSTATUS)
673                .map(|v| v.ce_integer()),
674            None
675        );
676        assert_eq!(
677            event
678                .attributes
679                .get(EXTENSION_NAME_PFORMAT)
680                .map(|v| v.ce_integer()),
681            Some(UPayloadFormat::Protobuf.as_i32())
682        );
683        assert!(!event.has_binary_data());
684        assert!(!event.has_text_data());
685        assert_eq!(
686            event.proto_data().value,
687            Any::pack(&payload)
688                .expect("failed to pack payload into Any")
689                .value
690        );
691    }
692
693    //
694    // tests asserting conversion of CloudEvent -> UMessage
695    // [utest->dsn~cloudevents-umessage-mapping~2]
696    //
697
698    fn assert_standard_umessage_attributes(
699        attribs: &UAttributes,
700        message_type: UMessageType,
701        source: &str,
702        sink: Option<String>,
703    ) {
704        assert_eq!(attribs.type_(), message_type);
705        assert_eq!(attribs.id().to_hyphenated_string(), MESSAGE_ID);
706        assert_eq!(attribs.source().to_uri(false), source);
707        assert_eq!(attribs.sink.as_ref().map(|uuri| uuri.to_uri(false)), sink);
708        assert_eq!(attribs.priority_unchecked(), UPriority::CS4);
709        assert_eq!(attribs.ttl, Some(TTL));
710        assert_eq!(attribs.traceparent(), Some(TRACEPARENT));
711    }
712
713    #[test]
714    fn test_try_from_cloudevent_without_sink_fails() {
715        let mut event = CloudEvent::new();
716        event.spec_version = CLOUDEVENTS_SPEC_VERSION.into();
717        event.type_ = UMessageType::Notification.to_cloudevent_type();
718        event.id = MESSAGE_ID.into();
719        event.source = TOPIC.into();
720
721        assert!(UMessage::try_from(event).is_err());
722    }
723
724    #[test]
725    fn test_try_from_publish_cloudevent_succeeds() {
726        let mut event = CloudEvent::new();
727        event.spec_version = CLOUDEVENTS_SPEC_VERSION.into();
728        event.set_type(UMessageType::Publish);
729        event.id = MESSAGE_ID.into();
730        event.source = TOPIC.into();
731        event.set_priority(UPriority::CS4);
732        event.set_ttl(TTL).expect("failed to set TTL on message");
733        event.set_traceparent(TRACEPARENT);
734        event.set_payload_format(UPayloadFormat::Text);
735        event.set_text_data("test".to_string());
736
737        let umessage =
738            UMessage::try_from(event).expect("failed to create UMessage from CloudEvent");
739        let attribs = umessage.attributes();
740        assert_standard_umessage_attributes(attribs, UMessageType::Publish, TOPIC, None);
741        assert_eq!(attribs.payload_format_unchecked(), UPayloadFormat::Text);
742        assert_eq!(umessage.payload(), Some("test".as_bytes().into()))
743    }
744
745    #[test]
746    fn test_try_from_notification_cloudevent_succeeds() {
747        let mut event = CloudEvent::new();
748        event.spec_version = CLOUDEVENTS_SPEC_VERSION.into();
749        event.set_type(UMessageType::Notification);
750        event.id = MESSAGE_ID.into();
751        event.source = TOPIC.into();
752        event.set_sink(DESTINATION);
753        event.set_priority(UPriority::CS4);
754        event.set_ttl(TTL).expect("failed to set TTL on message");
755        event.set_traceparent(TRACEPARENT);
756        event.set_payload_format(UPayloadFormat::Json);
757        event.set_text_data("{\"count\": 5}".to_string());
758
759        let umessage =
760            UMessage::try_from(event).expect("failed to create UMessage from CloudEvent");
761        let attribs = umessage.attributes();
762        assert_standard_umessage_attributes(
763            attribs,
764            UMessageType::Notification,
765            TOPIC,
766            Some(DESTINATION.to_string()),
767        );
768        assert_eq!(attribs.payload_format_unchecked(), UPayloadFormat::Json);
769        assert_eq!(umessage.payload(), Some("{\"count\": 5}".as_bytes().into()))
770    }
771
772    #[test]
773    fn test_try_from_request_cloudevent_succeeds() {
774        let mut event = CloudEvent::new();
775        event.spec_version = CLOUDEVENTS_SPEC_VERSION.into();
776        event.set_type(UMessageType::Request);
777        event.id = MESSAGE_ID.into();
778        event.source = REPLY_TO.into();
779        event.set_sink(METHOD);
780        event.set_priority(UPriority::CS4);
781        event.set_ttl(TTL).expect("failed to set TTL on message");
782        event.set_traceparent(TRACEPARENT);
783        event
784            .set_permission_level(PERMISSION_LEVEL)
785            .expect("failed to set permission level on message");
786        event.set_token("my-token");
787
788        let mut payload = StringValue::new();
789        payload.value = "Hello".into();
790        let payload_wrapped_in_any = Any::pack(&payload).expect("failed to wrap payload in Any");
791        let serialized_payload = payload_wrapped_in_any
792            .write_to_bytes()
793            .expect("failed to serialize payload");
794        event.set_proto_data(Any {
795            value: serialized_payload.clone(),
796            ..Default::default()
797        });
798
799        let umessage =
800            UMessage::try_from(event).expect("failed to create UMessage from CloudEvent");
801        let attribs = umessage.attributes();
802        assert_standard_umessage_attributes(
803            attribs,
804            UMessageType::Request,
805            REPLY_TO,
806            Some(METHOD.to_string()),
807        );
808        assert_eq!(attribs.permission_level, Some(PERMISSION_LEVEL));
809        assert_eq!(attribs.token(), Some("my-token"));
810        assert_eq!(
811            attribs.payload_format_unchecked(),
812            UPayloadFormat::Unspecified
813        );
814        assert_eq!(
815            umessage.payload(),
816            Some(Bytes::copy_from_slice(serialized_payload.as_slice()))
817        );
818    }
819
820    #[test]
821    fn test_try_from_response_cloudevent_succeeds() {
822        let request_id = UUID::build();
823        let mut event = CloudEvent::new();
824        event.spec_version = CLOUDEVENTS_SPEC_VERSION.into();
825        event.set_type(UMessageType::Response);
826        event.id = MESSAGE_ID.into();
827        event.source = METHOD.into();
828        event.set_sink(REPLY_TO);
829        event.set_priority(UPriority::CS4);
830        event.set_ttl(TTL).expect("failed to set TTL on message");
831        event.set_traceparent(TRACEPARENT);
832        event.set_request_id(&request_id);
833        event.set_commstatus(UCode::Ok);
834        event.set_payload_format(UPayloadFormat::Protobuf);
835        event.set_proto_data(Any {
836            value: DATA.to_vec(),
837            ..Default::default()
838        });
839
840        let umessage =
841            UMessage::try_from(event).expect("failed to create UMessage from CloudEvent");
842        let attribs = umessage.attributes();
843        assert_standard_umessage_attributes(
844            attribs,
845            UMessageType::Response,
846            METHOD,
847            Some(REPLY_TO.to_string()),
848        );
849        assert_eq!(attribs.commstatus, None);
850        assert_eq!(attribs.reqid, Some(request_id));
851        assert_eq!(attribs.payload_format_unchecked(), UPayloadFormat::Protobuf);
852        assert_eq!(umessage.payload(), Some(DATA.as_slice().into()));
853    }
854
855    /// [utest->dsn~up-cloudevents-open-identity-unrepresentable~1]
856    #[test]
857    fn umessage_with_open_payload_encoding_identity_fails_mapping() {
858        let encoding =
859            crate::PayloadEncoding::custom("up.xcdr-v2", "application/vnd.uprotocol.xcdr-v2")
860                .expect("encoding");
861        let message =
862            UMessageBuilder::publish(UUri::try_from_parts("vin", 0x10AB, 1, 0x8000).expect("uri"))
863                .build_with_payload(vec![0xC0, 0xFF], UPayloadFormat::Unspecified)
864                .expect("message")
865                .with_assumed_payload_encoding(&encoding);
866
867        let result = CloudEvent::try_from(message);
868        assert!(
869            matches!(result, Err(UMessageError::PayloadError(_))),
870            "open identity must fail the CE mapping explicitly"
871        );
872    }
873}
up_rust/cloudevents.rs

up_rust/
cloudevents.rs
