fdomain_client/
channel.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 crate::handle::handle_type;
6use crate::responder::Responder;
7use crate::{Error, Event, EventPair, Handle, OnFDomainSignals, Socket, Vmo, ordinals};
8use fidl_fuchsia_fdomain as proto;
9use futures::future::Either;
10use futures::stream::Stream;
11use std::future::Future;
12use std::pin::Pin;
13use std::sync::{Arc, Weak};
14use std::task::{Context, Poll, ready};
15
16/// A channel in a remote FDomain.
17#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
18pub struct Channel(pub(crate) Handle);
19
20handle_type!(Channel CHANNEL peered);
21
22/// A message which has been read from a channel.
23#[derive(Debug)]
24pub struct MessageBuf {
25    pub bytes: Vec<u8>,
26    pub handles: Vec<HandleInfo>,
27}
28
29impl MessageBuf {
30    /// Create a new [`MessageBuf`]
31    pub fn new() -> Self {
32        MessageBuf { bytes: Vec::new(), handles: Vec::new() }
33    }
34
35    /// Get the components of this buffer separately.
36    pub fn split(self) -> (Vec<u8>, Vec<HandleInfo>) {
37        (self.bytes, self.handles)
38    }
39
40    /// Make sure this buffer has room for a certain number of bytes.
41    pub fn ensure_capacity_bytes(&mut self, bytes: usize) {
42        self.bytes.reserve(bytes);
43    }
44
45    /// Clear out the contents of this buffer.
46    pub fn clear(&mut self) {
47        self.bytes.clear();
48        self.handles.clear();
49    }
50
51    /// Get the byte content of this buffer.
52    pub fn bytes(&self) -> &[u8] {
53        self.bytes.as_slice()
54    }
55
56    /// Convert a proto ChannelMessage to a MessageBuf.
57    fn from_proto(client: &Arc<crate::Client>, message: proto::ChannelMessage) -> MessageBuf {
58        let proto::ChannelMessage { data, handles } = message;
59        MessageBuf {
60            bytes: data,
61            handles: handles
62                .into_iter()
63                .map(|info| {
64                    let handle = Handle { id: info.handle.id, client: Arc::downgrade(client) };
65                    HandleInfo {
66                        rights: info.rights,
67                        handle: AnyHandle::from_handle(handle, info.type_),
68                    }
69                })
70                .collect(),
71        }
72    }
73}
74
75/// A handle which has been read from a channel.
76#[derive(Debug)]
77pub struct HandleInfo {
78    pub handle: AnyHandle,
79    pub rights: fidl::Rights,
80}
81
82/// Sum type of all the handle types which can be read from a channel. Allows
83/// the user to learn the type of a handle after it has been read.
84#[derive(Debug, PartialEq)]
85pub enum AnyHandle {
86    Channel(Channel),
87    Socket(Socket),
88    Event(Event),
89    EventPair(EventPair),
90    Vmo(Vmo),
91    Unknown(Handle, fidl::ObjectType),
92}
93
94impl AnyHandle {
95    /// Construct an `AnyHandle` from a `Handle` and an object type.
96    pub fn from_handle(handle: Handle, ty: fidl::ObjectType) -> AnyHandle {
97        match ty {
98            fidl::ObjectType::CHANNEL => AnyHandle::Channel(Channel(handle)),
99            fidl::ObjectType::SOCKET => AnyHandle::Socket(Socket(handle)),
100            fidl::ObjectType::EVENT => AnyHandle::Event(Event(handle)),
101            fidl::ObjectType::EVENTPAIR => AnyHandle::EventPair(EventPair(handle)),
102            fidl::ObjectType::VMO => AnyHandle::Vmo(Vmo(handle)),
103            _ => AnyHandle::Unknown(handle, ty),
104        }
105    }
106
107    /// Get an `AnyHandle` wrapping an invalid handle.
108    pub fn invalid() -> AnyHandle {
109        AnyHandle::Unknown(Handle::invalid(), fidl::ObjectType::NONE)
110    }
111
112    /// Check whether this handle is valid.
113    pub fn is_invalid(&self) -> bool {
114        match self {
115            AnyHandle::Channel(h) => h.is_invalid(),
116            AnyHandle::Socket(h) => h.is_invalid(),
117            AnyHandle::Event(h) => h.is_invalid(),
118            AnyHandle::EventPair(h) => h.is_invalid(),
119            AnyHandle::Vmo(h) => h.is_invalid(),
120            AnyHandle::Unknown(h, _) => h.is_invalid(),
121        }
122    }
123
124    /// Get the object type for a handle.
125    pub fn object_type(&self) -> fidl::ObjectType {
126        match self {
127            AnyHandle::Channel(_) => fidl::ObjectType::CHANNEL,
128            AnyHandle::Socket(_) => fidl::ObjectType::SOCKET,
129            AnyHandle::Event(_) => fidl::ObjectType::EVENT,
130            AnyHandle::EventPair(_) => fidl::ObjectType::EVENTPAIR,
131            AnyHandle::Vmo(_) => fidl::ObjectType::VMO,
132            AnyHandle::Unknown(_, t) => *t,
133        }
134    }
135}
136
137impl From<AnyHandle> for Handle {
138    fn from(item: AnyHandle) -> Handle {
139        match item {
140            AnyHandle::Channel(h) => h.into(),
141            AnyHandle::Socket(h) => h.into(),
142            AnyHandle::Event(h) => h.into(),
143            AnyHandle::EventPair(h) => h.into(),
144            AnyHandle::Vmo(h) => h.into(),
145            AnyHandle::Unknown(h, _) => h,
146        }
147    }
148}
149
150/// Operation to perform on a handle when writing it to a channel.
151pub enum HandleOp<'h> {
152    Move(Handle, fidl::Rights),
153    Duplicate(&'h Handle, fidl::Rights),
154}
155
156impl Channel {
157    /// Reads a message from the channel.
158    pub fn recv_msg(&self) -> impl Future<Output = Result<MessageBuf, Error>> + use<> {
159        let client = Arc::downgrade(&self.0.client());
160        let handle = self.0.proto();
161
162        futures::future::poll_fn(move |ctx| {
163            let client = client.upgrade().unwrap_or_else(|| Arc::clone(&crate::DEAD_CLIENT));
164            client.poll_channel(handle, ctx, false).map(|x| {
165                x.expect("Got stream termination indication from non-streaming read!")
166                    .map(|x| MessageBuf::from_proto(&client, x))
167            })
168        })
169    }
170
171    /// Poll to try and read a channel message.
172    pub fn poll_read(&self, cx: &mut Context<'_>) -> Poll<Result<MessageBuf, Error>> {
173        let client = self.0.client();
174        let handle = self.0.proto();
175
176        client.poll_channel(handle, cx, false).map(|x| {
177            x.expect("Got stream termination indication from non-streaming read!")
178                .map(|x| MessageBuf::from_proto(&client, x))
179        })
180    }
181
182    /// Poll a channel for a message to read.
183    pub fn recv_from(&self, cx: &mut Context<'_>, buf: &mut MessageBuf) -> Poll<Result<(), Error>> {
184        let client = self.0.client();
185        match ready!(client.poll_channel(self.0.proto(), cx, false))
186            .expect("Got stream termination indication from non-streaming read!")
187        {
188            Ok(msg) => {
189                *buf = MessageBuf::from_proto(&client, msg);
190                Poll::Ready(Ok(()))
191            }
192            Err(e) => Poll::Ready(Err(e)),
193        }
194    }
195
196    /// Writes a message into the channel.
197    pub fn write(&self, bytes: &[u8], handles: Vec<Handle>) -> Result<(), Error> {
198        if bytes.len() > zx_types::ZX_CHANNEL_MAX_MSG_BYTES as usize
199            || handles.len() > zx_types::ZX_CHANNEL_MAX_MSG_HANDLES as usize
200        {
201            return Err(Error::FDomain(proto::Error::TargetError(
202                fidl::Status::OUT_OF_RANGE.into_raw(),
203            )));
204        }
205
206        let _ = self.write_inner(
207            bytes,
208            proto::Handles::Handles(handles.into_iter().map(|x| x.take_proto()).collect()),
209        );
210        Ok(())
211    }
212
213    /// Writes a message into the channel. Returns a future that will allow you
214    /// to wait for the write to move across the FDomain connection and return
215    /// with the result of the actual write call on the target.
216    pub fn fdomain_write(
217        &self,
218        bytes: &[u8],
219        handles: Vec<Handle>,
220    ) -> impl Future<Output = Result<(), Error>> + use<> {
221        if bytes.len() > zx_types::ZX_CHANNEL_MAX_MSG_BYTES as usize
222            || handles.len() > zx_types::ZX_CHANNEL_MAX_MSG_HANDLES as usize
223        {
224            Either::Left(async {
225                Err(Error::FDomain(proto::Error::TargetError(
226                    fidl::Status::OUT_OF_RANGE.into_raw(),
227                )))
228            })
229        } else {
230            Either::Right(self.write_inner(
231                bytes,
232                proto::Handles::Handles(handles.into_iter().map(|x| x.take_proto()).collect()),
233            ))
234        }
235    }
236
237    /// A future that returns when the channel is closed.
238    pub fn on_closed(&self) -> OnFDomainSignals {
239        OnFDomainSignals::new(&self.0, fidl::Signals::OBJECT_PEER_CLOSED)
240    }
241
242    /// Whether this handle is closed.
243    pub fn is_closed(&self) -> bool {
244        self.0.client.upgrade().is_none()
245    }
246
247    /// Writes a message into the channel. Optionally duplicates some of the
248    /// handles rather than consuming them, and can update the handle's rights
249    /// before sending.
250    pub fn fdomain_write_etc<'b>(
251        &self,
252        bytes: &[u8],
253        handles: Vec<HandleOp<'b>>,
254    ) -> impl Future<Output = Result<(), Error>> + use<'b> {
255        let handles = handles
256            .into_iter()
257            .map(|handle| match handle {
258                HandleOp::Move(x, rights) => {
259                    if Weak::ptr_eq(&x.client, &self.0.client) {
260                        Ok(proto::HandleDisposition {
261                            handle: proto::HandleOp::Move_(x.take_proto()),
262                            rights,
263                        })
264                    } else {
265                        Err(Error::ConnectionMismatch)
266                    }
267                }
268                HandleOp::Duplicate(x, rights) => {
269                    if Weak::ptr_eq(&x.client, &self.0.client) {
270                        Ok(proto::HandleDisposition {
271                            handle: proto::HandleOp::Duplicate(x.proto()),
272                            rights,
273                        })
274                    } else {
275                        Err(Error::ConnectionMismatch)
276                    }
277                }
278            })
279            .collect::<Result<Vec<_>, Error>>();
280
281        let handles = if handles
282            .as_ref()
283            .map(|x| x.len() > zx_types::ZX_CHANNEL_MAX_MSG_HANDLES as usize)
284            .unwrap_or(false)
285            || bytes.len() > zx_types::ZX_CHANNEL_MAX_MSG_BYTES as usize
286        {
287            Err(Error::FDomain(proto::Error::TargetError(fidl::Status::OUT_OF_RANGE.into_raw())))
288        } else {
289            handles
290        };
291
292        match handles {
293            Ok(handles) => {
294                Either::Left(self.write_inner(bytes, proto::Handles::Dispositions(handles)))
295            }
296            Err(e) => Either::Right(async move { Err(e) }),
297        }
298    }
299
300    /// Writes a message into the channel.
301    fn write_inner(
302        &self,
303        bytes: &[u8],
304        handles: proto::Handles,
305    ) -> impl Future<Output = Result<(), Error>> + use<> {
306        let data = bytes.to_vec();
307        let client = self.0.client();
308        let handle = self.0.proto();
309
310        client.clear_handles_for_transfer(&handles);
311        client.transaction(
312            ordinals::WRITE_CHANNEL,
313            proto::ChannelWriteChannelRequest { handle, data, handles },
314            move |x| Responder::WriteChannel(x),
315        )
316    }
317
318    /// Split this channel into a streaming reader and a writer. This is more
319    /// efficient on the read side if you intend to consume all of the messages
320    /// from the channel. However it will prevent you from transferring the
321    /// handle in the future. It also means messages will build up in the
322    /// buffer, so it may lead to memory issues if you don't intend to use the
323    /// messages from the channel as fast as they come.
324    pub fn stream(self) -> Result<(ChannelMessageStream, ChannelWriter), Error> {
325        let (a, b, err) = self.force_stream();
326        if let Some(err) = err { Err(err) } else { Ok((a, b)) }
327    }
328
329    /// Same as `stream` but will always try to generate the stream. If the
330    /// error is populated, the stream should also return the error at use time,
331    /// which is a more convenient way to handle the error in some cases.
332    pub(crate) fn force_stream(self) -> (ChannelMessageStream, ChannelWriter, Option<Error>) {
333        let err = self.0.client().start_channel_streaming(self.0.proto()).err();
334
335        let a = Arc::new(self);
336        let b = Arc::clone(&a);
337
338        (ChannelMessageStream(a), ChannelWriter(b), err)
339    }
340}
341
342/// A write-only handle to a socket.
343#[derive(Debug, Clone)]
344pub struct ChannelWriter(Arc<Channel>);
345
346impl ChannelWriter {
347    /// Writes a message into the channel.
348    pub fn write(&self, bytes: &[u8], handles: Vec<Handle>) -> Result<(), Error> {
349        self.0.write(bytes, handles)
350    }
351
352    /// Writes a message into the channel. Returns a future that will allow you
353    /// to wait for the write to move across the FDomain connection and return
354    /// with the result of the actual write call on the target.
355    pub fn fdomain_write(
356        &self,
357        bytes: &[u8],
358        handles: Vec<Handle>,
359    ) -> impl Future<Output = Result<(), Error>> {
360        self.0.fdomain_write(bytes, handles)
361    }
362
363    /// Writes a message into the channel.
364    pub fn fdomain_write_etc<'b>(
365        &self,
366        bytes: &[u8],
367        handles: Vec<HandleOp<'b>>,
368    ) -> impl Future<Output = Result<(), Error>> + 'b {
369        self.0.fdomain_write_etc(bytes, handles)
370    }
371
372    /// Get a reference to the inner channel.
373    pub fn as_channel(&self) -> &Channel {
374        &*self.0
375    }
376}
377
378/// A stream of data issuing from a socket.
379#[derive(Debug)]
380pub struct ChannelMessageStream(Arc<Channel>);
381
382impl ChannelMessageStream {
383    /// Turn a `ChannelMessageStream` and its accompanying `ChannelWriter` back
384    /// into a `Channel`.
385    ///
386    /// # Panics
387    /// If this stream and the writer passed didn't come from the same call to
388    /// `Channel::stream`, or if there is more than one writer.
389    pub fn rejoin(mut self, writer: ChannelWriter) -> Channel {
390        assert!(Arc::ptr_eq(&self.0, &writer.0), "Tried to join stream with wrong writer!");
391        if let Some(client) = self.0.0.client.upgrade() {
392            client.stop_channel_streaming(self.0.0.proto())
393        }
394        std::mem::drop(writer);
395        let channel = std::mem::replace(&mut self.0, Arc::new(Channel(Handle::invalid())));
396        Arc::try_unwrap(channel).expect("Stream pointer no longer unique!")
397    }
398
399    /// Whether this stream is closed.
400    pub fn is_closed(&self) -> bool {
401        let client = self.0.0.client();
402
403        !client.channel_is_streaming(self.0.0.proto())
404    }
405
406    /// Get a reference to the inner channel.
407    pub fn as_channel(&self) -> &Channel {
408        &*self.0
409    }
410}
411
412impl Stream for ChannelMessageStream {
413    type Item = Result<MessageBuf, Error>;
414    fn poll_next(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
415        let client = self.0.0.client();
416        client
417            .poll_channel(self.0.0.proto(), ctx, true)
418            .map(|x| x.map(|x| x.map(|x| MessageBuf::from_proto(&client, x))))
419    }
420}
421
422impl Drop for ChannelMessageStream {
423    fn drop(&mut self) {
424        if let Some(client) = self.0.0.client.upgrade() {
425            client.stop_channel_streaming(self.0.0.proto());
426        }
427    }
428}
fdomain_client/channel.rs

fdomain_client/
channel.rs
