Add a transports crate & initial Network abstraction

Cargo.toml

@@ -8,12 +8,25 @@ edition = "2021"
 rust-version = "1.65"
 authors = ["Alloy Contributors"]
 license = "MIT OR Apache-2.0"
-homepage = "https://github.com/ethers-rs/next"
-repository = "https://github.com/ethers-rs/next"
+homepage = "https://github.com/alloy-rs/next"
+repository = "https://github.com/alloy-rs/next"
 exclude = ["benches/", "tests/"]
 
 [workspace.metadata.docs.rs]
 all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
-[workspace.dependencies]
+[workspace.dependencies]
+
+# futures
+futures-channel = "0.3"
+
+# jsonrpsee-types
+jsonrpsee-types = "0.16"
+
+# serde
+serde = { version = "1.0", default-features = false, features = ["alloc"] }
+serde_json = { version = "1.0", default-features = false, features = ["alloc"] }
+
+# thiserror
+thiserror = "1.0"

crates/transports/Cargo.toml

@@ -6,3 +6,12 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+base64 = "0.21.0"
+reqwest = { version = "0.11.18", features = ["serde_json", "json"] }
+
+futures-channel.workspace = true
+jsonrpsee-types.workspace = true
+serde.workspace = true
+serde_json = { workspace = true, features = ["raw_value"] }
+thiserror.workspace = true
+

crates/transports/src/call.rs

@@ -0,0 +1,165 @@
+use std::{
+    borrow::Borrow,
+    future::Future,
+    marker::PhantomData,
+    pin::Pin,
+    task::{ready, Context, Poll},
+};
+
+use jsonrpsee_types::ErrorObjectOwned;
+use serde::{Deserialize, Serialize};
+
+use crate::{
+    common::{Id, Request, RpcFuture, RpcOutcome},
+    utils::{from_json, to_json_raw_value},
+    Connection, TransportError,
+};
+
+pub(crate) enum CallState<B, T, Params> {
+    Prepared {
+        connection: B,
+        method: &'static str,
+        params: Params,
+        id: Id<'static>,
+        _pd: PhantomData<T>,
+    },
+    AwaitingResponse {
+        fut: RpcFuture,
+    },
+    Complete,
+    Running,
+}
+impl<B, T, Params> std::fmt::Debug for CallState<B, T, Params> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Prepared { method, id, .. } => f
+                .debug_struct("Prepared")
+                .field("method", method)
+                .field("id", id)
+                .finish(),
+            Self::AwaitingResponse { .. } => f.debug_struct("AwaitingResponse").finish(),
+            Self::Complete => write!(f, "Complete"),
+            Self::Running => write!(f, "Running"),
+        }
+    }
+}
+
+impl<B, T, Params> CallState<B, T, Params> {
+    pub(crate) fn new(
+        connection: B,
+        method: &'static str,
+        params: Params,
+        id: Id<'static>,
+    ) -> CallState<B, T, Params> {
+        Self::Prepared {
+            connection,
+            method,
+            params,
+            id,
+            _pd: PhantomData,
+        }
+    }
+}
+
+impl<B, T, Params> CallState<B, T, Params>
+where
+    B: Borrow<T> + Unpin,
+    T: Connection + Unpin,
+    Params: Serialize + Unpin,
+{
+    fn poll_prepared(&mut self, cx: &mut Context<'_>) -> Poll<RpcOutcome> {
+        let this = std::mem::replace(self, CallState::Running);
+
+        match this {
+            CallState::Prepared {
+                connection,
+                method,
+                params,
+                id,
+                ..
+            } => {
+                let params = to_json_raw_value(&params);
+                if let Err(err) = params {
+                    *self = CallState::Complete;
+                    return Poll::Ready(Err(err));
+                }
+                let params = params.unwrap();
+                let req = Request::owned(id, method, Some(params));
+                let fut = connection.borrow().json_rpc_request(&req);
+                *self = CallState::AwaitingResponse { fut };
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            }
+            _ => panic!(""),
+        }
+    }
+
+    fn poll_awaiting(&mut self, cx: &mut Context<'_>) -> Poll<RpcOutcome> {
+        let this = std::mem::replace(self, CallState::Running);
+        match this {
+            CallState::AwaitingResponse { mut fut } => {
+                if let Poll::Ready(val) = fut.as_mut().poll(cx) {
+                    *self = CallState::Complete;
+                    return Poll::Ready(val);
+                }
+                *self = CallState::AwaitingResponse { fut };
+                Poll::Pending
+            }
+            _ => panic!(""),
+        }
+    }
+}
+
+impl<B, T, Params> Future for CallState<B, T, Params>
+where
+    B: Borrow<T> + Unpin,
+    T: Connection + Unpin,
+    Params: Serialize + Unpin,
+{
+    type Output = RpcOutcome;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let state = self.get_mut();
+        match state {
+            CallState::Prepared { .. } => state.poll_prepared(cx),
+            CallState::AwaitingResponse { .. } => state.poll_awaiting(cx),
+            _ => panic!("Polled in bad state"),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct RpcCall<B, T, Params, Resp> {
+    state: CallState<B, T, Params>,
+    resp: PhantomData<fn() -> Resp>,
+}
+
+impl<B, T, Params, Resp> RpcCall<B, T, Params, Resp> {
+    pub fn new(connection: B, method: &'static str, params: Params, id: Id<'static>) -> Self {
+        Self {
+            state: CallState::new(connection, method, params, id),
+            resp: PhantomData,
+        }
+    }
+}
+
+impl<B, T, Params, Resp> Future for RpcCall<B, T, Params, Resp>
+where
+    B: Borrow<T> + Unpin,
+    T: Connection + Unpin,
+    Params: Serialize + Unpin,
+    Resp: for<'de> Deserialize<'de> + Unpin,
+{
+    type Output = Result<Result<Resp, ErrorObjectOwned>, TransportError>;
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let state = Pin::new(&mut self.get_mut().state);
+        let res = ready!(state.poll(cx));
+
+        match res {
+            Ok(Ok(val)) => Poll::Ready(from_json(val.get()).map(Result::Ok)),
+            Ok(Err(err)) => Poll::Ready(Ok(Err(err))),
+            Err(e) => Poll::Ready(Err(e)),
+        }
+    }
+}

crates/transports/src/common.rs

@@ -0,0 +1,56 @@
+use base64::{engine::general_purpose, Engine};
+use serde_json::value::RawValue;
+use std::{borrow::Cow, fmt, future::Future, pin::Pin};
+
+pub use jsonrpsee_types::{ErrorObject, ErrorResponse, Id, RequestSer as Request, Response};
+
+use crate::TransportError;
+
+#[cfg(target_arch = "wasm32")]
+pub(crate) type DynFuture<'a, T> = Pin<Box<dyn Future<Output = T> + 'a>>;
+#[cfg(not(target_arch = "wasm32"))]
+pub(crate) type DynFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
+
+pub type JsonRpcResult<'a> = Result<Cow<'a, RawValue>, ErrorObject<'a>>;
+pub type JsonRpcResultOwned = JsonRpcResult<'static>;
+
+pub type RpcOutcome = Result<JsonRpcResultOwned, TransportError>;
+pub type BatchRpcOutcome = Result<Vec<JsonRpcResultOwned>, TransportError>;
+
+pub type RpcFuture = DynFuture<'static, RpcOutcome>;
+pub type BatchRpcFuture = DynFuture<'static, BatchRpcOutcome>;
+
+/// Basic or bearer authentication in http or websocket transport
+///
+/// Use to inject username and password or an auth token into requests
+#[derive(Clone, Debug)]
+pub enum Authorization {
+    /// HTTP Basic Auth
+    Basic(String),
+    /// Bearer Auth
+    Bearer(String),
+}
+
+impl Authorization {
+    /// Make a new basic auth
+    pub fn basic(username: impl AsRef<str>, password: impl AsRef<str>) -> Self {
+        let username = username.as_ref();
+        let password = password.as_ref();
+        let auth_secret = general_purpose::STANDARD.encode(format!("{username}:{password}"));
+        Self::Basic(auth_secret)
+    }
+
+    /// Make a new bearer auth
+    pub fn bearer(token: impl Into<String>) -> Self {
+        Self::Bearer(token.into())
+    }
+}
+
+impl fmt::Display for Authorization {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Authorization::Basic(auth_secret) => write!(f, "Basic {auth_secret}"),
+            Authorization::Bearer(token) => write!(f, "Bearer {token}"),
+        }
+    }
+}

crates/transports/src/error.rs

@@ -0,0 +1,31 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum TransportError {
+    /// SerdeJson (de)ser
+    #[error("{err}")]
+    SerdeJson {
+        err: serde_json::Error,
+        text: String,
+    },
+
+    /// Http transport
+    #[error(transparent)]
+    Reqwest(#[from] reqwest::Error),
+}
+
+impl TransportError {
+    pub fn ser_err(err: serde_json::Error) -> Self {
+        Self::SerdeJson {
+            err,
+            text: "".to_string(),
+        }
+    }
+
+    pub fn deser_err(err: serde_json::Error, text: impl AsRef<str>) -> Self {
+        Self::SerdeJson {
+            err,
+            text: text.as_ref().to_string(),
+        }
+    }
+}

crates/transports/src/lib.rs

@@ -1,14 +1,25 @@
-pub fn add(left: usize, right: usize) -> usize {
-    left + right
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn it_works() {
-        let result = add(2, 2);
-        assert_eq!(result, 4);
-    }
-}
+#![warn(
+    missing_debug_implementations,
+    // missing_docs,
+    unreachable_pub,
+    // unused_crate_dependencies
+)]
+#![deny(unused_must_use, rust_2018_idioms)]
+#![doc(test(
+    no_crate_inject,
+    attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
+))]
+
+pub mod common;
+pub(crate) mod utils;
+
+mod error;
+pub use error::TransportError;
+
+mod call;
+
+mod transport;
+pub use transport::{Connection, PubSubConnection};
+
+pub mod transports;
+pub use transports::Http;

crates/transports/src/transport.rs

@@ -0,0 +1,57 @@
+use futures_channel::mpsc::UnboundedReceiver;
+use serde::{Deserialize, Serialize};
+use serde_json::value::RawValue;
+
+use std::{borrow::Cow, fmt::Debug};
+
+use crate::{call::RpcCall, common::*, TransportError};
+
+pub trait Connection: Debug + Send + Sync {
+    fn is_local(&self) -> bool;
+
+    fn increment_id(&self) -> u64;
+
+    fn next_id(&self) -> Id<'static> {
+        Id::Number(self.increment_id())
+    }
+
+    fn json_rpc_request(&self, req: &Request<'_>) -> RpcFuture;
+
+    fn batch_request(&self, reqs: &[Request<'_>]) -> BatchRpcFuture;
+
+    fn request<Params, Resp>(
+        &self,
+        method: &'static str,
+        params: Params,
+    ) -> RpcCall<&Self, Self, Params, Resp>
+    where
+        Self: Sized,
+        Params: Serialize,
+        Resp: for<'de> Deserialize<'de>,
+    {
+        RpcCall::new(self, method, params, self.next_id())
+    }
+}
+
+pub trait PubSubConnection: Connection {
+    #[doc(hidden)]
+    fn uninstall_listener(&self, id: [u8; 32]) -> Result<(), TransportError>;
+
+    #[doc(hidden)]
+    fn install_listener(
+        &self,
+        id: [u8; 32],
+    ) -> Result<UnboundedReceiver<Cow<'_, RawValue>>, TransportError>;
+}
+
+#[cfg(test)]
+mod test {
+    use crate::{Connection, PubSubConnection};
+
+    fn __compile_check() -> Box<dyn Connection> {
+        todo!()
+    }
+    fn __compile_check_pubsub() -> Box<dyn PubSubConnection> {
+        todo!()
+    }
+}