Add support for FD passing

async-byte-channel/Cargo.toml

@@ -7,7 +7,12 @@ description = "Helper library for writing async tests"
 repository = "https://github.com/dwrensha/capnproto-rust"
 edition = "2021"
 
-[dependencies.futures]
+[dependencies.capnp-futures]
+version = "0.26.0"
+path = "../capnp-futures"
+default-features = false
+
+[dev-dependencies.futures]
 version = "0.3.0"
 default-features = false
 features = ["std", "executor"]

async-byte-channel/src/lib.rs

@@ -1,10 +1,10 @@
 // Simple in-memory byte stream.
 
-use std::pin::Pin;
+use std::io;
 use std::sync::{Arc, Mutex};
 
-use futures::{AsyncRead, AsyncWrite};
-use std::task::{Poll, Waker};
+use capnp_futures::io::{AsyncFdRead, AsyncFdWrite, Count, FdReadBuf, FdWriteBuf};
+use std::task::{Context, Poll, Waker};
 
 #[derive(Debug)]
 struct Inner {
@@ -68,16 +68,17 @@ pub fn channel() -> (Sender, Receiver) {
     (sender, receiver)
 }
 
-impl AsyncRead for Receiver {
-    fn poll_read(
-        self: Pin<&mut Self>,
-        cx: &mut futures::task::Context,
+impl AsyncFdRead for Receiver {
+    fn poll_read_with_fds(
+        &mut self,
+        cx: &mut Context<'_>,
         buf: &mut [u8],
-    ) -> futures::task::Poll<Result<usize, futures::io::Error>> {
+        _fd_buf: &mut FdReadBuf,
+    ) -> Poll<io::Result<Count>> {
         let mut inner = self.inner.lock().unwrap();
         if inner.read_cursor == inner.write_cursor {
             if inner.write_end_closed {
-                Poll::Ready(Ok(0))
+                Poll::Ready(Ok(Count { bytes: 0, fds: 0 }))
             } else {
                 inner.read_waker = Some(cx.waker().clone());
                 Poll::Pending
@@ -91,17 +92,21 @@ impl AsyncRead for Receiver {
             if let Some(write_waker) = inner.write_waker.take() {
                 write_waker.wake();
             }
-            Poll::Ready(Ok(copy_len))
+            Poll::Ready(Ok(Count {
+                bytes: copy_len,
+                fds: 0,
+            }))
         }
     }
 }
 
-impl AsyncWrite for Sender {
-    fn poll_write(
-        self: Pin<&mut Self>,
-        cx: &mut futures::task::Context,
+impl AsyncFdWrite for Sender {
+    fn poll_write_with_fds(
+        &mut self,
+        cx: &mut Context<'_>,
         buf: &[u8],
-    ) -> futures::task::Poll<Result<usize, futures::io::Error>> {
+        _fd_buf: &FdWriteBuf<'_>,
+    ) -> Poll<io::Result<Count>> {
         let mut inner = self.inner.lock().unwrap();
         if inner.read_end_closed {
             return Poll::Ready(Err(std::io::Error::new(
@@ -128,33 +133,23 @@ impl AsyncWrite for Sender {
         if let Some(read_waker) = inner.read_waker.take() {
             read_waker.wake();
         }
-        Poll::Ready(Ok(copy_len))
+        Poll::Ready(Ok(Count {
+            bytes: copy_len,
+            fds: 0,
+        }))
     }
 
-    fn poll_flush(
-        self: Pin<&mut Self>,
-        _cx: &mut futures::task::Context,
-    ) -> Poll<Result<(), futures::io::Error>> {
-        Poll::Ready(Ok(()))
-    }
-
-    fn poll_close(
-        self: Pin<&mut Self>,
-        _cx: &mut futures::task::Context,
-    ) -> Poll<Result<(), futures::io::Error>> {
-        let mut inner = self.inner.lock().unwrap();
-        inner.write_end_closed = true;
-        if let Some(read_waker) = inner.read_waker.take() {
-            read_waker.wake();
-        }
+    fn poll_flush(&mut self, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         Poll::Ready(Ok(()))
     }
 }
 
 #[cfg(test)]
 pub mod test {
+    use std::io;
+
+    use capnp_futures::io::{AsyncFdReadExt, AsyncFdWriteExt as _};
     use futures::task::LocalSpawnExt;
-    use futures::{AsyncReadExt, AsyncWriteExt};
 
     #[test]
     fn basic() {
@@ -173,10 +168,12 @@ pub mod test {
             })
             .unwrap();
 
-        let mut buf3 = vec![];
-        pool.run_until(receiver.read_to_end(&mut buf3)).unwrap();
-
-        assert_eq!(buf.len(), buf3.len());
+        pool.run_until(async {
+            receiver.read_exact(&mut vec![0; buf.len()]).await?;
+            assert_eq!(receiver.read(&mut [0]).await?, 0);
+            io::Result::Ok(())
+        })
+        .unwrap();
     }
 
     #[test]

capnp-futures/Cargo.toml

@@ -20,10 +20,28 @@ version = "0.3.0"
 default-features = false
 features = ["std"]
 
-[dependencies.futures-util]
+[dependencies.futures-core]
 version = "0.3.0"
 default-features = false
-features = ["io", "std"]
+features = ["std"]
+
+[dependencies.futures-io]
+version = "0.3.0"
+default-features = false
+features = ["std"]
+optional = true
+
+[dependencies.rustix]
+version = "1.1.4"
+default-features = false
+features = ["std", "net"]
+optional = true
+
+[dependencies.tokio]
+version = "1.52.3"
+default-features = false
+features = ["net"]
+optional = true
 
 [dev-dependencies.futures]
 version = "0.3.0"
@@ -34,5 +52,11 @@ features = ["executor"]
 capnp = { version = "0.26.0", path = "../capnp", features = ["quickcheck"] }
 quickcheck = "1"
 
+[features]
+default = ["futures-io"]
+futures-io = ["dep:futures-io"]
+tokio = ["dep:tokio"]
+tokio-unix-fd-stream = ["tokio", "dep:rustix"]
+
 [lints]
 workspace = true

capnp-futures/src/io/futures_io/mod.rs

@@ -0,0 +1,101 @@
+// Copyright (c) 2026 Sandstorm Development Group, Inc. and contributors
+// Licensed under the MIT License:
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+use std::{
+    io,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use futures_io::{AsyncRead, AsyncWrite};
+
+pub use read_stream::ReadStream;
+pub use write_queue::{write_queue, Sender};
+
+use crate::io::{AsyncFdRead, AsyncFdWrite, Count, FdReadBuf, FdWriteBuf};
+
+mod read_stream;
+pub mod serialize;
+pub mod serialize_packed;
+mod write_queue;
+
+#[derive(Debug)]
+pub struct Compat<T> {
+    inner: T,
+}
+
+impl<T> Compat<T> {
+    pub fn new(inner: T) -> Self {
+        Self { inner }
+    }
+
+    pub fn into_inner(self) -> T {
+        self.inner
+    }
+}
+
+impl<T> From<T> for Compat<T> {
+    fn from(value: T) -> Self {
+        Self::new(value)
+    }
+}
+
+impl<T> AsRef<T> for Compat<T> {
+    fn as_ref(&self) -> &T {
+        &self.inner
+    }
+}
+
+impl<T> AsMut<T> for Compat<T> {
+    fn as_mut(&mut self) -> &mut T {
+        &mut self.inner
+    }
+}
+
+impl<R: AsyncRead + Unpin> AsyncFdRead for Compat<R> {
+    fn poll_read_with_fds(
+        &mut self,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+        _fd_buf: &mut FdReadBuf,
+    ) -> Poll<io::Result<Count>> {
+        Pin::new(&mut self.inner)
+            .poll_read(cx, buf)
+            .map_ok(|bytes| Count { bytes, fds: 0 })
+    }
+}
+
+impl<W: AsyncWrite + Unpin> AsyncFdWrite for Compat<W> {
+    fn poll_write_with_fds(
+        &mut self,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+        _fd_buf: &FdWriteBuf<'_>,
+    ) -> Poll<io::Result<Count>> {
+        Pin::new(&mut self.inner)
+            .poll_write(cx, buf)
+            .map_ok(|bytes| Count { bytes, fds: 0 })
+    }
+
+    fn poll_flush(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+}

capnp-futures/src/io/futures_io/read_stream.rs

@@ -0,0 +1,61 @@
+// Copyright (c) 2016 Sandstorm Development Group, Inc. and contributors
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use capnp::{message, Error};
+use futures_core::stream::Stream;
+use futures_io::AsyncRead;
+
+use crate::io::futures_io::Compat;
+
+/// An incoming sequence of messages.
+#[must_use = "streams do nothing unless polled"]
+pub struct ReadStream<'a, R>
+where
+    R: AsyncRead + Unpin,
+{
+    inner: crate::io::read_stream::ReadStream<'a, Compat<R>>,
+}
+
+impl<R> Unpin for ReadStream<'_, R> where R: AsyncRead + Unpin {}
+
+impl<'a, R> ReadStream<'a, R>
+where
+    R: AsyncRead + Unpin + 'a,
+{
+    pub fn new(reader: R, options: message::ReaderOptions) -> Self {
+        ReadStream {
+            inner: crate::io::read_stream::ReadStream::new(Compat::new(reader), options),
+        }
+    }
+}
+
+impl<'a, R> Stream for ReadStream<'a, R>
+where
+    R: AsyncRead + Unpin + 'a,
+{
+    type Item = Result<message::Reader<capnp::serialize::OwnedSegments>, Error>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        Pin::new(&mut self.inner).poll_next(cx)
+    }
+}