up_rust/core/udiscovery/
udiscovery_client.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 ********************************************************************************/
13use std::sync::Arc;
14
15use async_trait::async_trait;
16
17use crate::{
18    communication::{CallOptions, RpcClient},
19    core::udiscovery::{
20        TopicInfo, UDiscovery, RESOURCE_ID_FIND_SERVICES, RESOURCE_ID_GET_SERVICE_TOPICS,
21        UDISCOVERY_TYPE_ID, UDISCOVERY_VERSION_MAJOR,
22    },
23    up_core_api::{
24        udiscovery::{
25            FindServicesRequest, FindServicesResponse, GetServiceTopicsRequest,
26            GetServiceTopicsResponse, ServiceTopicInfo,
27        },
28        uri::UUri as UUriProto,
29    },
30    UCode, UStatus, UUri,
31};
32
33/// Gets a UUri referring to one of the _local_ uDiscovery service's resources.
34fn udiscovery_uri(resource_id: u16) -> UUri {
35    UUri::try_from_parts(
36        "",
37        UDISCOVERY_TYPE_ID,
38        UDISCOVERY_VERSION_MAJOR,
39        resource_id,
40    )
41    .unwrap()
42}
43
44impl TryFrom<&ServiceTopicInfo> for TopicInfo {
45    type Error = UStatus;
46
47    fn try_from(service_topic_info_proto: &ServiceTopicInfo) -> Result<Self, Self::Error> {
48        let Some(topic_proto) = service_topic_info_proto.topic.as_ref() else {
49            return Err(UStatus::fail_with_code(
50                UCode::InvalidArgument,
51                "Service returned invalid ServiceTopicInfo: no topic",
52            ));
53        };
54        let Ok(topic) = UUri::try_from(topic_proto) else {
55            return Err(UStatus::fail_with_code(
56                UCode::InvalidArgument,
57                "Service returned invalid ServiceTopicInfo: malformed topic URI",
58            ));
59        };
60        let Some(uservice_topic) = service_topic_info_proto.info.as_ref() else {
61            return Err(UStatus::fail_with_code(
62                UCode::InvalidArgument,
63                "Service returned invalid ServiceTopicInfo: no info object",
64            ));
65        };
66        Ok(TopicInfo {
67            topic,
68            message_type: uservice_topic.message.clone(),
69            permission_level: uservice_topic.permission_level,
70            ttl: service_topic_info_proto.ttl,
71        })
72    }
73}
74
75/// A [`UDiscovery`] client implementation for invoking operations of a local uDiscovery service.
76///
77/// The client requires an [`RpcClient`] for performing the remote procedure calls.
78pub struct RpcClientUDiscovery {
79    rpc_client: Arc<dyn RpcClient>,
80}
81
82impl core::fmt::Debug for RpcClientUDiscovery {
83    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
84        f.debug_struct("RpcClientUDiscovery")
85            .finish_non_exhaustive()
86    }
87}
88
89impl RpcClientUDiscovery {
90    /// Creates a new uDiscovery client for a given transport.
91    ///
92    /// # Arguments
93    ///
94    /// * `rpc_client` - The client to use for performing the remote procedure calls on the service.
95    pub fn new(rpc_client: Arc<dyn RpcClient>) -> Self {
96        RpcClientUDiscovery { rpc_client }
97    }
98
99    fn default_call_options() -> CallOptions {
100        CallOptions::for_rpc_request(5_000, None, None, None)
101    }
102}
103
104#[async_trait]
105impl UDiscovery for RpcClientUDiscovery {
106    async fn find_services(
107        &self,
108        uri_pattern: UUri,
109        recursive: bool,
110    ) -> Result<Vec<UUri>, UStatus> {
111        let request_message = FindServicesRequest {
112            uri: Some(UUriProto::from(&uri_pattern)).into(),
113            recursive,
114            ..Default::default()
115        };
116        self.rpc_client
117            .invoke_proto_method::<_, FindServicesResponse>(
118                udiscovery_uri(RESOURCE_ID_FIND_SERVICES),
119                Self::default_call_options(),
120                request_message,
121            )
122            .await
123            .and_then(|response_message| {
124                let mut result = vec![];
125                if let Some(uri_batch) = response_message.uris.as_ref() {
126                    for uri in &uri_batch.uris {
127                        result.push(UUri::try_from(uri).map_err(|_| {
128                            UStatus::fail_with_code(
129                                UCode::InvalidArgument,
130                                "Service returned invalid URI in FindServicesResponse",
131                            )
132                        })?);
133                    }
134                }
135                Ok(result)
136            })
137            .map_err(UStatus::from)
138    }
139
140    async fn get_service_topics(
141        &self,
142        topic_pattern: UUri,
143        recursive: bool,
144    ) -> Result<Vec<TopicInfo>, UStatus> {
145        let request_message = GetServiceTopicsRequest {
146            topic: Some(UUriProto::from(&topic_pattern)).into(),
147            recursive,
148            ..Default::default()
149        };
150        self.rpc_client
151            .invoke_proto_method::<_, GetServiceTopicsResponse>(
152                udiscovery_uri(RESOURCE_ID_GET_SERVICE_TOPICS),
153                Self::default_call_options(),
154                request_message,
155            )
156            .await
157            .and_then(|response_message| {
158                let mut result = vec![];
159                for topic in response_message.topics {
160                    result.push(TopicInfo::try_from(&topic)?);
161                }
162                Ok(result)
163            })
164            .map_err(UStatus::from)
165    }
166}
167
168#[cfg(test)]
169mod tests {
170    use mockall::Sequence;
171
172    use super::*;
173    use crate::{
174        communication::{MockRpcClient, UPayload},
175        up_core_api::{udiscovery::ServiceTopicInfo, uoptions::UServiceTopic, uri::UUriBatch},
176        UCode, UUri,
177    };
178    use std::sync::Arc;
179
180    #[tokio::test]
181    async fn test_find_services_invokes_rpc_client() {
182        let service_pattern_uri = UUri::try_from_parts("other", 0xFFFF_D5A3, 0x01, 0xFFFF)
183            .expect("failed to create service pattern URI");
184        let request = FindServicesRequest {
185            uri: Some(UUriProto::from(&service_pattern_uri)).into(),
186            ..Default::default()
187        };
188        let expected_request = request.clone();
189        let mut rpc_client = MockRpcClient::new();
190        let mut seq = Sequence::new();
191        rpc_client
192            .expect_invoke_method()
193            .once()
194            .in_sequence(&mut seq)
195            .withf(|method, _options, payload| {
196                method == &udiscovery_uri(RESOURCE_ID_FIND_SERVICES) && payload.is_some()
197            })
198            .return_const(Err(crate::communication::ServiceInvocationError::Internal(
199                "internal error".to_string(),
200            )));
201        rpc_client
202            .expect_invoke_method()
203            .once()
204            .in_sequence(&mut seq)
205            .withf(move |method, _options, payload| {
206                let request = payload
207                    .to_owned()
208                    .unwrap()
209                    .extract_protobuf::<FindServicesRequest>()
210                    .unwrap();
211                request == expected_request && method == &udiscovery_uri(RESOURCE_ID_FIND_SERVICES)
212            })
213            .returning(move |_method, _options, _payload| {
214                let response = FindServicesResponse {
215                    uris: Some(UUriBatch {
216                        uris: vec![UUriProto {
217                            authority_name: "other".to_string(),
218                            ue_id: 0x0004_D5A3,
219                            ue_version_major: 0x01,
220                            resource_id: 0xD3FE,
221                            ..Default::default()
222                        }],
223                        ..Default::default()
224                    })
225                    .into(),
226                    ..Default::default()
227                };
228                Ok(Some(UPayload::try_from_protobuf(response).unwrap()))
229            });
230
231        let udiscovery_client = RpcClientUDiscovery::new(Arc::new(rpc_client));
232
233        assert!(udiscovery_client
234            .find_services(service_pattern_uri.clone(), false)
235            .await
236            .is_err_and(|e| e.code() == UCode::Internal));
237        assert!(udiscovery_client
238            .find_services(service_pattern_uri.clone(), false)
239            .await
240            .is_ok_and(
241                |result| result.len() == 1 && service_pattern_uri.matches(result.first().unwrap())
242            ));
243    }
244
245    #[tokio::test]
246    async fn test_get_service_topics_invokes_rpc_client() {
247        let topic_pattern_uri = UUri::try_from_parts("*", 0xFFFF_D5A3, 0x01, 0xFFFF)
248            .expect("failed to create topic pattern URI");
249        let request = GetServiceTopicsRequest {
250            topic: Some((&topic_pattern_uri).into()).into(),
251            ..Default::default()
252        };
253        let expected_request = request.clone();
254        let mut rpc_client = MockRpcClient::new();
255        let mut seq = Sequence::new();
256        rpc_client
257            .expect_invoke_method()
258            .once()
259            .in_sequence(&mut seq)
260            .withf(|method, _options, payload| {
261                method == &udiscovery_uri(RESOURCE_ID_GET_SERVICE_TOPICS) && payload.is_some()
262            })
263            .return_const(Err(crate::communication::ServiceInvocationError::Internal(
264                "internal error".to_string(),
265            )));
266        rpc_client
267            .expect_invoke_method()
268            .once()
269            .in_sequence(&mut seq)
270            .withf(move |method, _options, payload| {
271                let request = payload
272                    .to_owned()
273                    .unwrap()
274                    .extract_protobuf::<GetServiceTopicsRequest>()
275                    .unwrap();
276                request == expected_request
277                    && method == &udiscovery_uri(RESOURCE_ID_GET_SERVICE_TOPICS)
278            })
279            .returning(move |_method, _options, _payload| {
280                let topic_info = ServiceTopicInfo {
281                    topic: Some(UUriProto {
282                        authority_name: "other".to_string(),
283                        ue_id: 0x0004_D5A3,
284                        ue_version_major: 0x01,
285                        resource_id: 0xD3FE,
286                        ..Default::default()
287                    })
288                    .into(),
289                    info: Some(UServiceTopic {
290                        id: 0x9000,
291                        name: "TestTopic".to_string(),
292                        message: "TestTopicMessage".to_string(),
293                        ..Default::default()
294                    })
295                    .into(),
296                    ttl: 600,
297                    ..Default::default()
298                };
299                let response = GetServiceTopicsResponse {
300                    topics: vec![topic_info],
301                    ..Default::default()
302                };
303                Ok(Some(UPayload::try_from_protobuf(response).unwrap()))
304            });
305
306        let udiscovery_client = RpcClientUDiscovery::new(Arc::new(rpc_client));
307
308        assert!(udiscovery_client
309            .get_service_topics(topic_pattern_uri.clone(), false)
310            .await
311            .is_err_and(|e| e.code() == UCode::Internal));
312        assert!(udiscovery_client
313            .get_service_topics(topic_pattern_uri.clone(), false)
314            .await
315            .is_ok_and(|result| result.len() == 1
316                && topic_pattern_uri.matches(result.first().map(|r| r.topic()).unwrap())));
317    }
318}
up_rust/core/udiscovery/udiscovery_client.rs

up_rust/core/udiscovery/
udiscovery_client.rs
