Add support for FD passing to capnp-rpc

Closes: #625

Author: emilazy

capnp-futures/src/io/write_queue.rs

@@ -37,6 +37,7 @@ where
     M: AsOutputSegments,
 {
     Message(M, oneshot::Sender<M>),
+    MessageMove(M, oneshot::Sender<()>),
     Done(Result<(), Error>, oneshot::Sender<()>),
 }
 
@@ -91,6 +92,13 @@ where
                     writer.flush().await?;
                     let _ = returner.send(m);
                 }
+                Item::MessageMove(m, finisher) => {
+                    let result = crate::io::serialize::write_message(&mut writer, m).await;
+                    in_flight.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
+                    result?;
+                    writer.flush().await?;
+                    let _ = finisher.send(());
+                }
                 Item::Done(r, finisher) => {
                     let _ = finisher.send(());
                     return r;
@@ -134,6 +142,21 @@ where
         MapErr(oneshot)
     }
 
+    pub fn send_move(
+        &mut self,
+        message: M,
+    ) -> impl Future<Output = Result<(), Error>> + Unpin + 'static {
+        self.in_flight
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        let (complete, oneshot) = oneshot::channel();
+
+        let _ = self
+            .sender
+            .unbounded_send(Item::MessageMove(message, complete));
+
+        MapErr(oneshot)
+    }
+
     /// Returns the number of messages queued to be written.
     pub fn len(&self) -> usize {
         let result = self.in_flight.load(std::sync::atomic::Ordering::SeqCst);

capnp-rpc/src/broken.rs

@@ -20,6 +20,7 @@
 // THE SOFTWARE.
 
 use capnp::any_pointer;
+use capnp::fd::BorrowedFd;
 use capnp::private::capability::{
     ClientHook, ParamsHook, PipelineHook, PipelineOp, RequestHook, ResultsHook,
 };
@@ -158,6 +159,10 @@ impl ClientHook for Client {
     fn when_resolved(&self) -> Promise<(), Error> {
         crate::rpc::default_when_resolved_impl(self)
     }
+
+    fn get_fd(&self) -> Option<BorrowedFd<'_>> {
+        None
+    }
 }
 
 pub(crate) fn new_cap(exception: Error) -> Box<dyn ClientHook> {

capnp-rpc/src/lib.rs

@@ -59,8 +59,10 @@
 //! For a more complete example, see <https://github.com/capnproto/capnproto-rust/tree/master/capnp-rpc/examples/calculator>
 
 use capnp::capability::Promise;
+use capnp::fd::FdHooks;
 use capnp::private::capability::ClientHook;
 use capnp::Error;
+use capnp_futures::io::FdReadBuf;
 use futures_channel::oneshot;
 use futures_util::{FutureExt as _, TryFutureExt as _};
 use std::cell::RefCell;
@@ -123,6 +125,10 @@ pub trait OutgoingMessage {
     /// Same as `get_body()`, but returns the corresponding reader type.
     fn get_body_as_reader(&self) -> ::capnp::Result<::capnp::any_pointer::Reader<'_>>;
 
+    fn set_fds(&mut self, fds: FdHooks) {
+        let _ = fds;
+    }
+
     /// Sends the message. Returns a promise that resolves once the send has completed.
     /// Dropping the returned promise does *not* cancel the send.
     fn send(
@@ -149,6 +155,15 @@ pub trait IncomingMessage {
     /// The standard RPC implementation interprets it as a Message as defined
     /// in `schema/rpc.capnp`.
     fn get_body(&self) -> ::capnp::Result<::capnp::any_pointer::Reader<'_>>;
+
+    fn get_body_and_attached_fds(
+        &mut self,
+    ) -> (
+        ::capnp::Result<::capnp::any_pointer::Reader<'_>>,
+        &mut FdReadBuf,
+    ) {
+        (self.get_body(), &mut [])
+    }
 }
 
 /// A two-way RPC connection.

capnp-rpc/src/local.rs

@@ -19,6 +19,7 @@
 // THE SOFTWARE.
 
 use capnp::capability::{self, Promise};
+use capnp::fd::BorrowedFd;
 use capnp::private::capability::{
     ClientHook, ParamsHook, PipelineHook, PipelineOp, RequestHook, ResponseHook, ResultsHook,
 };
@@ -612,4 +613,8 @@ where
     fn when_resolved(&self) -> Promise<(), Error> {
         crate::rpc::default_when_resolved_impl(self)
     }
+
+    fn get_fd(&self) -> Option<BorrowedFd<'_>> {
+        self.inner.server.get_fd()
+    }
 }

capnp-rpc/src/queued.rs

@@ -21,6 +21,7 @@
 
 use capnp::any_pointer;
 use capnp::capability::Promise;
+use capnp::fd::BorrowedFd;
 use capnp::private::capability::{ClientHook, ParamsHook, PipelineHook, PipelineOp, ResultsHook};
 use capnp::Error;
 use futures_util::{FutureExt as _, TryFutureExt as _};
@@ -352,4 +353,8 @@ impl ClientHook for Client {
     fn when_resolved(&self) -> Promise<(), Error> {
         crate::rpc::default_when_resolved_impl(self)
     }
+
+    fn get_fd(&self) -> Option<BorrowedFd<'_>> {
+        self.inner.redirect.get().and_then(|p| p.get_fd())
+    }
 }

capnp-rpc/src/reconnect.rs

@@ -3,6 +3,7 @@ use std::marker::PhantomData;
 use std::rc::Rc;
 
 use capnp::capability::{FromClientHook, Promise};
+use capnp::fd::BorrowedFd;
 use capnp::private::capability::{ClientHook, RequestHook};
 use futures_util::TryFutureExt as _;
 
@@ -206,6 +207,13 @@ where
     fn when_resolved(&self) -> Promise<(), capnp::Error> {
         Promise::ok(())
     }
+
+    fn get_fd(&self) -> Option<BorrowedFd<'_>> {
+        // We can’t return `self.get_current().get_fd()` because it might not
+        // live long enough. This matches the C++ behaviour; see
+        // <https://github.com/capnproto/capnproto/blob/291d8ee870ab394d32fbdd5d2160add84b65e229/c%2B%2B/src/capnp/reconnect.c%2B%2B#L72-L77>.
+        None
+    }
 }
 
 struct Request<F, C> {