misc

gdriver-client/src/main.rs

@@ -1,4 +1,4 @@
-use fuser::{mount2, MountOption};
+use fuser::{mount2, spawn_mount2, MountOption};
 use std::path::Path;
 use std::{error::Error, net::IpAddr, result::Result as StdResult};
 
@@ -18,13 +18,15 @@ async fn main() -> Result<()> {
     println!("Hello, world!");
     let config = &CONFIGURATION;
     println!("Config: {:?}", **config);
-    let client: GDriverServiceClient = create_client(config.ip, config.port).await?;
-    let fs = DriveFilesystem::new(client);
-    let mountpoint = Path::new("/tmp/gdriver");
-    std::fs::create_dir_all(mountpoint)?;
-    mount2(fs, mountpoint, &[MountOption::RW, MountOption::AutoUnmount])?;
+    let (client, _handler) = create_client(&config.socket_path).await?;
 
-    // service::start().await?;
+    // let fs = DriveFilesystem::new(client);
+    // let mountpoint = Path::new("/tmp/gdriver");
+    // std::fs::create_dir_all(mountpoint)?;
+    // // let mount_handle =
+    // mount2(fs, mountpoint, &[MountOption::RW, MountOption::AutoUnmount])?;
+    // mount_handle.guard.join().unwrap()?;
+    service::start_with_client(client, _handler).await?;
     Ok(())
 }
 pub mod prelude;

gdriver-client/src/sample.rs

@@ -6,7 +6,7 @@ pub async fn start() -> Result<()> {
 
     let name = "test1".to_string();
     let config = &CONFIGURATION;
-    let client: WorldClient = create_client(config.ip, config.port).await?;
+    let client: WorldClient = create_client(&config.socket_path).await?;
 
     let hello = client
         .hello(tarpc::context::current(), name.to_string())
@@ -18,14 +18,10 @@ pub async fn start() -> Result<()> {
     }
     Ok(())
 }
-pub async fn create_client(ip: IpAddr, port: u16) -> Result<WorldClient> {
-    let server_addr = (ip, port);
-    let transport = tarpc::serde_transport::tcp::connect(&server_addr, Json::default)
+pub async fn create_client(socket_path: impl AsRef<Path>) -> Result<WorldClient> {
+    let transport = tarpc::serde_transport::unix::connect(socket_path, Json::default)
         .await
-        .map_err(|e| {
-            println!("Could not connect");
-            e
-        })?;
+        .inspect_err(|_| println!("Could not connect"))?;
     let var_name = WorldClient::new(client::Config::default(), transport);
     let client = var_name.spawn();
     Ok(client)

gdriver-client/src/service.rs

@@ -1,15 +1,46 @@
 use std::time;
 
-use gdriver_common::ipc::gdriver_service::{BackendActionRequest, GDriverServiceClient};
+use futures::prelude::*;
+use gdriver_common::ipc::gdriver_service::{
+    AsyncResponse, BackendActionRequest, GDriverClient, GDriverServiceClient, TaskId, TaskResult,
+};
+use sample_sym::spawn_twoway;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use tarpc::context::Context;
+use tarpc::server::{BaseChannel, Channel};
+use tokio::sync::mpsc;
 
 use super::*;
 
+#[derive(Clone)]
+pub(crate) struct ClientHandler {
+    pending_tasks: Arc<Mutex<HashMap<String, mpsc::Sender<TaskResult>>>>,
+}
+
+impl GDriverClient for ClientHandler {
+    async fn report_task_result(self, _: Context, id: TaskId, result: TaskResult) {
+        println!("Received task result for task {}: {:?}", id.0, result);
+        let mut tasks = self.pending_tasks.lock().unwrap();
+        if let Some(sender) = tasks.remove(&id.0) {
+            let _ = sender.send(result).await;
+        }
+    }
+}
+
 pub async fn start() -> Result<()> {
     println!("Hello, world!");
     let config = &CONFIGURATION;
     println!("Config: {:?}", **config);
-    let client: GDriverServiceClient = create_client(config.ip, config.port).await?;
+    let (client, handler) = create_client(&config.socket_path).await?;
 
+    start_with_client(client, handler).await
+}
+
+pub(crate) async fn start_with_client(
+    client: GDriverServiceClient,
+    handler: ClientHandler,
+) -> Result<()> {
     let hello = client
         .do_something2(tarpc::context::current(), BackendActionRequest::Ping)
         .await;
@@ -42,20 +73,150 @@ pub async fn start() -> Result<()> {
         .as_secs();
 
     match hello {
-        Ok(hello) => println!("Start Long returned after {} seconds: {:?}", seconds, hello),
-        Err(e) => println!(":( {:?}", (e)),
+        Ok(Ok(AsyncResponse::Pending(task_id))) => {
+            println!("Start Long returned pending task: {:?}", task_id);
+            let (tx, mut rx) = mpsc::channel(1);
+            handler
+                .pending_tasks
+                .lock()
+                .unwrap()
+                .insert(task_id.0.clone(), tx);
+
+            if let Some(result) = rx.recv().await {
+                println!(
+                    "Received async result after {} seconds: {:?}",
+                    (time::SystemTime::now().duration_since(start))
+                        .unwrap()
+                        .as_secs(),
+                    result
+                );
+            }
+        }
+        Ok(Ok(AsyncResponse::Immediate(res))) => {
+            println!("Start Long returned immediate: {:?}", res)
+        }
+        Ok(Err(e)) => println!("Backend Error: {:?}", e),
+        Err(e) => println!("RPC Error: {:?}", e),
     }
     Ok(())
 }
-pub async fn create_client(ip: IpAddr, port: u16) -> Result<GDriverServiceClient> {
-    let server_addr = (ip, port);
-    let transport = tarpc::serde_transport::tcp::connect(&server_addr, Json::default)
+
+pub async fn create_client(
+    socket_path: impl AsRef<Path>,
+) -> Result<(GDriverServiceClient, ClientHandler)> {
+    let transport = tarpc::serde_transport::unix::connect(socket_path, Json::default)
         .await
         .map_err(|e| {
             println!("Could not connect");
             e
         })?;
-    let service = GDriverServiceClient::new(client::Config::default(), transport);
-    let client = service.spawn();
-    Ok(client)
+
+    let (client_channel, server_channel) = spawn_twoway(transport);
+
+    let client = GDriverServiceClient::new(client::Config::default(), server_channel).spawn();
+
+    let handler = ClientHandler {
+        pending_tasks: Arc::new(Mutex::new(HashMap::new())),
+    };
+
+    let handler_clone = handler.clone();
+    tokio::spawn(async move {
+        BaseChannel::with_defaults(client_channel)
+            .execute(handler_clone.serve())
+            .for_each(|_| async {})
+            .await;
+    });
+
+    Ok((client, handler))
+}
+
+mod sample_sym {
+    use futures::future::{AbortHandle, Abortable};
+    use futures::{Sink, TryFutureExt};
+    use futures::{SinkExt, Stream, StreamExt, TryStreamExt};
+    use serde::{Deserialize, Serialize};
+    use std::io;
+    use tarpc::transport::channel::{ChannelError, UnboundedChannel};
+    use tracing::{debug, warn};
+
+    #[derive(Debug, Serialize, Deserialize)]
+    pub enum TwoWayMessage<Req, Resp> {
+        Request(tarpc::ClientMessage<Req>),
+        Response(tarpc::Response<Resp>),
+    }
+
+    #[derive(thiserror::Error, Debug)]
+    pub enum ChannelOrIoError {
+        #[error("{0}")]
+        ChannelError(#[from] ChannelError),
+        #[error("{0}")]
+        IoError(#[from] io::Error),
+    }
+
+    /// Returns two transports that multiplex over the given transport.
+    /// The first transport can be used by a server: it receives requests and sends back responses.
+    /// The second transport can be used by a client: it sends requests and receives back responses.
+    pub fn spawn_twoway<Req1, Resp1, Req2, Resp2, T>(
+        transport: T,
+    ) -> (
+        UnboundedChannel<tarpc::ClientMessage<Req1>, tarpc::Response<Resp1>>,
+        UnboundedChannel<tarpc::Response<Resp2>, tarpc::ClientMessage<Req2>>,
+    )
+    where
+        T: Stream<Item = Result<TwoWayMessage<Req1, Resp2>, io::Error>>,
+        T: Sink<TwoWayMessage<Req2, Resp1>, Error = io::Error>,
+        T: Unpin + Send + 'static,
+        Req1: Send + 'static + Serialize + for<'de> Deserialize<'de>,
+        Resp1: Send + 'static + Serialize + for<'de> Deserialize<'de>,
+        Req2: Send + 'static + Serialize + for<'de> Deserialize<'de>,
+        Resp2: Send + 'static + Serialize + for<'de> Deserialize<'de>,
+    {
+        let (server, server_ret) = tarpc::transport::channel::unbounded();
+        let (client, client_ret) = tarpc::transport::channel::unbounded();
+        let (mut server_sink, server_stream) = server.split();
+        let (mut client_sink, client_stream) = client.split();
+        let (transport_sink, mut transport_stream) = transport.split();
+
+        let (abort_handle, abort_registration) = AbortHandle::new_pair();
+
+        // Task for inbound message handling
+        tokio::spawn(async move {
+            let e: Result<(), ChannelOrIoError> = async move {
+                while let Some(msg) = transport_stream.next().await {
+                    match msg? {
+                        TwoWayMessage::Request(req) => server_sink.send(req).await?,
+                        TwoWayMessage::Response(resp) => client_sink.send(resp).await?,
+                    }
+                }
+                Ok(())
+            }
+            .await;
+
+            match e {
+                Ok(()) => debug!("transport_stream done"),
+                Err(e) => warn!("Error in inbound multiplexing: {}", e),
+            }
+
+            abort_handle.abort();
+        });
+
+        let abortable_sink_channel = Abortable::new(
+            futures::stream::select(
+                server_stream.map_ok(TwoWayMessage::Response),
+                client_stream.map_ok(TwoWayMessage::Request),
+            )
+            .map_err(ChannelOrIoError::ChannelError),
+            abort_registration,
+        );
+
+        // Task for outbound message handling
+        tokio::spawn(
+            abortable_sink_channel
+                .forward(transport_sink.sink_map_err(ChannelOrIoError::IoError))
+                .inspect_ok(|_| debug!("transport_sink done"))
+                .inspect_err(|e| warn!("Error in outbound multiplexing: {}", e)),
+        );
+
+        (server_ret, client_ret)
+    }
 }