Merge pull request #5 from n0-computer/Frando/ergonomics

feat: add methods on `Client` to issue requests on either local or remote APIs

Author: rklaehn

examples/derive.rs

@@ -4,12 +4,12 @@ use std::{
     sync::Arc,
 };
 
-use anyhow::bail;
+use anyhow::{Context, Result};
 use irpc::{
     channel::{oneshot, spsc},
-    rpc::{listen, Handler},
+    rpc::Handler,
     util::{make_client_endpoint, make_server_endpoint},
-    Client, LocalSender, Request, Service, WithChannels,
+    Client, Error, LocalSender, Service, WithChannels,
 };
 // Import the macro
 use irpc_derive::rpc_requests;
@@ -37,6 +37,15 @@ struct Set {
     value: String,
 }
 
+impl From<(String, String)> for Set {
+    fn from((key, value): (String, String)) -> Self {
+        Self { key, value }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+struct SetMany;
+
 // Use the macro to generate both the StorageProtocol and StorageMessage enums
 // plus implement Channels for each type
 #[rpc_requests(StorageService, message = StorageMessage)]
@@ -46,6 +55,8 @@ enum StorageProtocol {
     Get(Get),
     #[rpc(tx=oneshot::Sender<()>)]
     Set(Set),
+    #[rpc(tx=oneshot::Sender<u64>, rx=spsc::Receiver<(String, String)>)]
+    SetMany(SetMany),
     #[rpc(tx=spsc::Sender<String>)]
     List(List),
 }
@@ -56,7 +67,7 @@ struct StorageActor {
 }
 
 impl StorageActor {
-    pub fn local() -> StorageApi {
+    pub fn spawn() -> StorageApi {
         let (tx, rx) = tokio::sync::mpsc::channel(1);
         let actor = Self {
             recv: rx,
@@ -88,6 +99,16 @@ impl StorageActor {
                 self.state.insert(inner.key, inner.value);
                 tx.send(()).await.ok();
             }
+            StorageMessage::SetMany(set) => {
+                info!("set-many {:?}", set);
+                let WithChannels { mut rx, tx, .. } = set;
+                let mut count = 0;
+                while let Ok(Some((key, value))) = rx.recv().await {
+                    self.state.insert(key, value);
+                    count += 1;
+                }
+                tx.send(count).await.ok();
+            }
             StorageMessage::List(list) => {
                 info!("list {:?}", list);
                 let WithChannels { mut tx, .. } = list;
@@ -106,120 +127,98 @@ struct StorageApi {
 }
 
 impl StorageApi {
-    pub fn connect(endpoint: quinn::Endpoint, addr: SocketAddr) -> anyhow::Result<StorageApi> {
+    pub fn connect(endpoint: quinn::Endpoint, addr: SocketAddr) -> Result<StorageApi> {
         Ok(StorageApi {
             inner: Client::quinn(endpoint, addr),
         })
     }
 
-    pub fn listen(&self, endpoint: quinn::Endpoint) -> anyhow::Result<AbortOnDropHandle<()>> {
-        let Some(local) = self.inner.local() else {
-            bail!("cannot listen on a remote service");
-        };
-        let handler: Handler<StorageProtocol> = Arc::new(move |msg, _, tx| {
+    pub fn listen(&self, endpoint: quinn::Endpoint) -> Result<AbortOnDropHandle<()>> {
+        let local = self.inner.local().context("cannot listen on remote API")?;
+        let handler: Handler<StorageProtocol> = Arc::new(move |msg, rx, tx| {
             let local = local.clone();
             Box::pin(match msg {
                 StorageProtocol::Get(msg) => local.send((msg, tx)),
                 StorageProtocol::Set(msg) => local.send((msg, tx)),
+                StorageProtocol::SetMany(msg) => local.send((msg, tx, rx)),
                 StorageProtocol::List(msg) => local.send((msg, tx)),
             })
         });
-        Ok(AbortOnDropHandle::new(task::spawn(listen(
-            endpoint, handler,
-        ))))
+        let join_handle = task::spawn(irpc::rpc::listen(endpoint, handler));
+        Ok(AbortOnDropHandle::new(join_handle))
     }
 
-    pub async fn get(&self, key: String) -> anyhow::Result<oneshot::Receiver<Option<String>>> {
-        let msg = Get { key };
-        match self.inner.request().await? {
-            Request::Local(request) => {
-                let (tx, rx) = oneshot::channel();
-                request.send((msg, tx)).await?;
-                Ok(rx)
-            }
-            Request::Remote(request) => {
-                let (_tx, rx) = request.write(msg).await?;
-                Ok(rx.into())
-            }
-        }
+    pub async fn get(&self, key: String) -> Result<Option<String>, Error> {
+        self.inner.rpc(Get { key }).await
     }
 
-    pub async fn list(&self) -> anyhow::Result<spsc::Receiver<String>> {
-        let msg = List;
-        match self.inner.request().await? {
-            Request::Local(request) => {
-                let (tx, rx) = spsc::channel(10);
-                request.send((msg, tx)).await?;
-                Ok(rx)
-            }
-            Request::Remote(request) => {
-                let (_tx, rx) = request.write(msg).await?;
-                Ok(rx.into())
-            }
-        }
+    pub async fn list(&self) -> Result<spsc::Receiver<String>, Error> {
+        self.inner.server_streaming(List, 16).await
     }
 
-    pub async fn set(&self, key: String, value: String) -> anyhow::Result<oneshot::Receiver<()>> {
-        let msg = Set { key, value };
-        match self.inner.request().await? {
-            Request::Local(request) => {
-                let (tx, rx) = oneshot::channel();
-                request.send((msg, tx)).await?;
-                Ok(rx)
-            }
-            Request::Remote(request) => {
-                let (_tx, rx) = request.write(msg).await?;
-                Ok(rx.into())
-            }
-        }
+    pub async fn set(&self, key: String, value: String) -> Result<(), Error> {
+        self.inner.rpc(Set { key, value }).await
+    }
+
+    pub async fn set_many(
+        &self,
+    ) -> Result<(spsc::Sender<(String, String)>, oneshot::Receiver<u64>), Error> {
+        self.inner.client_streaming(SetMany, 4).await
     }
 }
 
-async fn local() -> anyhow::Result<()> {
-    let api = StorageActor::local();
-    api.set("hello".to_string(), "world".to_string())
-        .await?
-        .await?;
-    let value = api.get("hello".to_string()).await?.await?;
+async fn client_demo(api: StorageApi) -> Result<()> {
+    api.set("hello".to_string(), "world".to_string()).await?;
+    let value = api.get("hello".to_string()).await?;
+    println!("get: hello = {:?}", value);
+
+    let (mut tx, rx) = api.set_many().await?;
+    for i in 0..3 {
+        tx.send((format!("key{i}"), format!("value{i}"))).await?;
+    }
+    drop(tx);
+    let count = rx.await?;
+    println!("set-many: {count} values set");
+
     let mut list = api.list().await?;
     while let Some(value) = list.recv().await? {
         println!("list value = {:?}", value);
     }
-    println!("value = {:?}", value);
     Ok(())
 }
 
-async fn remote() -> anyhow::Result<()> {
+async fn local() -> Result<()> {
+    let api = StorageActor::spawn();
+    client_demo(api).await?;
+    Ok(())
+}
+
+async fn remote() -> Result<()> {
     let port = 10113;
-    let (server, cert) =
-        make_server_endpoint(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, port).into())?;
-    let client =
+    let addr: SocketAddr = SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into();
+
+    let (server_handle, cert) = {
+        let (endpoint, cert) = make_server_endpoint(addr)?;
+        let api = StorageActor::spawn();
+        let handle = api.listen(endpoint)?;
+        (handle, cert)
+    };
+
+    let endpoint =
         make_client_endpoint(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 0).into(), &[&cert])?;
-    let store = StorageActor::local();
-    let handle = store.listen(server)?;
-    let api = StorageApi::connect(client, SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into())?;
-    api.set("hello".to_string(), "world".to_string())
-        .await?
-        .await?;
-    api.set("goodbye".to_string(), "world".to_string())
-        .await?
-        .await?;
-    let value = api.get("hello".to_string()).await?.await?;
-    println!("value = {:?}", value);
-    let mut list = api.list().await?;
-    while let Some(value) = list.recv().await? {
-        println!("list value = {:?}", value);
-    }
-    drop(handle);
+    let api = StorageApi::connect(endpoint, addr)?;
+    client_demo(api).await?;
+
+    drop(server_handle);
     Ok(())
 }
 
 #[tokio::main]
-async fn main() -> anyhow::Result<()> {
-    tracing_subscriber::fmt().init();
+async fn main() -> Result<()> {
+    tracing_subscriber::fmt::init();
     println!("Local use");
     local().await?;
     println!("Remote use");
-    remote().await?;
+    remote().await.unwrap();
     Ok(())
 }