up_rust/communication/
owned.rs

1/********************************************************************************
2 * Copyright (c) 2026 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//! Owned native-frame communication-layer facade.
15//!
16//! This module is additive to the existing `communication` API. It builds L2
17//! roles on top of [`UOwnedTransport`] without changing [`crate::UTransport`]
18//! or the existing `communication::{Publisher, Subscriber, Notifier, RpcClient,
19//! RpcServer}` traits. The owned receive/listener roles convert owned native
20//! frames back to `UMessage` before invoking ordinary [`crate::UListener`]
21//! handlers, so callers do not need to name transport core, loan, or metadata
22//! codec types.
23
24use std::collections::{hash_map::Entry, HashMap};
25use std::sync::Arc;
26
27use async_trait::async_trait;
28
29#[cfg(feature = "selected-wire-transport-adapter")]
30use crate::payload::codec::{DecodePayload, EncodePayload, PayloadCodec};
31#[cfg(feature = "selected-wire-transport-adapter")]
32use crate::wire::UWireDecodeOwned;
33#[cfg(feature = "selected-wire-transport-adapter")]
34use crate::wire::UWireEncode;
35#[cfg(feature = "selected-wire-transport-adapter")]
36use crate::wire_transport::UHasWire;
37use crate::{
38    communication::{
39        CallOptions, NotificationError, PubSubError, RegistrationError, ServiceInvocationError,
40        SubscriptionChangeHandler, SubscriptionStatus, UPayload,
41    },
42    LocalUriProvider, UCode, UListener, UMessage, UMessageBuilder, UOwnedFrame, UOwnedListener,
43    UOwnedTransport, UStatus, UUri, UUID,
44};
45
46pub use crate::communication::RequestHandler;
47
48/// Front door for owned native-frame communication-layer clients.
49///
50/// *Role: the up-L2 role set over the **owned-frame family** — same
51/// role semantics as the `UTransport`-family layer, carrying validated frames
52/// instead of messages. See the
53/// [guide](crate::guide::applications::communication).*
54pub struct Endpoint<T, P>
55where
56    T: UOwnedTransport + ?Sized,
57    P: LocalUriProvider + ?Sized,
58{
59    transport: Arc<T>,
60    uri_provider: Arc<P>,
61}
62
63impl<T, P> core::fmt::Debug for Endpoint<T, P>
64where
65    T: UOwnedTransport + ?Sized,
66    P: LocalUriProvider + ?Sized,
67{
68    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
69        f.debug_struct("Endpoint").finish_non_exhaustive()
70    }
71}
72
73impl<T, P> Endpoint<T, P>
74where
75    T: UOwnedTransport + ?Sized,
76    P: LocalUriProvider + ?Sized,
77{
78    /// Creates an owned native-frame communication endpoint.
79    #[must_use]
80    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
81        Self {
82            transport,
83            uri_provider,
84        }
85    }
86
87    /// Creates a publisher for owned native-frame publish messages.
88    #[must_use]
89    pub fn publisher(&self) -> Publisher<T, P> {
90        Publisher::new(self.transport.clone(), self.uri_provider.clone())
91    }
92
93    /// Creates a subscriber for owned native-frame topic listeners.
94    ///
95    /// Subscription is a service interaction: the subscriber needs a
96    /// uSubscription client in addition to this endpoint's transport.
97    #[must_use]
98    pub fn subscriber(
99        &self,
100        usubscription: Arc<dyn crate::core::usubscription::USubscription>,
101    ) -> Subscriber<T> {
102        Subscriber::new(self.transport.clone(), usubscription)
103    }
104
105    /// Creates a notifier for owned native-frame notification send/listen roles.
106    #[must_use]
107    pub fn notifier(&self) -> Notifier<T, P> {
108        Notifier::new(self.transport.clone(), self.uri_provider.clone())
109    }
110
111    /// Creates an RPC client for owned native-frame request/response messages.
112    #[must_use]
113    pub fn rpc_client(&self) -> RpcClient<T, P> {
114        RpcClient::new(self.transport.clone(), self.uri_provider.clone())
115    }
116}
117
118impl<T, P> Endpoint<T, P>
119where
120    T: UOwnedTransport + ?Sized + 'static,
121    P: LocalUriProvider + ?Sized,
122{
123    /// Creates an RPC server for owned native-frame request/response messages.
124    #[must_use]
125    pub fn rpc_server(&self) -> RpcServer<T, P> {
126        RpcServer::new(self.transport.clone(), self.uri_provider.clone())
127    }
128}
129
130fn build_message_with_payload<S: crate::umessage::BuilderState>(
131    builder: &mut UMessageBuilder<S>,
132    payload: Option<UPayload>,
133) -> Result<UMessage, crate::UMessageError> {
134    match payload {
135        Some(payload) => {
136            builder.build_with_payload(payload.payload().clone(), payload.payload_format())
137        }
138        None => builder.build(),
139    }
140}
141
142fn frame_from_message(message: &UMessage) -> Result<UOwnedFrame, crate::UFrameMetadataError> {
143    let metadata = crate::frame::metadata::try_project_umessage_to_frame_metadata(message)?;
144    UOwnedFrame::new(metadata, message.payload())
145}
146
147fn message_from_frame(frame: UOwnedFrame) -> Result<UMessage, crate::UFrameMetadataError> {
148    crate::frame::metadata::try_project_frame_to_umessage(
149        frame.metadata().clone(),
150        frame.payload().cloned(),
151    )
152}
153
154fn listener_pointer(listener: &Arc<dyn UListener>) -> usize {
155    let ptr = Arc::as_ptr(listener);
156    let thin_ptr = ptr as *const ();
157    thin_ptr as usize
158}
159
160#[derive(Clone, Debug, Eq, Hash, PartialEq)]
161struct MessageListenerKey {
162    source_filter: UUri,
163    sink_filter: Option<UUri>,
164    listener: usize,
165}
166
167impl MessageListenerKey {
168    fn new(
169        source_filter: &UUri,
170        sink_filter: Option<&UUri>,
171        listener: &Arc<dyn UListener>,
172    ) -> Self {
173        Self {
174            source_filter: source_filter.clone(),
175            sink_filter: sink_filter.cloned(),
176            listener: listener_pointer(listener),
177        }
178    }
179}
180
181struct MessageListener {
182    inner: Arc<dyn UListener>,
183}
184
185#[async_trait]
186impl UOwnedListener for MessageListener {
187    async fn on_receive_owned(&self, frame: UOwnedFrame) {
188        if let Ok(message) = message_from_frame(frame) {
189            self.inner.on_receive(message).await;
190        }
191    }
192}
193
194type MessageListenerMap = tokio::sync::Mutex<HashMap<MessageListenerKey, Arc<dyn UOwnedListener>>>;
195
196async fn register_message_listener<T>(
197    transport: &T,
198    listeners: &MessageListenerMap,
199    source_filter: &UUri,
200    sink_filter: Option<&UUri>,
201    listener: Arc<dyn UListener>,
202) -> Result<(), RegistrationError>
203where
204    T: UOwnedTransport + ?Sized,
205{
206    let key = MessageListenerKey::new(source_filter, sink_filter, &listener);
207    if listeners.lock().await.contains_key(&key) {
208        return Err(RegistrationError::AlreadyExists);
209    }
210    let owned_listener: Arc<dyn UOwnedListener> = Arc::new(MessageListener { inner: listener });
211    transport
212        .register_owned_listener(source_filter, sink_filter, owned_listener.clone())
213        .await
214        .map_err(RegistrationError::from)?;
215
216    let mut listeners = listeners.lock().await;
217    match listeners.entry(key) {
218        Entry::Vacant(entry) => {
219            entry.insert(owned_listener);
220            Ok(())
221        }
222        Entry::Occupied(_) => {
223            drop(listeners);
224            let _ = transport
225                .unregister_owned_listener(source_filter, sink_filter, owned_listener)
226                .await;
227            Err(RegistrationError::AlreadyExists)
228        }
229    }
230}
231
232async fn unregister_message_listener<T>(
233    transport: &T,
234    listeners: &MessageListenerMap,
235    source_filter: &UUri,
236    sink_filter: Option<&UUri>,
237    listener: Arc<dyn UListener>,
238) -> Result<(), RegistrationError>
239where
240    T: UOwnedTransport + ?Sized,
241{
242    let key = MessageListenerKey::new(source_filter, sink_filter, &listener);
243    let Some(owned_listener) = listeners.lock().await.remove(&key) else {
244        return Err(RegistrationError::NoSuchListener);
245    };
246    if let Err(error) = transport
247        .unregister_owned_listener(source_filter, sink_filter, owned_listener)
248        .await
249    {
250        return Err(RegistrationError::from(error));
251    }
252    Ok(())
253}
254
255/// Subscriber implemented over an owned native-frame transport.
256///
257/// *Role: the up-L2 `Subscriber` for the owned family. Like its `UTransport`-family
258/// sibling, subscription is a **service interaction**: it requires a
259/// [`USubscription`](crate::core::usubscription::USubscription) client and
260/// informs the uSubscription service before registering the local listener,
261/// per the uProtocol specification.*
262pub struct Subscriber<T>
263where
264    T: UOwnedTransport + ?Sized,
265{
266    transport: Arc<T>,
267    usubscription: Arc<dyn crate::core::usubscription::USubscription>,
268    listeners: MessageListenerMap,
269}
270
271impl<T> core::fmt::Debug for Subscriber<T>
272where
273    T: UOwnedTransport + ?Sized,
274{
275    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
276        f.debug_struct("Subscriber").finish_non_exhaustive()
277    }
278}
279
280impl<T> Subscriber<T>
281where
282    T: UOwnedTransport + ?Sized,
283{
284    /// Creates a subscriber over an owned-frame transport and a uSubscription
285    /// service client.
286    pub fn new(
287        transport: Arc<T>,
288        usubscription: Arc<dyn crate::core::usubscription::USubscription>,
289    ) -> Self {
290        Self {
291            transport,
292            usubscription,
293            listeners: MessageListenerMap::default(),
294        }
295    }
296
297    /// Subscribes to a topic: informs the uSubscription service first, then
298    /// registers the owned-frame listener on success.
299    pub async fn subscribe(
300        &self,
301        topic: &UUri,
302        handler: Arc<dyn UListener>,
303        subscription_change_handler: Option<Arc<dyn SubscriptionChangeHandler>>,
304    ) -> Result<(), RegistrationError> {
305        super::validate_listener_topic(topic)?;
306        let state = self
307            .usubscription
308            .subscribe(topic, None, None)
309            .await
310            .map_err(|status| RegistrationError::Unknown(Box::new(status)))?;
311        if state != SubscriptionStatus::Subscribed && state != SubscriptionStatus::SubscribePending
312        {
313            return Err(RegistrationError::Unknown(Box::new(
314                crate::UStatus::fail_with_code(
315                    crate::UCode::FailedPrecondition,
316                    format!("uSubscription service returned {state:?}"),
317                ),
318            )));
319        }
320        register_message_listener(&*self.transport, &self.listeners, topic, None, handler).await?;
321        if let Some(handler) = subscription_change_handler {
322            handler.on_subscription_change(topic.clone(), state);
323        }
324        Ok(())
325    }
326
327    /// Unsubscribes from a topic: informs the uSubscription service, then
328    /// unregisters the owned-frame listener.
329    pub async fn unsubscribe(
330        &self,
331        topic: &UUri,
332        handler: Arc<dyn UListener>,
333    ) -> Result<(), RegistrationError> {
334        super::validate_listener_topic(topic)?;
335        self.usubscription
336            .unsubscribe(topic)
337            .await
338            .map_err(|status| RegistrationError::Unknown(Box::new(status)))?;
339        unregister_message_listener(&*self.transport, &self.listeners, topic, None, handler).await
340    }
341}
342
343#[async_trait]
344impl<T> crate::communication::Subscriber for Subscriber<T>
345where
346    T: UOwnedTransport + ?Sized,
347{
348    async fn subscribe(
349        &self,
350        topic: &UUri,
351        handler: Arc<dyn UListener>,
352        subscription_change_handler: Option<Arc<dyn SubscriptionChangeHandler>>,
353    ) -> Result<(), RegistrationError> {
354        Subscriber::subscribe(self, topic, handler, subscription_change_handler).await
355    }
356
357    async fn unsubscribe(
358        &self,
359        topic: &UUri,
360        handler: Arc<dyn UListener>,
361    ) -> Result<(), RegistrationError> {
362        Subscriber::unsubscribe(self, topic, handler).await
363    }
364}
365
366/// Notifier implemented over an owned native-frame transport.
367pub struct Notifier<T, P>
368where
369    T: UOwnedTransport + ?Sized,
370    P: LocalUriProvider + ?Sized,
371{
372    transport: Arc<T>,
373    uri_provider: Arc<P>,
374    listeners: MessageListenerMap,
375}
376
377impl<T, P> core::fmt::Debug for Notifier<T, P>
378where
379    T: UOwnedTransport + ?Sized,
380    P: LocalUriProvider + ?Sized,
381{
382    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
383        f.debug_struct("Notifier").finish_non_exhaustive()
384    }
385}
386
387impl<T, P> Notifier<T, P>
388where
389    T: UOwnedTransport + ?Sized,
390    P: LocalUriProvider + ?Sized,
391{
392    /// Creates a notifier over an owned native-frame transport.
393    #[must_use]
394    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
395        Self {
396            transport,
397            uri_provider,
398            listeners: tokio::sync::Mutex::new(HashMap::new()),
399        }
400    }
401
402    fn build_frame(
403        &self,
404        resource_id: u16,
405        destination: &UUri,
406        call_options: CallOptions,
407        payload: Option<UPayload>,
408    ) -> Result<UOwnedFrame, NotificationError> {
409        let mut builder = UMessageBuilder::notification(
410            self.uri_provider.get_resource_uri(resource_id),
411            destination.clone(),
412        );
413        builder.with_ttl(call_options.ttl());
414        if let Some(message_id) = call_options.message_id() {
415            builder.with_message_id(message_id.clone());
416        }
417        if let Some(priority) = call_options.priority() {
418            builder.with_priority(priority);
419        }
420        let message = build_message_with_payload(&mut builder, payload)
421            .map_err(|error| NotificationError::InvalidArgument(error.to_string()))?;
422        frame_from_message(&message)
423            .map_err(|error| NotificationError::InvalidArgument(error.to_string()))
424    }
425
426    /// Sends an owned native-frame notification.
427    pub async fn notify(
428        &self,
429        resource_id: u16,
430        destination: &UUri,
431        call_options: CallOptions,
432        payload: Option<UPayload>,
433    ) -> Result<(), NotificationError> {
434        let frame = self.build_frame(resource_id, destination, call_options, payload)?;
435        self.transport
436            .send_owned(frame)
437            .await
438            .map_err(Box::from)
439            .map_err(NotificationError::NotifyError)
440    }
441
442    /// Starts listening for owned native-frame notifications on a topic.
443    pub async fn start_listening(
444        &self,
445        topic: &UUri,
446        listener: Arc<dyn UListener>,
447    ) -> Result<(), RegistrationError> {
448        super::validate_listener_topic(topic)?;
449        register_message_listener(
450            &*self.transport,
451            &self.listeners,
452            topic,
453            Some(&self.uri_provider.get_source_uri()),
454            listener,
455        )
456        .await
457    }
458
459    /// Stops listening for owned native-frame notifications on a topic.
460    pub async fn stop_listening(
461        &self,
462        topic: &UUri,
463        listener: Arc<dyn UListener>,
464    ) -> Result<(), RegistrationError> {
465        super::validate_listener_topic(topic)?;
466        unregister_message_listener(
467            &*self.transport,
468            &self.listeners,
469            topic,
470            Some(&self.uri_provider.get_source_uri()),
471            listener,
472        )
473        .await
474    }
475}
476
477#[async_trait]
478impl<T, P> crate::communication::Notifier for Notifier<T, P>
479where
480    T: UOwnedTransport + ?Sized,
481    P: LocalUriProvider + ?Sized,
482{
483    async fn notify(
484        &self,
485        resource_id: u16,
486        destination: &UUri,
487        call_options: CallOptions,
488        payload: Option<UPayload>,
489    ) -> Result<(), NotificationError> {
490        Notifier::notify(self, resource_id, destination, call_options, payload).await
491    }
492
493    async fn start_listening(
494        &self,
495        topic: &UUri,
496        listener: Arc<dyn UListener>,
497    ) -> Result<(), RegistrationError> {
498        Notifier::start_listening(self, topic, listener).await
499    }
500
501    async fn stop_listening(
502        &self,
503        topic: &UUri,
504        listener: Arc<dyn UListener>,
505    ) -> Result<(), RegistrationError> {
506        Notifier::stop_listening(self, topic, listener).await
507    }
508}
509
510/// RPC client implemented over an owned native-frame transport.
511pub struct RpcClient<T, P>
512where
513    T: UOwnedTransport + ?Sized,
514    P: LocalUriProvider + ?Sized,
515{
516    transport: Arc<T>,
517    uri_provider: Arc<P>,
518}
519
520impl<T, P> core::fmt::Debug for RpcClient<T, P>
521where
522    T: UOwnedTransport + ?Sized,
523    P: LocalUriProvider + ?Sized,
524{
525    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
526        f.debug_struct("RpcClient").finish_non_exhaustive()
527    }
528}
529
530impl<T, P> RpcClient<T, P>
531where
532    T: UOwnedTransport + ?Sized,
533    P: LocalUriProvider + ?Sized,
534{
535    /// Creates an RPC client over an owned native-frame transport.
536    #[must_use]
537    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
538        Self {
539            transport,
540            uri_provider,
541        }
542    }
543
544    fn build_request_frame(
545        &self,
546        method: UUri,
547        call_options: &CallOptions,
548        message_id: UUID,
549        payload: Option<UPayload>,
550    ) -> Result<UOwnedFrame, ServiceInvocationError> {
551        let mut builder = UMessageBuilder::request(
552            method,
553            self.uri_provider.get_source_uri(),
554            call_options.ttl(),
555        );
556        builder.with_message_id(message_id);
557        if let Some(token) = call_options.token() {
558            builder.with_token(token.clone());
559        }
560        if let Some(priority) = call_options.priority() {
561            builder.with_priority(priority);
562        }
563        let message = build_message_with_payload(&mut builder, payload)
564            .map_err(|error| ServiceInvocationError::InvalidArgument(error.to_string()))?;
565        frame_from_message(&message)
566            .map_err(|error| ServiceInvocationError::InvalidArgument(error.to_string()))
567    }
568
569    fn response_payload(response: UMessage) -> Result<Option<UPayload>, ServiceInvocationError> {
570        match response.commstatus() {
571            Some(UCode::Ok) | None => {
572                let payload_format = response
573                    .payload_format()
574                    .unwrap_or(crate::UPayloadFormat::Unspecified);
575                Ok(response
576                    .payload()
577                    .map(|payload| UPayload::new(payload, payload_format)))
578            }
579            Some(code) => {
580                let status = response.extract_protobuf().unwrap_or_else(|_| {
581                    UStatus::fail_with_code(code, "failed to invoke service operation")
582                });
583                Err(ServiceInvocationError::from(status))
584            }
585        }
586    }
587
588    /// Invokes an RPC method using owned native frames.
589    pub async fn invoke_method(
590        &self,
591        method: UUri,
592        call_options: CallOptions,
593        payload: Option<UPayload>,
594    ) -> Result<Option<UPayload>, ServiceInvocationError> {
595        let message_id = call_options
596            .message_id()
597            .map_or_else(UUID::build, Clone::clone);
598        let request_frame =
599            self.build_request_frame(method.clone(), &call_options, message_id.clone(), payload)?;
600        self.transport.send_owned(request_frame).await?;
601        let response_frame = self
602            .transport
603            .receive_owned(&method, Some(&self.uri_provider.get_source_uri()))
604            .await?;
605        let response = crate::frame::metadata::try_project_frame_to_umessage(
606            response_frame.metadata().clone(),
607            response_frame.payload().cloned(),
608        )
609        .map_err(|error| ServiceInvocationError::InvalidArgument(error.to_string()))?;
610        if response.request_id_unchecked() != &message_id {
611            return Err(ServiceInvocationError::Internal(
612                "received RPC response for unexpected request ID".to_string(),
613            ));
614        }
615        Self::response_payload(response)
616    }
617}
618
619#[cfg(feature = "selected-wire-transport-adapter")]
620impl<T, P> RpcClient<T, P>
621where
622    T: UOwnedTransport + UHasWire + ?Sized,
623    P: LocalUriProvider + ?Sized,
624{
625    /// Invokes an RPC method using the transport endpoint's selected wire.
626    pub async fn invoke_typed<Request, Response>(
627        &self,
628        method: UUri,
629        call_options: CallOptions,
630        request: &Request,
631    ) -> Result<Response, ServiceInvocationError>
632    where
633        T::Wire: UWireEncode<Request> + UWireDecodeOwned<Response>,
634    {
635        let payload_bytes = <T::Wire as EncodePayload<Request>>::encode_payload_owned(request)
636            .map_err(|error| ServiceInvocationError::InvalidArgument(error.to_string()))?;
637        let payload_format = <T::Wire as PayloadCodec>::payload_encoding()
638            .to_legacy_format()
639            .ok_or_else(|| {
640                ServiceInvocationError::InvalidArgument(
641                    "selected wire uses a native-only payload encoding that cannot be sent by P73U2 owned RPC"
642                        .to_string(),
643                )
644            })?;
645        let response = self
646            .invoke_method(
647                method,
648                call_options,
649                Some(UPayload::new(payload_bytes, payload_format)),
650            )
651            .await?
652            .ok_or_else(|| ServiceInvocationError::InvalidArgument("No payload".to_string()))?;
653        <T::Wire as DecodePayload<'_, Response>>::decode_payload(response.payload())
654            .map_err(|error| ServiceInvocationError::InvalidArgument(error.to_string()))
655    }
656}
657
658#[async_trait]
659impl<T, P> crate::communication::RpcClient for RpcClient<T, P>
660where
661    T: UOwnedTransport + ?Sized + 'static,
662    P: LocalUriProvider + ?Sized,
663{
664    async fn invoke_method(
665        &self,
666        method: UUri,
667        call_options: CallOptions,
668        payload: Option<UPayload>,
669    ) -> Result<Option<UPayload>, ServiceInvocationError> {
670        RpcClient::invoke_method(self, method, call_options, payload).await
671    }
672}
673
674struct RequestListener<T>
675where
676    T: UOwnedTransport + ?Sized + 'static,
677{
678    request_handler: Arc<dyn crate::communication::RequestHandler>,
679    transport: Arc<T>,
680}
681
682impl<T> RequestListener<T>
683where
684    T: UOwnedTransport + ?Sized + 'static,
685{
686    async fn process_request(&self, request_frame: UOwnedFrame) {
687        let Ok(request_message) = crate::frame::metadata::try_project_frame_to_umessage(
688            request_frame.metadata().clone(),
689            request_frame.payload().cloned(),
690        ) else {
691            return;
692        };
693        if !request_message.is_request() {
694            return;
695        }
696        let resource_id = request_message.sink_unchecked().resource_id();
697        let payload_format = request_message
698            .payload_format()
699            .unwrap_or(crate::UPayloadFormat::Unspecified);
700        let request_payload = request_message
701            .payload()
702            .map(|payload| UPayload::new(payload, payload_format));
703        let mut response_builder =
704            UMessageBuilder::response_for_request(request_message.attributes());
705        let response = match self
706            .request_handler
707            .handle_request(resource_id, request_message.attributes(), request_payload)
708            .await
709        {
710            Ok(response_payload) => {
711                build_message_with_payload(&mut response_builder, response_payload)
712            }
713            Err(error) => {
714                let status = UStatus::from(error);
715                response_builder
716                    .with_comm_status(status.code())
717                    .build_with_protobuf_payload(&status)
718            }
719        };
720        if let Ok(response_message) = response {
721            if let Ok(response_frame) = frame_from_message(&response_message) {
722                let _ = self.transport.send_owned(response_frame).await;
723            }
724        }
725    }
726}
727
728#[async_trait]
729impl<T> UOwnedListener for RequestListener<T>
730where
731    T: UOwnedTransport + ?Sized + 'static,
732{
733    async fn on_receive_owned(&self, frame: UOwnedFrame) {
734        self.process_request(frame).await;
735    }
736}
737
738/// RPC server implemented over an owned native-frame transport.
739pub struct RpcServer<T, P>
740where
741    T: UOwnedTransport + ?Sized,
742    P: LocalUriProvider + ?Sized,
743{
744    transport: Arc<T>,
745    uri_provider: Arc<P>,
746    request_listeners: tokio::sync::Mutex<std::collections::HashMap<u16, Arc<dyn UOwnedListener>>>,
747}
748
749impl<T, P> core::fmt::Debug for RpcServer<T, P>
750where
751    T: UOwnedTransport + ?Sized,
752    P: LocalUriProvider + ?Sized,
753{
754    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
755        f.debug_struct("RpcServer").finish_non_exhaustive()
756    }
757}
758
759impl<T, P> RpcServer<T, P>
760where
761    T: UOwnedTransport + ?Sized + 'static,
762    P: LocalUriProvider + ?Sized,
763{
764    /// Creates an RPC server over an owned native-frame transport.
765    #[must_use]
766    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
767        Self {
768            transport,
769            uri_provider,
770            request_listeners: tokio::sync::Mutex::new(std::collections::HashMap::new()),
771        }
772    }
773
774    fn validate_sink_filter(filter: &UUri) -> Result<(), RegistrationError> {
775        if !filter.is_rpc_method() {
776            return Err(RegistrationError::InvalidFilter(
777                "RPC endpoint's resource ID must be in range [0x0001, 0x7FFF]".to_string(),
778            ));
779        }
780        Ok(())
781    }
782
783    fn validate_origin_filter(filter: Option<&UUri>) -> Result<(), RegistrationError> {
784        if let Some(uri) = filter {
785            if !uri.is_rpc_response() {
786                return Err(RegistrationError::InvalidFilter(
787                    "origin filter's resource ID must be 0".to_string(),
788                ));
789            }
790        }
791        Ok(())
792    }
793
794    /// Registers an owned RPC endpoint.
795    pub async fn register_endpoint(
796        &self,
797        origin_filter: Option<&UUri>,
798        resource_id: u16,
799        request_handler: Arc<dyn crate::communication::RequestHandler>,
800    ) -> Result<(), RegistrationError> {
801        Self::validate_origin_filter(origin_filter)?;
802        let sink_filter = self.uri_provider.get_resource_uri(resource_id);
803        Self::validate_sink_filter(&sink_filter)?;
804        let mut listener_map = self.request_listeners.lock().await;
805        if listener_map.contains_key(&resource_id) {
806            return Err(RegistrationError::MaxListenersExceeded);
807        }
808        let listener: Arc<dyn UOwnedListener> = Arc::new(RequestListener {
809            request_handler,
810            transport: self.transport.clone(),
811        });
812        self.transport
813            .register_owned_listener(
814                origin_filter.unwrap_or(&UUri::any_with_resource_id(
815                    crate::uri::RESOURCE_ID_RESPONSE,
816                )),
817                Some(&sink_filter),
818                listener.clone(),
819            )
820            .await
821            .map_err(RegistrationError::from)?;
822        listener_map.insert(resource_id, listener);
823        Ok(())
824    }
825
826    /// Unregisters an owned RPC endpoint.
827    pub async fn unregister_endpoint(
828        &self,
829        origin_filter: Option<&UUri>,
830        resource_id: u16,
831        _request_handler: Arc<dyn crate::communication::RequestHandler>,
832    ) -> Result<(), RegistrationError> {
833        Self::validate_origin_filter(origin_filter)?;
834        let sink_filter = self.uri_provider.get_resource_uri(resource_id);
835        Self::validate_sink_filter(&sink_filter)?;
836        let mut listener_map = self.request_listeners.lock().await;
837        let Some(listener) = listener_map.remove(&resource_id) else {
838            return Err(RegistrationError::NoSuchListener);
839        };
840        self.transport
841            .unregister_owned_listener(
842                origin_filter.unwrap_or(&UUri::any_with_resource_id(
843                    crate::uri::RESOURCE_ID_RESPONSE,
844                )),
845                Some(&sink_filter),
846                listener,
847            )
848            .await
849            .map_err(RegistrationError::from)
850    }
851}
852
853#[async_trait]
854impl<T, P> crate::communication::RpcServer for RpcServer<T, P>
855where
856    T: UOwnedTransport + ?Sized + 'static,
857    P: LocalUriProvider + ?Sized,
858{
859    async fn register_endpoint(
860        &self,
861        origin_filter: Option<&UUri>,
862        resource_id: u16,
863        request_handler: Arc<dyn crate::communication::RequestHandler>,
864    ) -> Result<(), RegistrationError> {
865        RpcServer::register_endpoint(self, origin_filter, resource_id, request_handler).await
866    }
867
868    async fn unregister_endpoint(
869        &self,
870        origin_filter: Option<&UUri>,
871        resource_id: u16,
872        request_handler: Arc<dyn crate::communication::RequestHandler>,
873    ) -> Result<(), RegistrationError> {
874        RpcServer::unregister_endpoint(self, origin_filter, resource_id, request_handler).await
875    }
876}
877
878/// Publisher implemented over an owned native-frame transport.
879pub struct Publisher<T, P>
880where
881    T: UOwnedTransport + ?Sized,
882    P: LocalUriProvider + ?Sized,
883{
884    transport: Arc<T>,
885    uri_provider: Arc<P>,
886}
887
888impl<T, P> core::fmt::Debug for Publisher<T, P>
889where
890    T: UOwnedTransport + ?Sized,
891    P: LocalUriProvider + ?Sized,
892{
893    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
894        f.debug_struct("Publisher").finish_non_exhaustive()
895    }
896}
897
898impl<T, P> Publisher<T, P>
899where
900    T: UOwnedTransport + ?Sized,
901    P: LocalUriProvider + ?Sized,
902{
903    /// Creates a publisher over an owned native-frame transport.
904    #[must_use]
905    pub fn new(transport: Arc<T>, uri_provider: Arc<P>) -> Self {
906        Self {
907            transport,
908            uri_provider,
909        }
910    }
911
912    fn build_frame(
913        &self,
914        resource_id: u16,
915        call_options: CallOptions,
916        payload: Option<UPayload>,
917    ) -> Result<UOwnedFrame, PubSubError> {
918        let mut builder = UMessageBuilder::publish(self.uri_provider.get_resource_uri(resource_id));
919        builder.with_ttl(call_options.ttl());
920        if let Some(message_id) = call_options.message_id() {
921            builder.with_message_id(message_id.clone());
922        }
923        if let Some(priority) = call_options.priority() {
924            builder.with_priority(priority);
925        }
926        let message = match payload {
927            Some(payload) => {
928                builder.build_with_payload(payload.payload().clone(), payload.payload_format())
929            }
930            None => builder.build(),
931        }
932        .map_err(|error| {
933            PubSubError::InvalidArgument(format!(
934                "failed to create Publish message from parameters: {error}"
935            ))
936        })?;
937        let metadata = crate::frame::metadata::try_project_umessage_to_frame_metadata(&message)
938            .map_err(|error| {
939                PubSubError::InvalidArgument(format!(
940                    "failed to create owned Publish frame metadata from parameters: {error}"
941                ))
942            })?;
943        UOwnedFrame::new(metadata, message.payload()).map_err(|error| {
944            PubSubError::InvalidArgument(format!(
945                "failed to create owned Publish frame from parameters: {error}"
946            ))
947        })
948    }
949
950    /// Publishes an owned native-frame message.
951    pub async fn publish(
952        &self,
953        resource_id: u16,
954        call_options: CallOptions,
955        payload: Option<UPayload>,
956    ) -> Result<(), PubSubError> {
957        let frame = self.build_frame(resource_id, call_options, payload)?;
958        self.transport
959            .send_owned(frame)
960            .await
961            .map_err(Box::from)
962            .map_err(PubSubError::PublishError)
963    }
964}
965
966#[cfg(feature = "selected-wire-transport-adapter")]
967impl<T, P> Publisher<T, P>
968where
969    T: UOwnedTransport + UHasWire + ?Sized,
970    P: LocalUriProvider + ?Sized,
971{
972    /// Publishes a typed payload using the transport endpoint's selected wire.
973    pub async fn publish_typed<Payload>(
974        &self,
975        resource_id: u16,
976        call_options: CallOptions,
977        payload: &Payload,
978    ) -> Result<(), PubSubError>
979    where
980        T::Wire: UWireEncode<Payload>,
981    {
982        let payload_bytes = <T::Wire as EncodePayload<Payload>>::encode_payload_owned(payload)
983            .map_err(|error| PubSubError::InvalidArgument(error.to_string()))?;
984        let payload_format = <T::Wire as PayloadCodec>::payload_encoding()
985            .to_legacy_format()
986            .ok_or_else(|| {
987                PubSubError::InvalidArgument(
988                    "selected wire uses a native-only payload encoding that cannot be sent by P73U1 owned publish"
989                        .to_string(),
990                )
991            })?;
992        self.publish(
993            resource_id,
994            call_options,
995            Some(UPayload::new(payload_bytes, payload_format)),
996        )
997        .await
998    }
999}
1000
1001#[async_trait]
1002impl<T, P> crate::communication::Publisher for Publisher<T, P>
1003where
1004    T: UOwnedTransport + ?Sized,
1005    P: LocalUriProvider + ?Sized,
1006{
1007    async fn publish(
1008        &self,
1009        resource_id: u16,
1010        call_options: CallOptions,
1011        payload: Option<UPayload>,
1012    ) -> Result<(), PubSubError> {
1013        Publisher::publish(self, resource_id, call_options, payload).await
1014    }
1015}
1016
1017#[cfg(test)]
1018mod tests {
1019    use std::collections::VecDeque;
1020    use std::sync::Mutex;
1021
1022    use super::*;
1023    use crate::{
1024        StaticUriProvider, UAttributes, UCode, UOwnedFrame, UOwnedTransportImpl, UPayloadFormat,
1025        UStatus,
1026    };
1027
1028    struct RecordingOwnedTransport {
1029        sent: Mutex<Vec<UOwnedFrame>>,
1030        received: Mutex<VecDeque<UOwnedFrame>>,
1031        listener: Mutex<Option<Arc<dyn UOwnedListener>>>,
1032        lifecycle: Option<Arc<Mutex<Vec<&'static str>>>>,
1033    }
1034
1035    impl RecordingOwnedTransport {
1036        fn new() -> Self {
1037            Self {
1038                sent: Mutex::new(Vec::new()),
1039                received: Mutex::new(VecDeque::new()),
1040                listener: Mutex::new(None),
1041                lifecycle: None,
1042            }
1043        }
1044
1045        fn with_lifecycle(lifecycle: Arc<Mutex<Vec<&'static str>>>) -> Self {
1046            Self {
1047                lifecycle: Some(lifecycle),
1048                ..Self::new()
1049            }
1050        }
1051
1052        fn record_lifecycle(&self, event: &'static str) {
1053            if let Some(lifecycle) = &self.lifecycle {
1054                lifecycle
1055                    .lock()
1056                    .expect("lifecycle lock poisoned")
1057                    .push(event);
1058            }
1059        }
1060
1061        fn sent_frames(&self) -> Vec<UOwnedFrame> {
1062            self.sent.lock().expect("sent lock poisoned").clone()
1063        }
1064
1065        fn push_received(&self, frame: UOwnedFrame) {
1066            self.received
1067                .lock()
1068                .expect("received lock poisoned")
1069                .push_back(frame);
1070        }
1071
1072        fn registered_listener(&self) -> Arc<dyn UOwnedListener> {
1073            self.listener
1074                .lock()
1075                .expect("listener lock poisoned")
1076                .as_ref()
1077                .expect("listener registered")
1078                .clone()
1079        }
1080    }
1081
1082    #[async_trait]
1083    impl UOwnedTransportImpl for RecordingOwnedTransport {
1084        async fn send_validated_owned(&self, frame: UOwnedFrame) -> Result<(), UStatus> {
1085            self.sent.lock().expect("sent lock poisoned").push(frame);
1086            Ok(())
1087        }
1088
1089        async fn receive_validated_owned(
1090            &self,
1091            _source_filter: &UUri,
1092            _sink_filter: Option<&UUri>,
1093        ) -> Result<UOwnedFrame, UStatus> {
1094            self.received
1095                .lock()
1096                .expect("received lock poisoned")
1097                .pop_front()
1098                .ok_or_else(|| UStatus::fail_with_code(UCode::NotFound, "no queued frame"))
1099        }
1100
1101        async fn register_validated_owned_listener(
1102            &self,
1103            _source_filter: &UUri,
1104            _sink_filter: Option<&UUri>,
1105            listener: Arc<dyn UOwnedListener>,
1106        ) -> Result<(), UStatus> {
1107            self.record_lifecycle("register");
1108            *self.listener.lock().expect("listener lock poisoned") = Some(listener);
1109            Ok(())
1110        }
1111
1112        async fn unregister_validated_owned_listener(
1113            &self,
1114            _source_filter: &UUri,
1115            _sink_filter: Option<&UUri>,
1116            _listener: Arc<dyn UOwnedListener>,
1117        ) -> Result<(), UStatus> {
1118            self.record_lifecycle("unregister");
1119            *self.listener.lock().expect("listener lock poisoned") = None;
1120            Ok(())
1121        }
1122    }
1123
1124    fn uri_provider() -> Arc<StaticUriProvider> {
1125        Arc::new(StaticUriProvider::new("", 0x0005, 0x02).expect("uri provider"))
1126    }
1127
1128    fn method_uri() -> UUri {
1129        uri_provider().get_resource_uri(0x1000)
1130    }
1131
1132    fn message_frame(message: &UMessage) -> UOwnedFrame {
1133        frame_from_message(message).expect("frame")
1134    }
1135
1136    #[derive(Default)]
1137    struct RecordingMessageListener {
1138        messages: Mutex<Vec<UMessage>>,
1139    }
1140
1141    impl RecordingMessageListener {
1142        fn messages(&self) -> Vec<UMessage> {
1143            self.messages
1144                .lock()
1145                .expect("messages lock poisoned")
1146                .clone()
1147        }
1148    }
1149
1150    #[async_trait]
1151    impl UListener for RecordingMessageListener {
1152        async fn on_receive(&self, msg: UMessage) {
1153            self.messages
1154                .lock()
1155                .expect("messages lock poisoned")
1156                .push(msg);
1157        }
1158    }
1159
1160    #[tokio::test]
1161    async fn publish_sends_owned_frame() {
1162        let transport = Arc::new(RecordingOwnedTransport::new());
1163        let publisher = Publisher::new(transport.clone(), uri_provider());
1164
1165        publisher
1166            .publish(
1167                0x9A00,
1168                CallOptions::for_publish(None, None, None),
1169                Some(UPayload::new("hello", UPayloadFormat::Text)),
1170            )
1171            .await
1172            .expect("publish succeeds");
1173
1174        let frames = transport.sent_frames();
1175        assert_eq!(frames.len(), 1);
1176        let frame = frames.first().expect("one sent frame");
1177        assert_eq!(frame.payload(), Some(&bytes::Bytes::from_static(b"hello")));
1178        assert_eq!(
1179            frame
1180                .metadata()
1181                .payload_encoding()
1182                .and_then(crate::PayloadEncoding::to_legacy_format),
1183            Some(UPayloadFormat::Text)
1184        );
1185    }
1186
1187    #[tokio::test]
1188    async fn publish_rejects_invalid_topic() {
1189        let transport = Arc::new(RecordingOwnedTransport::new());
1190        let publisher = Publisher::new(transport.clone(), uri_provider());
1191
1192        let result = publisher
1193            .publish(0x1000, CallOptions::for_publish(None, None, None), None)
1194            .await;
1195
1196        assert!(matches!(result, Err(PubSubError::InvalidArgument(_))));
1197        assert!(transport.sent_frames().is_empty());
1198    }
1199
1200    struct FailingOwnedTransport;
1201
1202    #[async_trait]
1203    impl UOwnedTransportImpl for FailingOwnedTransport {
1204        async fn send_validated_owned(&self, _frame: UOwnedFrame) -> Result<(), UStatus> {
1205            Err(UStatus::fail_with_code(
1206                UCode::Unavailable,
1207                "transport unavailable",
1208            ))
1209        }
1210    }
1211
1212    #[tokio::test]
1213    async fn publish_maps_transport_error() {
1214        let publisher = Publisher::new(Arc::new(FailingOwnedTransport), uri_provider());
1215
1216        let result = publisher
1217            .publish(0x9A00, CallOptions::for_publish(None, None, None), None)
1218            .await;
1219
1220        assert!(matches!(result, Err(PubSubError::PublishError(_))));
1221    }
1222
1223    #[derive(Default)]
1224    struct StubUSubscription {
1225        lifecycle: Option<Arc<Mutex<Vec<&'static str>>>>,
1226    }
1227
1228    impl StubUSubscription {
1229        fn recording(lifecycle: Arc<Mutex<Vec<&'static str>>>) -> Self {
1230            Self {
1231                lifecycle: Some(lifecycle),
1232            }
1233        }
1234
1235        fn record_lifecycle(&self, event: &'static str) {
1236            if let Some(lifecycle) = &self.lifecycle {
1237                lifecycle
1238                    .lock()
1239                    .expect("lifecycle lock poisoned")
1240                    .push(event);
1241            }
1242        }
1243    }
1244
1245    #[async_trait]
1246    impl crate::core::usubscription::USubscription for StubUSubscription {
1247        async fn subscribe(
1248            &self,
1249            _topic: &UUri,
1250            _expiration: Option<u64>,
1251            _min_sample_period: Option<u32>,
1252        ) -> Result<SubscriptionStatus, crate::UStatus> {
1253            self.record_lifecycle("subscribe");
1254            Ok(SubscriptionStatus::Subscribed)
1255        }
1256        async fn unsubscribe(&self, _topic: &UUri) -> Result<(), crate::UStatus> {
1257            self.record_lifecycle("unsubscribe");
1258            Ok(())
1259        }
1260        async fn fetch_subscriptions_by_topic(
1261            &self,
1262            _topic: &UUri,
1263        ) -> Result<Vec<crate::core::usubscription::SubscriptionInfo>, crate::UStatus> {
1264            Ok(Vec::new())
1265        }
1266        async fn fetch_subscriptions_by_subscriber(
1267            &self,
1268            _subscriber: &UUri,
1269        ) -> Result<Vec<crate::core::usubscription::SubscriptionInfo>, crate::UStatus> {
1270            Ok(Vec::new())
1271        }
1272        async fn register_for_notifications(&self, _topic: &UUri) -> Result<(), crate::UStatus> {
1273            Ok(())
1274        }
1275        async fn unregister_for_notifications(&self, _topic: &UUri) -> Result<(), crate::UStatus> {
1276            Ok(())
1277        }
1278        async fn fetch_subscribers(&self, _topic: &UUri) -> Result<Vec<UUri>, crate::UStatus> {
1279            Ok(Vec::new())
1280        }
1281        async fn reset(
1282            &self,
1283            _reason: crate::core::usubscription::ResetReason,
1284            _message: Option<String>,
1285        ) -> Result<(), crate::UStatus> {
1286            Ok(())
1287        }
1288    }
1289
1290    #[tokio::test]
1291    async fn subscriber_registers_owned_listener_and_delivers_messages() {
1292        let transport = Arc::new(RecordingOwnedTransport::new());
1293        let subscriber = Endpoint::new(transport.clone(), uri_provider())
1294            .subscriber(Arc::new(StubUSubscription::default()));
1295        let topic = uri_provider().get_resource_uri(0x9A00);
1296        let listener = Arc::new(RecordingMessageListener::default());
1297
1298        subscriber
1299            .subscribe(&topic, listener.clone(), None)
1300            .await
1301            .expect("subscriber registered");
1302        let message = UMessageBuilder::publish(topic.clone())
1303            .build_with_payload("event", UPayloadFormat::Text)
1304            .expect("publish message");
1305        transport
1306            .registered_listener()
1307            .on_receive_owned(message_frame(&message))
1308            .await;
1309
1310        let messages = listener.messages();
1311        assert_eq!(messages.len(), 1);
1312        let message = messages.first().expect("one received message");
1313        assert!(message.is_publish());
1314        assert_eq!(message.payload(), Some(bytes::Bytes::from_static(b"event")));
1315
1316        subscriber
1317            .unsubscribe(&topic, listener)
1318            .await
1319            .expect("subscriber unregistered");
1320    }
1321
1322    #[tokio::test]
1323    async fn subscriber_contacts_usubscription_before_listener_changes() {
1324        let lifecycle = Arc::new(Mutex::new(Vec::new()));
1325        let transport = Arc::new(RecordingOwnedTransport::with_lifecycle(lifecycle.clone()));
1326        let subscriber = Endpoint::new(transport, uri_provider())
1327            .subscriber(Arc::new(StubUSubscription::recording(lifecycle.clone())));
1328        let topic = uri_provider().get_resource_uri(0x9A00);
1329        let listener = Arc::new(RecordingMessageListener::default());
1330
1331        subscriber
1332            .subscribe(&topic, listener.clone(), None)
1333            .await
1334            .expect("subscriber registered");
1335        subscriber
1336            .unsubscribe(&topic, listener)
1337            .await
1338            .expect("subscriber unregistered");
1339
1340        assert_eq!(
1341            *lifecycle.lock().expect("lifecycle lock poisoned"),
1342            ["subscribe", "register", "unsubscribe", "unregister"]
1343        );
1344    }
1345
1346    #[tokio::test]
1347    async fn notifier_sends_and_listens_with_owned_frames() {
1348        let transport = Arc::new(RecordingOwnedTransport::new());
1349        let notifier = Endpoint::new(transport.clone(), uri_provider()).notifier();
1350        let topic = uri_provider().get_resource_uri(0xD100);
1351        let listener = Arc::new(RecordingMessageListener::default());
1352
1353        notifier
1354            .start_listening(&topic, listener.clone())
1355            .await
1356            .expect("notifier listener registered");
1357        let notification =
1358            UMessageBuilder::notification(topic.clone(), uri_provider().get_source_uri())
1359                .build_with_payload("notification", UPayloadFormat::Text)
1360                .expect("notification message");
1361        transport
1362            .registered_listener()
1363            .on_receive_owned(message_frame(&notification))
1364            .await;
1365
1366        notifier
1367            .notify(
1368                0xD100,
1369                &uri_provider().get_source_uri(),
1370                CallOptions::for_notification(None, None, None),
1371                Some(UPayload::new("notify", UPayloadFormat::Text)),
1372            )
1373            .await
1374            .expect("notification sent");
1375
1376        let messages = listener.messages();
1377        assert_eq!(messages.len(), 1);
1378        let message = messages.first().expect("one received message");
1379        assert!(message.is_notification());
1380        assert_eq!(
1381            message.payload(),
1382            Some(bytes::Bytes::from_static(b"notification"))
1383        );
1384        let frames = transport.sent_frames();
1385        assert_eq!(frames.len(), 1);
1386        let sent = message_from_frame(frames.first().expect("one sent frame").clone())
1387            .expect("sent notification");
1388        assert!(sent.is_notification());
1389        assert_eq!(sent.payload(), Some(bytes::Bytes::from_static(b"notify")));
1390
1391        notifier
1392            .stop_listening(&topic, listener)
1393            .await
1394            .expect("notifier listener unregistered");
1395    }
1396
1397    #[tokio::test]
1398    async fn rpc_client_sends_request_and_returns_response_payload() {
1399        let transport = Arc::new(RecordingOwnedTransport::new());
1400        let client = RpcClient::new(transport.clone(), uri_provider());
1401        let request_id = UUID::build();
1402        let request =
1403            UMessageBuilder::request(method_uri(), uri_provider().get_source_uri(), 5_000)
1404                .with_message_id(request_id.clone())
1405                .build_with_payload("request", UPayloadFormat::Text)
1406                .expect("request");
1407        let response = UMessageBuilder::response_for_request(request.attributes())
1408            .build_with_payload("response", UPayloadFormat::Text)
1409            .expect("response");
1410        transport.push_received(message_frame(&response));
1411
1412        let result = client
1413            .invoke_method(
1414                method_uri(),
1415                CallOptions::for_rpc_request(5_000, Some(request_id), None, None),
1416                Some(UPayload::new("request", UPayloadFormat::Text)),
1417            )
1418            .await
1419            .expect("rpc succeeds")
1420            .expect("response payload");
1421
1422        assert_eq!(result.payload(), &bytes::Bytes::from_static(b"response"));
1423        let frames = transport.sent_frames();
1424        assert_eq!(frames.len(), 1);
1425        let frame = frames.first().expect("one sent frame");
1426        let sent = crate::frame::metadata::try_project_frame_to_umessage(
1427            frame.metadata().clone(),
1428            frame.payload().cloned(),
1429        )
1430        .expect("sent request");
1431        assert!(sent.is_request());
1432        assert_eq!(sent.payload(), Some(bytes::Bytes::from_static(b"request")));
1433    }
1434
1435    struct EchoHandler;
1436
1437    #[async_trait]
1438    impl crate::communication::RequestHandler for EchoHandler {
1439        async fn handle_request(
1440            &self,
1441            resource_id: u16,
1442            message_attributes: &UAttributes,
1443            request_payload: Option<UPayload>,
1444        ) -> Result<Option<UPayload>, ServiceInvocationError> {
1445            assert_eq!(resource_id, 0x1000);
1446            assert!(message_attributes.is_request());
1447            Ok(request_payload)
1448        }
1449    }
1450
1451    #[tokio::test]
1452    async fn rpc_server_registers_listener_and_sends_response() {
1453        let transport = Arc::new(RecordingOwnedTransport::new());
1454        let server = RpcServer::new(transport.clone(), uri_provider());
1455        server
1456            .register_endpoint(None, 0x1000, Arc::new(EchoHandler))
1457            .await
1458            .expect("endpoint registered");
1459        let request =
1460            UMessageBuilder::request(method_uri(), uri_provider().get_source_uri(), 5_000)
1461                .build_with_payload("server request", UPayloadFormat::Text)
1462                .expect("request");
1463
1464        transport
1465            .registered_listener()
1466            .on_receive_owned(message_frame(&request))
1467            .await;
1468
1469        let frames = transport.sent_frames();
1470        assert_eq!(frames.len(), 1);
1471        let frame = frames.first().expect("one sent frame");
1472        let response = crate::frame::metadata::try_project_frame_to_umessage(
1473            frame.metadata().clone(),
1474            frame.payload().cloned(),
1475        )
1476        .expect("response");
1477        assert!(response.is_response());
1478        assert_eq!(
1479            response.payload(),
1480            Some(bytes::Bytes::from_static(b"server request"))
1481        );
1482        assert_eq!(response.request_id_unchecked(), request.id());
1483    }
1484}
1485
1486#[cfg(all(test, feature = "protobuf-support"))]
1487mod selected_wire_tests {
1488    use std::collections::VecDeque;
1489    use std::sync::Mutex;
1490
1491    use protobuf::well_known_types::wrappers::StringValue;
1492
1493    use super::*;
1494    use crate::{
1495        NativePrefixFrameMetadataCodec, ProtobufWire, UHasWire, UOwnedFrame, UOwnedTransportCore,
1496        UOwnedTransportImpl, UStatus, UWithNativePrefixWire,
1497    };
1498
1499    #[derive(Clone, Default)]
1500    struct RecordingOwnedCore {
1501        sent: Arc<Mutex<Vec<crate::PreparedOwnedFrame>>>,
1502    }
1503
1504    impl RecordingOwnedCore {
1505        fn sent_payloads(&self) -> Vec<Option<bytes::Bytes>> {
1506            self.sent
1507                .lock()
1508                .expect("sent lock poisoned")
1509                .iter()
1510                .map(|frame| frame.payload().cloned())
1511                .collect()
1512        }
1513    }
1514
1515    #[async_trait]
1516    impl UOwnedTransportCore for RecordingOwnedCore {
1517        async fn send_prepared_owned(
1518            &self,
1519            frame: crate::PreparedOwnedFrame,
1520        ) -> Result<(), UStatus> {
1521            self.sent.lock().expect("sent lock poisoned").push(frame);
1522            Ok(())
1523        }
1524    }
1525
1526    fn uri_provider() -> Arc<crate::StaticUriProvider> {
1527        Arc::new(crate::StaticUriProvider::new("", 0x0005, 0x02).expect("uri provider"))
1528    }
1529
1530    #[tokio::test]
1531    async fn publish_typed_uses_selected_wire() {
1532        let core = RecordingOwnedCore::default();
1533        let transport = Arc::new(core.clone().into_native_prefix_wire_transport(ProtobufWire));
1534        assert_eq!(transport.wire(), &ProtobufWire);
1535        let publisher = Publisher::new(transport, uri_provider());
1536        let payload = StringValue {
1537            value: "typed".to_string(),
1538            ..Default::default()
1539        };
1540
1541        publisher
1542            .publish_typed(0x9A00, CallOptions::for_publish(None, None, None), &payload)
1543            .await
1544            .expect("typed publish succeeds");
1545
1546        let payloads = core.sent_payloads();
1547        assert_eq!(payloads.len(), 1);
1548        assert!(payloads
1549            .first()
1550            .expect("one sent payload")
1551            .as_ref()
1552            .is_some_and(|payload| !payload.is_empty()));
1553        let _: NativePrefixFrameMetadataCodec = NativePrefixFrameMetadataCodec;
1554    }
1555
1556    struct DirectSelectedOwnedTransport {
1557        wire: ProtobufWire,
1558        sent: Mutex<Vec<UOwnedFrame>>,
1559        received: Mutex<VecDeque<UOwnedFrame>>,
1560    }
1561
1562    impl DirectSelectedOwnedTransport {
1563        fn new() -> Self {
1564            Self {
1565                wire: ProtobufWire,
1566                sent: Mutex::new(Vec::new()),
1567                received: Mutex::new(VecDeque::new()),
1568            }
1569        }
1570
1571        fn sent_frames(&self) -> Vec<UOwnedFrame> {
1572            self.sent.lock().expect("sent lock poisoned").clone()
1573        }
1574
1575        fn push_received(&self, frame: UOwnedFrame) {
1576            self.received
1577                .lock()
1578                .expect("received lock poisoned")
1579                .push_back(frame);
1580        }
1581    }
1582
1583    impl UHasWire for DirectSelectedOwnedTransport {
1584        type Wire = ProtobufWire;
1585
1586        fn wire(&self) -> &Self::Wire {
1587            &self.wire
1588        }
1589    }
1590
1591    #[async_trait]
1592    impl UOwnedTransportImpl for DirectSelectedOwnedTransport {
1593        async fn send_validated_owned(&self, frame: UOwnedFrame) -> Result<(), UStatus> {
1594            self.sent.lock().expect("sent lock poisoned").push(frame);
1595            Ok(())
1596        }
1597
1598        async fn receive_validated_owned(
1599            &self,
1600            _source_filter: &UUri,
1601            _sink_filter: Option<&UUri>,
1602        ) -> Result<UOwnedFrame, UStatus> {
1603            self.received
1604                .lock()
1605                .expect("received lock poisoned")
1606                .pop_front()
1607                .ok_or_else(|| crate::UStatus::fail_with_code(crate::UCode::NotFound, "no frame"))
1608        }
1609    }
1610
1611    #[tokio::test]
1612    async fn invoke_typed_uses_selected_wire() {
1613        let transport = Arc::new(DirectSelectedOwnedTransport::new());
1614        assert_eq!(transport.wire(), &ProtobufWire);
1615        let client = RpcClient::new(transport.clone(), uri_provider());
1616        let method = uri_provider().get_resource_uri(0x1000);
1617        let request_id = crate::UUID::build();
1618        let request =
1619            UMessageBuilder::request(method.clone(), uri_provider().get_source_uri(), 5_000)
1620                .with_message_id(request_id.clone())
1621                .build()
1622                .expect("request");
1623        let response_value = StringValue {
1624            value: "typed response".to_string(),
1625            ..Default::default()
1626        };
1627        let response_bytes =
1628            <ProtobufWire as EncodePayload<StringValue>>::encode_payload_owned(&response_value)
1629                .expect("encoded response");
1630        let response_format = <ProtobufWire as PayloadCodec>::payload_encoding()
1631            .to_legacy_format()
1632            .expect("standard format");
1633        let response = UMessageBuilder::response_for_request(request.attributes())
1634            .build_with_payload(response_bytes, response_format)
1635            .expect("response");
1636        transport.push_received(frame_from_message(&response).expect("response frame"));
1637        let request_value = StringValue {
1638            value: "typed request".to_string(),
1639            ..Default::default()
1640        };
1641
1642        let result: StringValue = client
1643            .invoke_typed(
1644                method,
1645                CallOptions::for_rpc_request(5_000, Some(request_id), None, None),
1646                &request_value,
1647            )
1648            .await
1649            .expect("typed rpc succeeds");
1650
1651        assert_eq!(result.value, "typed response");
1652        let frames = transport.sent_frames();
1653        assert_eq!(frames.len(), 1);
1654        assert!(frames
1655            .first()
1656            .expect("one sent frame")
1657            .payload()
1658            .is_some_and(|payload| !payload.is_empty()));
1659    }
1660}
up_rust/communication/owned.rs

up_rust/communication/
owned.rs
