rpcbase: Use with_fn.

Cargo.lock

@@ -4437,6 +4437,7 @@ dependencies = [
  "serde_json",
  "serde_repr",
  "thiserror",
+ "tor-async-utils",
  "tor-error",
  "typetag",
 ]

crates/tor-rpcbase/Cargo.toml

@@ -22,6 +22,7 @@ paste = "1"
 serde = { version = "1.0.103", features = ["derive"] }
 serde_repr = "0.1.12"
 thiserror = "1"
+tor-async-utils = { path = "../tor-async-utils", version = "0.0.1" }
 tor-error = { path = "../tor-error/", version = "0.4.1", features = ["rpc"] }
 typetag = "0.2.7"
 

crates/tor-rpcbase/src/dispatch.rs

@@ -23,6 +23,10 @@ pub type RpcValue = Box<dyn erased_serde::Serialize + Send + 'static>;
 #[doc(hidden)]
 pub type RpcResult = Result<RpcValue, RpcError>;
 
+/// The return type from sending an update.
+#[doc(hidden)]
+pub type RpcSendResult = Result<RpcValue, SendUpdateError>;
+
 /// A boxed future holding the result of an RPC method.
 type RpcResultFuture = BoxFuture<'static, RpcResult>;
 
@@ -127,7 +131,9 @@ macro_rules! rpc_invoke_fn {
             type Output = <$methodtype as $crate::Method>::Output;
             use $crate::futures::FutureExt;
             #[allow(unused)]
-            use $crate::futures::sink::{self, SinkExt};
+            use $crate::{
+                tor_async_utils::SinkExt as _
+            };
             let obj = obj
                 .downcast_arc::<$objtype>()
                 .unwrap_or_else(|_| panic!());
@@ -135,12 +141,9 @@ macro_rules! rpc_invoke_fn {
                 .downcast::<$methodtype>()
                 .unwrap_or_else(|_| panic!());
             $(
-                // TODO: I would prefer to have a `with` alternative that
+                let $sink = sink.with_fn(|update|
-                // applied a simple (non async) function to a Sink.
+                    $crate::dispatch::RpcSendResult::Ok(Box::new(update))
-                let $sink = Box::pin(sink.with(|update: <$methodtype as $crate::Method>::Update| async {
+                );
-                    let boxed: $crate::dispatch::RpcValue = Box::new(update);
-                    Ok::<_,$crate::SendUpdateError>(boxed)
-                }));
             )?
             $funcname(obj, method, ctx $(, $sink)?).map(|r| {
                 let r: $crate::RpcResult = match r {

crates/tor-rpcbase/src/lib.rs

@@ -52,7 +52,9 @@ pub use method::{DynMethod, Method, NoUpdates};
 pub use obj::{Object, ObjectId};
 
 #[doc(hidden)]
-pub use {dispatch::RpcResult, downcast_rs, erased_serde, futures, inventory, paste};
+pub use {
+    dispatch::RpcResult, downcast_rs, erased_serde, futures, inventory, paste, tor_async_utils,
+};
 
 /// An error returned from [`ContextExt::lookup`].
 ///