up_rust/communication/
rpc.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
14use std::sync::Arc;
15
16use async_trait::async_trait;
17use thiserror::Error;
18
19use crate::{UAttributes, UCode, UStatus, UUri};
20
21use super::{CallOptions, RegistrationError, UPayload};
22
23/// An error indicating a problem with invoking a (remote) service operation.
24// [impl->dsn~communication-layer-api-declaration~1]
25#[derive(Clone, Error, Debug)]
26#[non_exhaustive]
27pub enum ServiceInvocationError {
28    /// Indicates that the calling uE requested to add/create something that already exists.
29    #[error("entity already exists: {0}")]
30    AlreadyExists(String),
31    /// Indicates that a request's time-to-live (TTL) has expired.
32    ///
33    /// Note that this only means that the reply to the request has not been received in time. The request
34    /// may still have been processed by the (remote) service provider.
35    #[error("request timed out")]
36    DeadlineExceeded,
37    /// Indicates that the service provider is in a state that prevents it from handling the request.
38    #[error("failed precondition: {0}")]
39    FailedPrecondition(String),
40    /// Indicates that a serious but unspecified internal error has occurred while sending/processing the request.
41    #[error("internal error: {0}")]
42    Internal(String),
43    /// Indicates that the request cannot be processed because some of its parameters are invalid, e.g. not properly formatted.
44    #[error("invalid argument: {0}")]
45    InvalidArgument(String),
46    /// Indicates that the requested entity was not found.
47    #[error("no such entity: {0}")]
48    NotFound(String),
49    /// Indicates that the calling uE is authenticated but does not have the required authority to invoke the method.
50    #[error("permission denied: {0}")]
51    PermissionDenied(String),
52    /// Indicates that some of the resources required for processing the request have been exhausted, e.g. disk space, number of API calls.
53    #[error("resource exhausted: {0}")]
54    ResourceExhausted(String),
55    /// Indicates an unspecific error that occurred at the Transport Layer while trying to publish a message.
56    #[error("unknown error: {0}")]
57    RpcError(Box<UStatus>),
58    /// Indicates that the calling uE could not be authenticated properly.
59    #[error("unauthenticated")]
60    Unauthenticated,
61    /// Indicates that some of the resources required for processing the request are currently unavailable.
62    #[error("resource unavailable: {0}")]
63    Unavailable(String),
64    /// Indicates that part or all of the invoked operation has not been implemented yet.
65    #[error("unimplemented: {0}")]
66    Unimplemented(String),
67}
68
69impl From<UStatus> for ServiceInvocationError {
70    fn from(value: UStatus) -> Self {
71        match value.code() {
72            UCode::AlreadyExists => {
73                ServiceInvocationError::AlreadyExists(value.message_or_default("N/A").to_string())
74            }
75            UCode::DeadlineExceeded => ServiceInvocationError::DeadlineExceeded,
76            UCode::FailedPrecondition => ServiceInvocationError::FailedPrecondition(
77                value.message_or_default("N/A").to_string(),
78            ),
79            UCode::Internal => {
80                ServiceInvocationError::Internal(value.message_or_default("N/A").to_string())
81            }
82            UCode::InvalidArgument => {
83                ServiceInvocationError::InvalidArgument(value.message_or_default("N/A").to_string())
84            }
85            UCode::NotFound => {
86                ServiceInvocationError::NotFound(value.message_or_default("N/A").to_string())
87            }
88            UCode::PermissionDenied => ServiceInvocationError::PermissionDenied(
89                value.message_or_default("N/A").to_string(),
90            ),
91            UCode::ResourceExhausted => ServiceInvocationError::ResourceExhausted(
92                value.message_or_default("N/A").to_string(),
93            ),
94            UCode::Unauthenticated => ServiceInvocationError::Unauthenticated,
95            UCode::Unavailable => {
96                ServiceInvocationError::Unavailable(value.message_or_default("N/A").to_string())
97            }
98            UCode::Unimplemented => {
99                ServiceInvocationError::Unimplemented(value.message_or_default("N/A").to_string())
100            }
101            UCode::Ok
102            | UCode::Cancelled
103            | UCode::Unknown
104            | UCode::Aborted
105            | UCode::OutOfRange
106            | UCode::DataLoss => ServiceInvocationError::RpcError(Box::from(value)),
107        }
108    }
109}
110
111impl From<ServiceInvocationError> for UStatus {
112    fn from(value: ServiceInvocationError) -> Self {
113        match value {
114            ServiceInvocationError::AlreadyExists(msg) => {
115                UStatus::fail_with_code(UCode::AlreadyExists, msg)
116            }
117            ServiceInvocationError::DeadlineExceeded => {
118                UStatus::fail_with_code(UCode::DeadlineExceeded, "request timed out")
119            }
120            ServiceInvocationError::FailedPrecondition(msg) => {
121                UStatus::fail_with_code(UCode::FailedPrecondition, msg)
122            }
123            ServiceInvocationError::Internal(msg) => UStatus::fail_with_code(UCode::Internal, msg),
124            ServiceInvocationError::InvalidArgument(msg) => {
125                UStatus::fail_with_code(UCode::InvalidArgument, msg)
126            }
127            ServiceInvocationError::NotFound(msg) => UStatus::fail_with_code(UCode::NotFound, msg),
128            ServiceInvocationError::PermissionDenied(msg) => {
129                UStatus::fail_with_code(UCode::PermissionDenied, msg)
130            }
131            ServiceInvocationError::ResourceExhausted(msg) => {
132                UStatus::fail_with_code(UCode::ResourceExhausted, msg)
133            }
134            ServiceInvocationError::Unauthenticated => {
135                UStatus::fail_with_code(UCode::Unauthenticated, "client must authenticate")
136            }
137            ServiceInvocationError::Unavailable(msg) => {
138                UStatus::fail_with_code(UCode::Unavailable, msg)
139            }
140            ServiceInvocationError::Unimplemented(msg) => {
141                UStatus::fail_with_code(UCode::Unimplemented, msg)
142            }
143            ServiceInvocationError::RpcError(status) => *status,
144        }
145    }
146}
147
148/// A client for performing Remote Procedure Calls (RPC) on (other) uEntities.
149///
150/// Please refer to the
151/// [Communication Layer API specification](https://github.com/eclipse-uprotocol/up-spec/blob/main/up-l2/api.adoc)
152/// for details.
153// [impl->dsn~communication-layer-api-declaration~1]
154#[cfg_attr(any(test, feature = "test-util"), mockall::automock)]
155#[async_trait]
156pub trait RpcClient: Send + Sync {
157    /// Invokes a method on a service.
158    ///
159    /// # Arguments
160    ///
161    /// * `method` - The URI representing the method to invoke.
162    /// * `call_options` - Options to include in the request message.
163    /// * `payload` - The (optional) payload to include in the request message.
164    ///
165    /// # Returns
166    ///
167    /// The payload returned by the service operation.
168    ///
169    /// # Errors
170    ///
171    /// Returns an error if invocation fails or the given arguments cannot be turned into a valid RPC Request message.
172    async fn invoke_method(
173        &self,
174        method: UUri,
175        call_options: CallOptions,
176        payload: Option<UPayload>,
177    ) -> Result<Option<UPayload>, ServiceInvocationError>;
178}
179
180#[cfg(feature = "protobuf-support")]
181impl dyn RpcClient {
182    /// Invokes a method on a service using and returning proto-generated `Message` objects.
183    ///
184    /// # Arguments
185    ///
186    /// * `method` - The URI representing the method to invoke.
187    /// * `call_options` - Options to include in the request message.
188    /// * `request_message` - The protobuf `Message` to include in the request message.
189    ///
190    /// # Returns
191    ///
192    /// The payload returned by the service operation as a protobuf `Message`.
193    ///
194    /// # Errors
195    ///
196    /// Returns an error if invocation fails, the given arguments cannot be turned into a valid RPC Request message,
197    /// result protobuf deserialization fails, or result payload is empty.
198    pub async fn invoke_proto_method<T, R>(
199        &self,
200        method: UUri,
201        call_options: CallOptions,
202        request_message: T,
203    ) -> Result<R, ServiceInvocationError>
204    where
205        T: crate::ProtobufMappable,
206        R: crate::ProtobufMappable + Default,
207    {
208        let payload = UPayload::try_from_protobuf(request_message)
209            .map_err(|e| ServiceInvocationError::InvalidArgument(e.to_string()))?;
210
211        let result = self
212            .invoke_method(method, call_options, Some(payload))
213            .await?;
214
215        if let Some(result) = result {
216            UPayload::extract_protobuf::<R>(&result)
217                .map_err(|e| ServiceInvocationError::InvalidArgument(e.to_string()))
218        } else {
219            Err(ServiceInvocationError::InvalidArgument(
220                "No payload".to_string(),
221            ))
222        }
223    }
224}
225
226/// A handler for processing incoming RPC requests.
227///
228// [impl->dsn~communication-layer-api-declaration~1]
229#[cfg_attr(any(test, feature = "test-util"), mockall::automock)]
230#[async_trait]
231pub trait RequestHandler: Send + Sync {
232    /// Handles a request to invoke a method with given input parameters.
233    ///
234    /// Implementations MUST NOT block the calling thread. Long running
235    /// computations should be performed on a separate worker thread, yielding
236    /// on the calling thread.
237    ///
238    /// # Arguments
239    ///
240    /// * `resource_id` - The resource identifier of the method to invoke.
241    /// * `message_attributes` - Any metadata that is associated with the request message.
242    /// * `request_payload` - The raw payload that contains the input data for the method.
243    ///
244    /// # Returns
245    ///
246    /// the output data generated by the method.
247    ///
248    /// # Errors
249    ///
250    /// Returns an error if the request could not be processed successfully.
251    async fn handle_request(
252        &self,
253        resource_id: u16,
254        message_attributes: &UAttributes,
255        request_payload: Option<UPayload>,
256    ) -> Result<Option<UPayload>, ServiceInvocationError>;
257}
258
259/// A server for exposing Remote Procedure Call (RPC) endpoints.
260///
261/// Please refer to the
262/// [Communication Layer API specification](https://github.com/eclipse-uprotocol/up-spec/blob/main/up-l2/api.adoc)
263/// for details.
264// [impl->dsn~communication-layer-api-declaration~1]
265#[async_trait]
266pub trait RpcServer {
267    /// Registers an endpoint for RPC requests.
268    ///
269    /// Note that only a single endpoint can be registered for a given resource ID.
270    /// However, the same request handler can be registered for multiple endpoints.
271    ///
272    /// # Arguments
273    ///
274    /// * `origin_filter` - A pattern defining origin addresses to accept requests from. If `None`, requests
275    ///                     will be accepted from all sources.
276    /// * `resource_id` - The resource identifier of the (local) method to accept requests for.
277    /// * `request_handler` - The handler to invoke for each incoming request.
278    ///
279    /// # Errors
280    ///
281    /// Returns an error if the listener cannot be registered or if a listener has already been registered
282    /// for the given resource ID.
283    async fn register_endpoint(
284        &self,
285        origin_filter: Option<&UUri>,
286        resource_id: u16,
287        request_handler: Arc<dyn RequestHandler>,
288    ) -> Result<(), RegistrationError>;
289
290    /// Deregisters a previously [registered endpoint](Self::register_endpoint).
291    ///
292    /// # Arguments
293    ///
294    /// * `origin_filter` - The origin pattern that the endpoint had been registered for.
295    /// * `resource_id` - The (local) resource identifier that the endpoint had been registered for.
296    /// * `request_handler` - The handler to unregister.
297    ///
298    /// # Errors
299    ///
300    /// Returns an error if the listener cannot be unregistered.
301    async fn unregister_endpoint(
302        &self,
303        origin_filter: Option<&UUri>,
304        resource_id: u16,
305        request_handler: Arc<dyn RequestHandler>,
306    ) -> Result<(), RegistrationError>;
307}
308
309#[cfg(not(tarpaulin_include))]
310#[cfg(any(test, feature = "test-util"))]
311mockall::mock! {
312    /// This extra struct is necessary in order to comply with mockall's requirements regarding the parameter lifetimes
313    /// see <https://github.com/asomers/mockall/issues/571>
314    pub RpcServerImpl {
315        /// Mock hook; configure with mockall expectations.
316    ///
317    /// # Errors
318    ///
319    /// Returns whatever the test's expectations are configured to return.
320        pub async fn do_register_endpoint<'a>(&'a self, origin_filter: Option<&'a UUri>, resource_id: u16, request_handler: Arc<dyn RequestHandler>) -> Result<(), RegistrationError>;
321        /// Mock hook; configure with mockall expectations.
322    ///
323    /// # Errors
324    ///
325    /// Returns whatever the test's expectations are configured to return.
326        pub async fn do_unregister_endpoint<'a>(&'a self, origin_filter: Option<&'a UUri>, resource_id: u16, request_handler: Arc<dyn RequestHandler>) -> Result<(), RegistrationError>;
327    }
328}
329
330#[cfg(not(tarpaulin_include))]
331#[cfg(any(test, feature = "test-util"))]
332#[async_trait]
333/// This delegates the invocation of the UTransport functions to the mocked functions of the Transport struct.
334/// see <https://github.com/asomers/mockall/issues/571>
335impl RpcServer for MockRpcServerImpl {
336    async fn register_endpoint(
337        &self,
338        origin_filter: Option<&UUri>,
339        resource_id: u16,
340        request_handler: Arc<dyn RequestHandler>,
341    ) -> Result<(), RegistrationError> {
342        self.do_register_endpoint(origin_filter, resource_id, request_handler)
343            .await
344    }
345    async fn unregister_endpoint(
346        &self,
347        origin_filter: Option<&UUri>,
348        resource_id: u16,
349        request_handler: Arc<dyn RequestHandler>,
350    ) -> Result<(), RegistrationError> {
351        self.do_unregister_endpoint(origin_filter, resource_id, request_handler)
352            .await
353    }
354}
355
356#[cfg(all(test, feature = "protobuf-support"))]
357mod tests {
358    use crate::{communication::CallOptions, UUri};
359    use protobuf::well_known_types::wrappers::{DoubleValue, StringValue};
360
361    use super::*;
362
363    #[tokio::test]
364    async fn test_invoke_proto_method_fails_for_unexpected_return_type() {
365        let mut rpc_client = MockRpcClient::new();
366        rpc_client
367            .expect_invoke_method()
368            .once()
369            .returning(|_method, _options, _payload| {
370                let response_payload = UPayload::try_from_protobuf(DoubleValue::new()).unwrap();
371                Ok(Some(response_payload))
372            });
373        let client: Arc<dyn RpcClient> = Arc::new(rpc_client);
374        let mut request = StringValue::new();
375        request.value = "hello".to_string();
376        let result = client
377            .invoke_proto_method::<StringValue, StringValue>(
378                UUri::try_from_parts("", 0x1000, 0x01, 0x0001).unwrap(),
379                CallOptions::for_rpc_request(5_000, None, None, None),
380                request,
381            )
382            .await;
383        assert!(result.is_err_and(|e| matches!(e, ServiceInvocationError::InvalidArgument(_))));
384    }
385
386    #[tokio::test]
387    async fn test_invoke_proto_method_fails_for_missing_response_payload() {
388        let mut rpc_client = MockRpcClient::new();
389        rpc_client
390            .expect_invoke_method()
391            .once()
392            .return_const(Ok(None));
393        let client: Arc<dyn RpcClient> = Arc::new(rpc_client);
394        let mut request = StringValue::new();
395        request.value = "hello".to_string();
396        let result = client
397            .invoke_proto_method::<StringValue, StringValue>(
398                UUri::try_from_parts("", 0x1000, 0x01, 0x0001).unwrap(),
399                CallOptions::for_rpc_request(5_000, None, None, None),
400                request,
401            )
402            .await;
403        assert!(result.is_err_and(|e| matches!(e, ServiceInvocationError::InvalidArgument(_))));
404    }
405}
up_rust/communication/rpc.rs

up_rust/communication/
rpc.rs
