vfs/directory/immutable/
connection.rs1use crate::directory::connection::{BaseConnection, ConnectionState};
9use crate::directory::entry_container::Directory;
10use crate::execution_scope::ExecutionScope;
11use crate::node::OpenNode;
12use crate::object_request::ConnectionCreator;
13use crate::request_handler::{RequestHandler, RequestListener};
14use crate::{ObjectRequestRef, ProtocolsExt};
15
16use fidl_fuchsia_io as fio;
17use fio::DirectoryRequest;
18use std::ops::ControlFlow;
19use std::pin::Pin;
20use std::sync::Arc;
21use zx_status::Status;
22
23pub struct ImmutableConnection<DirectoryType: Directory> {
24 base: BaseConnection<DirectoryType>,
25}
26
27impl<DirectoryType: Directory> ImmutableConnection<DirectoryType> {
28 pub async fn create(
34 scope: ExecutionScope,
35 directory: Arc<DirectoryType>,
36 protocols: impl ProtocolsExt,
37 object_request: ObjectRequestRef<'_>,
38 ) -> Result<(), Status> {
39 Self::create_transform_stream(
40 scope,
41 directory,
42 protocols,
43 object_request,
44 std::convert::identity,
45 )
46 .await
47 }
48
49 pub async fn create_transform_stream<Transform, RS>(
54 scope: ExecutionScope,
55 directory: Arc<DirectoryType>,
56 protocols: impl ProtocolsExt,
57 object_request: ObjectRequestRef<'_>,
58 transform: Transform,
59 ) -> Result<(), Status>
60 where
61 Transform: FnOnce(fio::DirectoryRequestStream) -> RS,
62 RS: futures::stream::Stream<Item = Result<DirectoryRequest, fidl::Error>> + Send + 'static,
63 {
64 let directory = OpenNode::new(directory);
66
67 let connection = ImmutableConnection {
68 base: BaseConnection::new(scope.clone(), directory, protocols.to_directory_options()?),
69 };
70
71 if let Ok(requests) = object_request.take().into_request_stream(&connection.base).await {
76 scope.spawn(RequestListener::new(transform(requests), connection));
77 }
78 Ok(())
79 }
80}
81
82impl<DirectoryType: Directory> RequestHandler for ImmutableConnection<DirectoryType> {
83 type Request = Result<DirectoryRequest, fidl::Error>;
84
85 async fn handle_request(self: Pin<&mut Self>, request: Self::Request) -> ControlFlow<()> {
86 let this = self.get_mut();
87 let _guard = this.base.scope.active_guard();
88 match request {
89 Ok(request) => match this.base.handle_request(request).await {
90 Ok(ConnectionState::Alive) => ControlFlow::Continue(()),
91 Ok(ConnectionState::Closed) | Err(_) => ControlFlow::Break(()),
92 },
93 Err(_) => ControlFlow::Break(()),
94 }
95 }
96}
97
98impl<DirectoryType: Directory> ConnectionCreator<DirectoryType>
99 for ImmutableConnection<DirectoryType>
100{
101 async fn create<'a>(
102 scope: ExecutionScope,
103 node: Arc<DirectoryType>,
104 protocols: impl ProtocolsExt,
105 object_request: ObjectRequestRef<'a>,
106 ) -> Result<(), Status> {
107 Self::create(scope, node, protocols, object_request).await
108 }
109}