[WIP] IPC

Cargo.toml

@@ -21,6 +21,7 @@ alloy-json-rpc = { version = "0.1.0", path = "crates/json-rpc" }
 alloy-transport = { version = "0.1.0", path = "crates/transport" }
 alloy-pubsub = { version = "0.1.0", path = "crates/pubsub" }
 alloy-transport-http = { version = "0.1.0", path = "crates/transport-http" }
+alloy-transport-ipc = { version = "0.1.0", path = "crates/transport-ipc" }
 alloy-transport-ws = { version = "0.1.0", path = "crates/transport-ws" }
 alloy-networks = { version = "0.1.0", path = "crates/networks" }
 alloy-rpc-types = { version = "0.1.0", path = "crates/rpc-types" }

crates/transport-ipc/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "alloy-transport-ipc"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+authors.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+exclude.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+alloy-transport.workspace = true
+alloy-pubsub.workspace = true
+
+
+tokio = { workspace = true, features = ["io-util", "sync", "rt"] }
+winapi = "0.3.9"
+tracing.workspace = true
+serde_json.workspace = true

crates/transport-ipc/README.md

@@ -0,0 +1,3 @@
+# alloy-transport-ipc
+
+IPC transport implementation.

crates/transport-ipc/src/lib.rs

@@ -0,0 +1,39 @@
+// #![doc = include_str!("../README.md")]
+// #![doc(
+//     html_logo_url = "https://raw.githubusercontent.com/alloy-rs/core/main/assets/alloy.jpg",
+//     html_favicon_url = "https://raw.githubusercontent.com/alloy-rs/core/main/assets/favicon.ico"
+// )]
+// #![warn(
+//     missing_copy_implementations,
+//     missing_debug_implementations,
+//     missing_docs,
+//     unreachable_pub,
+//     clippy::missing_const_for_fn,
+//     rustdoc::all
+// )]
+// #![cfg_attr(not(test), warn(unused_crate_dependencies))]
+// #![deny(unused_must_use, rust_2018_idioms)]
+// #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
+
+#[cfg(unix)]
+mod unix;
+
+#[cfg(windows)]
+mod windows;
+
+use alloy_pubsub::ConnectionInterface;
+
+/// IPC Connection details
+pub struct IpcConnect {
+    path: std::path::PathBuf,
+}
+
+/// An ongoing IPC connection to a backend.
+pub struct IpcBackend<T> {
+    /// The IPC socket connection. For windows this is a named pipe, for unix
+    /// it is a unix socket.
+    pub(crate) stream: T,
+
+    /// The interface to the connection.
+    pub(crate) interface: ConnectionInterface,
+}

crates/transport-ipc/src/unix.rs

@@ -0,0 +1,118 @@
+use crate::{IpcBackend, IpcConnect};
+
+use alloy_pubsub::PubSubConnect;
+use alloy_transport::{TransportError, TransportErrorKind};
+use serde_json::value::RawValue;
+use std::time::Duration;
+use tokio::{net::UnixStream, time::sleep};
+use tracing::error;
+
+const KEEPALIVE: u64 = 10;
+
+impl PubSubConnect for IpcConnect {
+    fn is_local(&self) -> bool {
+        true
+    }
+
+    fn connect<'a: 'b, 'b>(
+        &'a self,
+    ) -> alloy_transport::Pbf<'b, alloy_pubsub::ConnectionHandle, TransportError> {
+        Box::pin(async move {
+            let stream =
+                UnixStream::connect(&self.path).await.map_err(TransportErrorKind::custom)?;
+
+            let (handle, interface) = alloy_pubsub::ConnectionHandle::new();
+            let backend = IpcBackend { stream, interface };
+
+            backend.spawn();
+
+            Ok(handle)
+        })
+    }
+}
+
+impl IpcBackend<UnixStream> {
+    /// Handle a message from the server.
+    pub async fn handle(&mut self, msg: Box<RawValue>) -> Result<(), ()> {
+        todo!()
+    }
+
+    /// Send a message to the server.
+    pub async fn send(&mut self, msg: Box<RawValue>) -> Result<(), ()> {
+        todo!()
+    }
+
+    fn spawn(self) {
+        tokio::spawn(async move {
+            let mut err = false;
+            let keepalive = sleep(Duration::from_secs(KEEPALIVE));
+
+            loop {
+                // We bias the loop as follows
+                // 1. New dispatch.
+                // 2. Keepalive.
+                // 3. Response or notification from server.
+                // This ensures that keepalive is sent only if no other messages
+                // have been sent in the last 10 seconds. And prioritizes new
+                // dispatches over responses from the server. This will fail if
+                // the client saturates the task with dispatches, but that's
+                // probably not a big deal.
+                tokio::select! {
+                    biased;
+                    // we've received a new dispatch, so we send it via
+                    // websocket. We handle new work before processing any
+                    // responses from the server.
+                    inst = self.interface.recv_from_frontend() => {
+                        match inst {
+                            Some(msg) => {
+                                // Reset the keepalive timer.
+                                keepalive.set(sleep(Duration::from_secs(KEEPALIVE)));
+                                if let Err(e) = self.send(msg).await {
+                                    error!(err = %e, "WS connection error");
+                                    err = true;
+                                    break
+                                }
+                            },
+                            // dispatcher has gone away, or shutdown was received
+                            None => {
+                                break
+                            },
+                        }
+                    },
+                    // Send a ping to the server, if no other messages have been
+                    // sent in the last 10 seconds.
+                    _ = &mut keepalive => {
+                        // Reset the keepalive timer.
+                        keepalive.set(sleep(Duration::from_secs(KEEPALIVE)));
+                        if let Err(e) = self.stream.send(Message::Ping(vec![])).await {
+                            error!(err = %e, "WS connection error");
+                            err = true;
+                            break
+                        }
+                    }
+                    resp = self.stream.next() => {
+                        match resp {
+                            Some(Ok(item)) => {
+                                err = self.handle(item).await.is_err();
+                                if err { break }
+                            },
+                            Some(Err(e)) => {
+                                tracing::error!(err = %e, "WS connection error");
+                                err = true;
+                                break
+                            }
+                            None => {
+                                error!("WS server has gone away");
+                                err = true;
+                                break
+                            },
+                        }
+                    }
+                }
+            }
+            if err {
+                self.interface.close_with_error();
+            }
+        });
+    }
+}