up_rust/
communication.rs

1/********************************************************************************
2 * Copyright (c) 2024 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
14/*!
15Traits representing uProtocol's [Communication Layer API](https://github.com/eclipse-uprotocol/up-spec/blob/v1.6.0-alpha.7/up-l2/api.adoc) for publishing and subscribing to topics and for invoking RPC methods. Also contains default implementations of the traits employing uProtocol's [Transport & Session Layer API](crate::UTransport).
16*/
17
18use bytes::Bytes;
19use thiserror::Error;
20
21use crate::{UCode, UPayloadFormat, UPriority, UStatus, UUID};
22
23mod notification;
24#[cfg(any(test, feature = "test-util"))]
25pub use notification::MockNotifier;
26pub use notification::{NotificationError, Notifier};
27
28#[cfg(feature = "up-l2-notifier")]
29mod simple_notifier;
30#[cfg(feature = "up-l2-notifier")]
31pub use simple_notifier::SimpleNotifier;
32
33mod pubsub;
34#[cfg(any(test, feature = "test-util"))]
35pub use pubsub::MockSubscriptionChangeHandler;
36pub use pubsub::{PubSubError, Publisher, Subscriber, SubscriptionChangeHandler};
37
38#[cfg(feature = "up-l2-publisher")]
39mod simple_publisher;
40#[cfg(feature = "up-l2-publisher")]
41pub use simple_publisher::SimplePublisher;
42
43#[cfg(feature = "up-l2-subscriber")]
44mod in_memory_subscriber;
45#[cfg(feature = "up-l2-subscriber")]
46pub use in_memory_subscriber::InMemorySubscriber;
47
48mod rpc;
49#[cfg(any(test, feature = "test-util"))]
50pub use rpc::{MockRequestHandler, MockRpcClient, MockRpcServerImpl};
51pub use rpc::{RequestHandler, RpcClient, RpcServer, ServiceInvocationError};
52
53#[cfg(feature = "up-l2-rpc-client")]
54mod in_memory_rpc_client;
55#[cfg(feature = "up-l2-rpc-client")]
56pub use in_memory_rpc_client::InMemoryRpcClient;
57
58#[cfg(feature = "up-l2-rpc-server")]
59mod in_memory_rpc_server;
60#[cfg(feature = "up-l2-rpc-server")]
61pub use in_memory_rpc_server::InMemoryRpcServer;
62
63/// Moves all common call options into the given message builder.
64///
65/// In particular, the following options are moved:
66/// * ttl
67/// * message ID
68/// * priority
69#[cfg(any(feature = "up-l2-notifier", feature = "up-l2-publisher"))]
70pub(crate) fn apply_common_options<S: crate::umessage::BuilderState>(
71    call_options: CallOptions,
72    message_builder: &mut crate::UMessageBuilder<S>,
73) {
74    message_builder.with_ttl(call_options.ttl);
75    if let Some(v) = call_options.message_id {
76        message_builder.with_message_id(v);
77    }
78    if let Some(v) = call_options.priority {
79        message_builder.with_priority(v);
80    }
81}
82
83/// Creates a message with given payload from a builder.
84#[cfg(any(
85    feature = "up-l2-notifier",
86    feature = "up-l2-publisher",
87    feature = "up-l2-rpc-client",
88    feature = "up-l2-rpc-server"
89))]
90pub(crate) fn build_message<S: crate::umessage::BuilderState>(
91    message_builder: &mut crate::UMessageBuilder<S>,
92    payload: Option<UPayload>,
93) -> Result<crate::UMessage, crate::UMessageError> {
94    if let Some(pl) = payload {
95        let format = pl.payload_format();
96        message_builder.build_with_payload(pl.payload, format)
97    } else {
98        message_builder.build()
99    }
100}
101
102/// The current status of a client's subscription to a topic.
103///
104/// The status goes through different stages during its lifecycle as defined in
105/// [uProtocol Specification, section 3.3.5](https://github.com/eclipse-uprotocol/up-spec/blob/v1.6.0-alpha.7/up-l3/usubscription/v3/README.adoc#usubscription-states).
106#[derive(Clone, Debug, PartialEq)]
107#[repr(C)]
108pub enum SubscriptionStatus {
109    /// Not subscribed.
110    Unsubscribed,
111    /// Subscription requested; confirmation pending.
112    SubscribePending,
113    /// Actively subscribed.
114    Subscribed,
115    /// Unsubscription requested; confirmation pending.
116    UnsubscribePending,
117}
118
119#[cfg(all(feature = "up-core-api", feature = "usubscription"))]
120mod core_types_support {
121    use super::{SubscriptionStatus, UCode, UStatus};
122    use crate::up_core_api::usubscription::subscription_status::State;
123    use crate::up_core_api::usubscription::SubscriptionStatus as SubscriptionStatusProto;
124
125    impl TryFrom<&SubscriptionStatusProto> for SubscriptionStatus {
126        type Error = UStatus;
127
128        fn try_from(status_proto: &SubscriptionStatusProto) -> Result<Self, Self::Error> {
129            let state = status_proto.state.enum_value();
130            match state {
131                Ok(State::UNSUBSCRIBED) => Ok(SubscriptionStatus::Unsubscribed),
132                Ok(State::SUBSCRIBE_PENDING) => Ok(SubscriptionStatus::SubscribePending),
133                Ok(State::SUBSCRIBED) => Ok(SubscriptionStatus::Subscribed),
134                Ok(State::UNSUBSCRIBE_PENDING) => Ok(SubscriptionStatus::UnsubscribePending),
135                Err(v) => Err(UStatus::fail_with_code(
136                    UCode::InvalidArgument,
137                    format!("unknown subscription status {:?}", v),
138                )),
139            }
140        }
141    }
142}
143
144/// An error indicating a problem with registering or unregistering a message listener.
145#[derive(Clone, Debug, Error)]
146pub enum RegistrationError {
147    /// Indicates that a listener for a given address already exists.
148    #[error("a listener for the given filter criteria already exists")]
149    AlreadyExists,
150    /// Indicates that the maximum number of listeners supported by the Transport Layer implementation
151    /// has already been registered.
152    #[error("maximum number of listeners has been reached")]
153    MaxListenersExceeded,
154    /// Indicates that no listener is registered for given pattern URIs.
155    #[error("no listener registered for given pattern")]
156    NoSuchListener,
157    /// Indicates that the underlying Transport Layer implementation does not support registration and
158    /// notification of message handlers.
159    #[error("the underlying transport implementation does not support the push delivery method")]
160    PushDeliveryMethodNotSupported,
161    /// Indicates that some of the given filters are inappropriate in this context.
162    #[error("invalid filter(s): {0}")]
163    InvalidFilter(String),
164    /// Indicates a generic error.
165    #[error("error un-/registering listener: {0}")]
166    Unknown(Box<UStatus>),
167}
168
169impl From<UStatus> for RegistrationError {
170    fn from(value: UStatus) -> Self {
171        match value.code() {
172            UCode::AlreadyExists => RegistrationError::AlreadyExists,
173            UCode::NotFound => RegistrationError::NoSuchListener,
174            UCode::ResourceExhausted => RegistrationError::MaxListenersExceeded,
175            UCode::Unimplemented => RegistrationError::PushDeliveryMethodNotSupported,
176            UCode::InvalidArgument => {
177                RegistrationError::InvalidFilter(value.message().unwrap_or("N/A").to_string())
178            }
179            UCode::Ok
180            | UCode::Cancelled
181            | UCode::Unknown
182            | UCode::DeadlineExceeded
183            | UCode::PermissionDenied
184            | UCode::Unauthenticated
185            | UCode::FailedPrecondition
186            | UCode::Aborted
187            | UCode::OutOfRange
188            | UCode::Internal
189            | UCode::Unavailable
190            | UCode::DataLoss => RegistrationError::Unknown(Box::from(value)),
191        }
192    }
193}
194
195/// General options that clients might want to specify when sending a uProtocol message.
196#[must_use = "CallOptions should be used when sending messages to specify message parameters"]
197#[derive(Clone, Debug, PartialEq)]
198pub struct CallOptions {
199    ttl: u32,
200    message_id: Option<UUID>,
201    token: Option<String>,
202    priority: Option<UPriority>,
203}
204
205impl CallOptions {
206    /// Creates new call options for an RPC Request.
207    ///
208    /// # Arguments
209    ///
210    /// * `ttl` - The message's time-to-live in milliseconds.
211    /// * `message_id` - The identifier to use for the message or `None` to use a generated identifier.
212    /// * `token` - The token to use for authenticating to infrastructure and service endpoints.
213    /// * `priority` - The message's priority or `None` to use the default priority for RPC Requests.
214    ///
215    /// # Returns
216    ///
217    /// Options suitable for invoking an RPC method.
218    ///
219    /// # Examples
220    ///
221    /// ```rust
222    /// use up_rust::{UPriority, UUID, communication::CallOptions};
223    ///
224    /// let uuid = UUID::build();
225    /// let token = String::from("token");
226    /// let options = CallOptions::for_rpc_request(15_000, Some(uuid.clone()), Some(token.clone()), Some(UPriority::CS6));
227    /// assert_eq!(options.ttl(), 15_000);
228    /// assert_eq!(options.message_id(), Some(&uuid));
229    /// assert_eq!(options.token(), Some(&token));
230    /// assert_eq!(options.priority(), Some(UPriority::CS6));
231    /// ```
232    pub fn for_rpc_request(
233        ttl: u32,
234        message_id: Option<UUID>,
235        token: Option<String>,
236        priority: Option<UPriority>,
237    ) -> Self {
238        CallOptions {
239            ttl,
240            message_id,
241            token,
242            priority,
243        }
244    }
245
246    /// Creates new call options for a Notification message.
247    ///
248    /// # Arguments
249    ///
250    /// * `ttl` - The message's time-to-live in milliseconds.
251    /// * `message_id` - The identifier to use for the message or `None` to use a generated identifier.
252    /// * `priority` - The message's priority or `None` to use the default priority for Notifications.
253    ///
254    /// # Returns
255    ///
256    /// Options suitable for sending a Notification.
257    ///
258    /// # Examples
259    ///
260    /// ```rust
261    /// use up_rust::{UPriority, UUID, communication::CallOptions};
262    ///
263    /// let uuid = UUID::build();
264    /// let options = CallOptions::for_notification(Some(15_000), Some(uuid.clone()), Some(UPriority::CS2));
265    /// assert_eq!(options.ttl(), 15_000);
266    /// assert_eq!(options.message_id(), Some(&uuid));
267    /// assert_eq!(options.priority(), Some(UPriority::CS2));
268    /// ```
269    pub fn for_notification(
270        ttl: Option<u32>,
271        message_id: Option<UUID>,
272        priority: Option<UPriority>,
273    ) -> Self {
274        CallOptions {
275            ttl: ttl.unwrap_or(0),
276            message_id,
277            token: None,
278            priority,
279        }
280    }
281
282    /// Creates new call options for a Publish message.
283    ///
284    /// # Arguments
285    ///
286    /// * `ttl` - The message's time-to-live in milliseconds or `None` if the message should not expire at all.
287    /// * `message_id` - The identifier to use for the message or `None` to use a generated identifier.
288    /// * `priority` - The message's priority or `None` to use the default priority for Publish messages.
289    ///
290    /// # Returns
291    ///
292    /// Options suitable for sending a Publish message.
293    ///
294    /// # Examples
295    ///
296    /// ```rust
297    /// use up_rust::{UPriority, UUID, communication::CallOptions};
298    ///
299    /// let uuid = UUID::build();
300    /// let options = CallOptions::for_publish(Some(15_000), Some(uuid.clone()), Some(UPriority::CS2));
301    /// assert_eq!(options.ttl(), 15_000);
302    /// assert_eq!(options.message_id(), Some(&uuid));
303    /// assert_eq!(options.priority(), Some(UPriority::CS2));
304    /// ```
305    pub fn for_publish(
306        ttl: Option<u32>,
307        message_id: Option<UUID>,
308        priority: Option<UPriority>,
309    ) -> Self {
310        CallOptions {
311            ttl: ttl.unwrap_or(0),
312            message_id,
313            token: None,
314            priority,
315        }
316    }
317
318    /// Gets the message's time-to-live in milliseconds.
319    pub fn ttl(&self) -> u32 {
320        self.ttl
321    }
322
323    /// Gets the identifier to use for the message.
324    pub fn message_id(&self) -> Option<&UUID> {
325        self.message_id.as_ref()
326    }
327
328    /// Gets the token to use for authenticating to infrastructure and service endpoints.
329    pub fn token(&self) -> Option<&String> {
330        self.token.as_ref()
331    }
332
333    /// Gets the message's priority.
334    pub fn priority(&self) -> Option<UPriority> {
335        self.priority
336    }
337}
338
339/// A wrapper around (raw) message payload data and the corresponding payload format.
340#[derive(Clone, Debug, PartialEq)]
341pub struct UPayload {
342    payload_format: UPayloadFormat,
343    payload: Bytes,
344}
345
346impl UPayload {
347    /// Creates a new payload for some data.
348    ///
349    /// # Examples
350    ///
351    /// ```rust
352    /// use up_rust::UPayloadFormat;
353    /// use up_rust::communication::UPayload;
354    ///
355    /// let data: Vec<u8> = vec![0x00_u8, 0x01_u8, 0x02_u8];
356    /// let payload = UPayload::new(data, UPayloadFormat::Raw);
357    /// assert_eq!(payload.payload_format(), UPayloadFormat::Raw);
358    /// assert_eq!(payload.payload().len(), 3);
359    /// ```
360    pub fn new<T: Into<Bytes>>(payload: T, payload_format: UPayloadFormat) -> Self {
361        UPayload {
362            payload_format,
363            payload: payload.into(),
364        }
365    }
366
367    /// Creates a new UPayload from an object that can be mapped to/from a protobuf.
368    ///
369    /// The resulting payload will have `UPayloadType::UPAYLOAD_FORMAT_PROTOBUF_WRAPPED_IN_ANY`.
370    ///
371    /// # Errors
372    ///
373    /// Returns an error if the given message cannot be serialized to bytes.
374    ///
375    /// # Examples
376    ///
377    /// ```rust
378    /// use up_rust::{communication::UPayload, UPayloadFormat};
379    /// use protobuf::{well_known_types::wrappers::StringValue};
380    ///
381    /// let mut data = StringValue::new();
382    /// data.value = "hello world".to_string();
383    /// assert!(UPayload::try_from_protobuf(data).is_ok_and(|pl|
384    ///     pl.payload_format() == UPayloadFormat::ProtobufWrappedInAny
385    ///         && pl.payload().len() > 0));
386    /// ```
387    #[cfg(feature = "protobuf-support")]
388    pub fn try_from_protobuf<T>(obj: T) -> Result<Self, crate::UMessageError>
389    where
390        T: crate::ProtobufMappable,
391    {
392        obj.write_to_packed_protobuf_bytes()
393            .map(|buf| UPayload::new(buf, UPayloadFormat::ProtobufWrappedInAny))
394            .map_err(crate::UMessageError::from)
395    }
396
397    /// Gets the payload format.
398    ///
399    /// # Returns
400    ///
401    /// payload value of `UPayload`.
402    pub fn payload_format(&self) -> UPayloadFormat {
403        self.payload_format
404    }
405
406    /// Gets the payload data.
407    #[must_use]
408    pub fn payload(&self) -> &Bytes {
409        &self.payload
410    }
411
412    /// Extracts the object that is contained in this message's payload.
413    ///
414    /// # Type Parameters
415    ///
416    /// * `T`: The target type of the data to be unpacked.
417    ///
418    /// # Returns
419    ///
420    /// * The deserialized object contained in the payload.
421    ///
422    /// # Errors
423    ///
424    /// * Returns an error if the unpacking process fails, for example if the payload format
425    ///   is neither [`UPayloadFormat::Protobuf`] nor [`UPayloadFormat::ProtobufWrappedInAny`],
426    ///   or if the payload could not be deserialized into the target type `T`.
427    ///
428    /// # Examples
429    ///
430    /// ```rust
431    /// use up_rust::{communication::UPayload, UPayloadFormat};
432    /// use protobuf::{well_known_types::wrappers::StringValue};
433    ///
434    /// let mut data = StringValue::new();
435    /// data.value = "hello world".to_string();
436    /// let payload = UPayload::try_from_protobuf(data).expect("should be able to create UPayload from StringValue");
437    ///
438    /// let string_value: StringValue = payload.extract_protobuf().expect("should be able to extract StringValue from UPayload");
439    /// assert_eq!(string_value.value, *"hello world");
440    /// ```
441    #[cfg(feature = "protobuf-support")]
442    pub fn extract_protobuf<T>(&self) -> Result<T, crate::UMessageError>
443    where
444        T: crate::ProtobufMappable + Default,
445    {
446        crate::umessage::deserialize_protobuf_bytes(&self.payload, &self.payload_format)
447    }
448}
449
450#[cfg(any(
451    all(
452        feature = "owned-frame-transport",
453        feature = "usubscription",
454        feature = "selected-wire-transport-core"
455    ),
456    all(
457        feature = "selected-wire-transport-adapter",
458        feature = "communication-api",
459        feature = "zero-copy-transport",
460        feature = "usubscription"
461    )
462))]
463pub(crate) fn validate_listener_topic(topic: &crate::UUri) -> Result<(), RegistrationError> {
464    topic
465        .verify_no_wildcards()
466        .map_err(|error| RegistrationError::InvalidFilter(error.to_string()))
467}
468
469#[cfg(all(
470    feature = "owned-frame-transport",
471    feature = "usubscription",
472    feature = "selected-wire-transport-core"
473))]
474pub mod owned;
475#[cfg(all(
476    feature = "selected-wire-transport-adapter",
477    feature = "communication-api",
478    feature = "zero-copy-transport"
479))]
480pub mod zero_copy;
up_rust/communication.rs

up_rust/
communication.rs
