fdomain_local/
lib.rs

1// Copyright 2024 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use fdomain_client::{Channel, Client, FDomainTransport};
6use fdomain_container::FDomain;
7use fdomain_container::wire::FDomainCodec;
8use fidl::endpoints::ClientEnd;
9use fidl_fuchsia_io as fio;
10use futures::StreamExt;
11use futures::stream::Stream;
12use std::pin::Pin;
13use std::sync::{Arc, OnceLock, Weak};
14use std::task::{Context, Poll};
15
16/// An FDomain that is designed to be used in the same process it was created
17/// in. I.e. no networking, just a bucket of handles right here where you can
18/// use them.
19struct LocalFDomain(FDomainCodec);
20
21impl LocalFDomain {
22    /// Create a new FDomain client that points to a new local FDomain.
23    fn new_client(
24        namespace: impl Fn() -> Result<ClientEnd<fio::DirectoryMarker>, fidl::Status> + Send + 'static,
25    ) -> Arc<Client> {
26        let (client, fut) = Client::new(LocalFDomain(FDomainCodec::new(FDomain::new(namespace))));
27        fuchsia_async::Task::spawn(fut).detach();
28        client
29    }
30
31    /// Create a new FDomain client that points to a new local FDomain with an
32    /// FDomain channel callback for the namespace.
33    fn new_client_fdomain(namespace: impl Fn(Channel) + Send + 'static) -> Arc<Client> {
34        let (sender, mut receiver) = futures::channel::mpsc::unbounded();
35        let client_holder = Arc::new(OnceLock::<Weak<Client>>::new());
36        let client_holder_clone = Arc::clone(&client_holder);
37        let fdomain = FDomain::new_with_namespace_channel(move |server_hid| {
38            if let Some(client_weak) = client_holder_clone.get() {
39                if let Some(client) = client_weak.upgrade() {
40                    let channel = client.channel_from_handle_id(server_hid);
41                    let _ = sender.unbounded_send(channel);
42                }
43            }
44        });
45        let (client, fut) = Client::new(LocalFDomain(FDomainCodec::new(fdomain)));
46        let _ = client_holder.set(Arc::downgrade(&client));
47        fuchsia_async::Task::spawn(fut).detach();
48        fuchsia_async::Task::spawn(async move {
49            while let Some(channel) = receiver.next().await {
50                namespace(channel);
51            }
52        })
53        .detach();
54        client
55    }
56}
57
58impl FDomainTransport for LocalFDomain {
59    fn poll_send_message(
60        mut self: Pin<&mut Self>,
61        msg: &[u8],
62        _ctx: &mut Context<'_>,
63    ) -> Poll<Result<(), Option<std::io::Error>>> {
64        Poll::Ready(self.0.message(msg).map_err(|x| Some(std::io::Error::other(x))))
65    }
66}
67
68impl Stream for LocalFDomain {
69    type Item = Result<Box<[u8]>, std::io::Error>;
70
71    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
72        Pin::new(&mut self.0).poll_next(cx).map_err(std::io::Error::other)
73    }
74}
75
76/// Create a new FDomain client that points to a new local FDomain.
77pub fn local_client(
78    namespace: impl Fn() -> Result<ClientEnd<fio::DirectoryMarker>, fidl::Status> + Send + 'static,
79) -> Arc<Client> {
80    LocalFDomain::new_client(namespace)
81}
82
83/// Create a new FDomain client that points to a new local FDomain using a pure
84/// FDomain channel callback to serve the namespace.
85pub fn local_client_fdomain(namespace: impl Fn(Channel) + Send + 'static) -> Arc<Client> {
86    LocalFDomain::new_client_fdomain(namespace)
87}
88
89pub fn local_client_empty() -> Arc<Client> {
90    local_client(|| Err(fidl::Status::NOT_SUPPORTED))
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96    use futures::StreamExt;
97
98    #[fuchsia::test]
99    async fn test_local_client_empty() {
100        let client = local_client_empty();
101        assert!(client.namespace().await.is_err());
102    }
103
104    #[fuchsia::test]
105    async fn test_local_client_native() {
106        let client = local_client(|| {
107            let (client_end, server_end) =
108                fidl::endpoints::create_endpoints::<fio::DirectoryMarker>();
109            fuchsia_async::Task::spawn(async move {
110                let mut stream = server_end.into_stream();
111                while let Some(Ok(_)) = stream.next().await {}
112            })
113            .detach();
114            Ok(client_end)
115        });
116
117        let ns = client.namespace().await.unwrap();
118        assert!(!ns.is_invalid());
119    }
120
121    #[fuchsia::test]
122    async fn test_local_client_fdomain() {
123        let (send_tx, mut send_rx) = futures::channel::mpsc::unbounded::<Vec<u8>>();
124        let client = local_client_fdomain(move |channel| {
125            let send_tx = send_tx.clone();
126            fuchsia_async::Task::spawn(async move {
127                let (mut stream, writer) = channel.stream().unwrap();
128                while let Some(Ok(msg)) = stream.next().await {
129                    let _ = send_tx.unbounded_send(msg.bytes.clone());
130                    let _ = writer.write(b"pong", vec![]);
131                }
132            })
133            .detach();
134        });
135
136        let ns = client.namespace().await.unwrap();
137        ns.write(b"ping", vec![]).unwrap();
138        assert_eq!(send_rx.next().await.unwrap(), b"ping");
139        let (mut stream, _) = ns.stream().unwrap();
140        let reply = stream.next().await.unwrap().unwrap();
141        assert_eq!(&reply.bytes, b"pong");
142    }
143
144    #[fuchsia::test]
145    async fn test_local_client_fdomain_sync_callback() {
146        let client = local_client_fdomain(|channel| {
147            channel.write(b"hello", vec![]).unwrap();
148        });
149
150        let ns = client.namespace().await.unwrap();
151        let (mut stream, _) = ns.stream().unwrap();
152        let reply = stream.next().await.unwrap().unwrap();
153        assert_eq!(&reply.bytes, b"hello");
154    }
155}
fdomain_local/lib.rs

fdomain_local/
lib.rs
