Allow generalising over RPC implementation

subxt/Cargo.toml

@@ -13,11 +13,17 @@ description = "Submit extrinsics (transactions) to a substrate node via RPC"
 keywords = ["parity", "substrate", "blockchain"]
 
 [features]
+default = ["jsonrpsee"]
+
 # Activate this to expose functionality only used for integration testing.
 # The exposed functionality is subject to breaking changes at any point,
 # and should not be relied upon.
 integration-tests = []
 
+# Jsonrpsee if the default RPC provider used in Subxt. However, it can be
+# swapped out for an alternative implementation, and so is optional.
+jsonrpsee = ["dep:jsonrpsee"]
+
 [dependencies]
 bitvec = { version = "1.0.0", default-features = false, features = ["alloc"] }
 codec = { package = "parity-scale-codec", version = "3.0.0", default-features = false, features = ["derive", "full", "bit-vec"] }
@@ -26,7 +32,7 @@ scale-value = "0.5.0"
 scale-decode = "0.3.0"
 futures = "0.3.13"
 hex = "0.4.3"
-jsonrpsee = { version = "0.15.1", features = ["async-client", "client-ws-transport"] }
+jsonrpsee = { version = "0.15.1", features = ["async-client", "client-ws-transport", "jsonrpsee-types"], optional = true }
 serde = { version = "1.0.124", features = ["derive"] }
 serde_json = "1.0.64"
 thiserror = "1.0.24"

subxt/src/client/online_client.rs

@@ -12,7 +12,7 @@ use crate::{
     events::EventsClient,
     rpc::{
         Rpc,
-        RpcClient,
+        RpcClientT,
         RuntimeVersion,
     },
     storage::StorageClient,
@@ -52,12 +52,14 @@ struct Inner<T: Config> {
 impl<T: Config> std::fmt::Debug for OnlineClient<T> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Client")
-            .field("rpc", &"<Rpc>")
+            .field("rpc", &"RpcClient")
             .field("inner", &self.inner)
             .finish()
     }
 }
 
+// The default constructors assume Jsonrpsee.
+#[cfg(feature = "jsonrpsee")]
 impl<T: Config> OnlineClient<T> {
     /// Construct a new [`OnlineClient`] using default settings which
     /// point to a locally running node on `ws://127.0.0.1:9944`.
@@ -68,16 +70,20 @@ impl<T: Config> OnlineClient<T> {
 
     /// Construct a new [`OnlineClient`], providing a URL to connect to.
     pub async fn from_url(url: impl AsRef<str>) -> Result<OnlineClient<T>, Error> {
-        let client = crate::rpc::ws_client(url.as_ref()).await?;
+        let client = jsonrpsee_helpers::ws_client(url.as_ref())
+            .await
+            .map_err(|e| crate::error::RpcError(e.to_string()))?;
         OnlineClient::from_rpc_client(client).await
     }
+}
 
-    /// Construct a new [`OnlineClient`] by providing the underlying [`RpcClient`]
-    /// to use to drive the connection.
-    pub async fn from_rpc_client(
-        rpc_client: impl Into<RpcClient>,
+impl<T: Config> OnlineClient<T> {
+    /// Construct a new [`OnlineClient`] by providing an underlying [`RpcClientT`]
+    /// implementation to drive the connection.
+    pub async fn from_rpc_client<R: RpcClientT>(
+        rpc_client: R,
     ) -> Result<OnlineClient<T>, Error> {
-        let rpc = Rpc::new(rpc_client.into());
+        let rpc = Rpc::new(rpc_client);
 
         let (genesis_hash, runtime_version, metadata) = future::join3(
             rpc.genesis_hash(),
@@ -232,3 +238,42 @@ impl<T: Config> ClientRuntimeUpdater<T> {
         Ok(())
     }
 }
+
+// helpers for a jsonrpsee specific OnlineClient.
+#[cfg(feature = "jsonrpsee")]
+mod jsonrpsee_helpers {
+    pub use jsonrpsee::{
+        client_transport::ws::{
+            InvalidUri,
+            Receiver,
+            Sender,
+            Uri,
+            WsTransportClientBuilder,
+        },
+        core::{
+            client::{
+                Client,
+                ClientBuilder,
+            },
+            Error,
+        },
+    };
+
+    /// Build WS RPC client from URL
+    pub async fn ws_client(url: &str) -> Result<Client, Error> {
+        let (sender, receiver) = ws_transport(url).await?;
+        Ok(ClientBuilder::default()
+            .max_notifs_per_subscription(4096)
+            .build_with_tokio(sender, receiver))
+    }
+
+    async fn ws_transport(url: &str) -> Result<(Sender, Receiver), Error> {
+        let url: Uri = url
+            .parse()
+            .map_err(|e: InvalidUri| Error::Transport(e.into()))?;
+        WsTransportClientBuilder::default()
+            .build(url)
+            .await
+            .map_err(|e| Error::Transport(e.into()))
+    }
+}

subxt/src/error.rs

@@ -15,7 +15,6 @@ pub use crate::metadata::{
     InvalidMetadataError,
     MetadataError,
 };
-pub use jsonrpsee::core::error::Error as RequestError;
 pub use scale_value::scale::{
     DecodeError,
     EncodeError,
@@ -36,7 +35,7 @@ pub enum Error {
     Codec(#[from] codec::Error),
     /// Rpc error.
     #[error("Rpc error: {0}")]
-    Rpc(#[from] RequestError),
+    Rpc(#[from] RpcError),
     /// Serde serialization error
     #[error("Serde json error: {0}")]
     Serialization(#[from] serde_json::error::Error),
@@ -102,6 +101,12 @@ impl From<DispatchError> for Error {
     }
 }
 
+/// An RPC error. Since we are generic over the RPC client that is used,
+/// the error is any custom string.
+#[derive(Debug, thiserror::Error)]
+#[error("RPC error: {0}")]
+pub struct RpcError(pub String);
+
 /// This is our attempt to decode a runtime DispatchError. We either
 /// successfully decode it into a [`ModuleError`], or we fail and keep
 /// hold of the bytes, which we can attempt to decode if we have an

subxt/src/events/event_subscription.rs

@@ -8,6 +8,7 @@ use crate::{
     client::OnlineClientT,
     error::Error,
     events::EventsClient,
+    rpc::Subscription,
     Config,
 };
 use derivative::Derivative;
@@ -18,7 +19,6 @@ use futures::{
     Stream,
     StreamExt,
 };
-use jsonrpsee::core::client::Subscription;
 use sp_runtime::traits::Header;
 use std::{
     marker::Unpin,
@@ -32,12 +32,12 @@ pub use super::{
     FilterEvents,
 };
 
-/// A `jsonrpsee` Subscription. This forms a part of the `EventSubscription` type handed back
+/// A Subscription. This forms a part of the `EventSubscription` type handed back
 /// in codegen from `subscribe_finalized`, and is exposed to be used in codegen.
 #[doc(hidden)]
 pub type FinalizedEventSub<Header> = BoxStream<'static, Result<Header, Error>>;
 
-/// A `jsonrpsee` Subscription. This forms a part of the `EventSubscription` type handed back
+/// A Subscription. This forms a part of the `EventSubscription` type handed back
 /// in codegen from `subscribe`, and is exposed to be used in codegen.
 #[doc(hidden)]
 pub type EventSub<Item> = Subscription<Item>;

subxt/src/rpc/jsonrpsee_impl.rs

@@ -0,0 +1,86 @@
+// Copyright 2019-2022 Parity Technologies (UK) Ltd.
+// This file is dual-licensed as Apache-2.0 or GPL-3.0.
+// see LICENSE for license details.
+
+use super::{
+    RpcClientT,
+    RpcFuture,
+    RpcSubscription,
+};
+use crate::error::RpcError;
+use futures::stream::{
+    StreamExt,
+    TryStreamExt,
+};
+use jsonrpsee::{
+    core::client::{
+        Client,
+        ClientT,
+        SubscriptionClientT,
+    },
+    types::ParamsSer,
+};
+use serde_json::value::{
+    RawValue,
+    Value,
+};
+
+impl RpcClientT for Client {
+    fn request_raw<'a>(
+        &'a self,
+        method: &'a str,
+        params: Option<Box<RawValue>>,
+    ) -> RpcFuture<'a, Box<RawValue>> {
+        Box::pin(async move {
+            let params = prep_params_for_jsonrpsee(params)?;
+            let res = ClientT::request(self, method, Some(params))
+                .await
+                .map_err(|e| RpcError(e.to_string()))?;
+            Ok(res)
+        })
+    }
+
+    fn subscribe_raw<'a>(
+        &'a self,
+        sub: &'a str,
+        params: Option<Box<RawValue>>,
+        unsub: &'a str,
+    ) -> RpcFuture<'a, RpcSubscription> {
+        Box::pin(async move {
+            let params = prep_params_for_jsonrpsee(params)?;
+            let sub = SubscriptionClientT::subscribe::<Box<RawValue>>(
+                self,
+                sub,
+                Some(params),
+                unsub,
+            )
+            .await
+            .map_err(|e| RpcError(e.to_string()))?
+            .map_err(|e| RpcError(e.to_string()))
+            .boxed();
+            Ok(sub)
+        })
+    }
+}
+
+// This is ugly; we have to encode to Value's to be compat with the jsonrpc interface.
+// Remove and simplify this once something like https://github.com/paritytech/jsonrpsee/issues/862 is in:
+fn prep_params_for_jsonrpsee(
+    params: Option<Box<RawValue>>,
+) -> Result<ParamsSer<'static>, RpcError> {
+    let params = match params {
+        Some(params) => params,
+        // No params? avoid any work and bail early.
+        None => return Ok(ParamsSer::Array(Vec::new())),
+    };
+    let val = serde_json::to_value(&params).expect("RawValue guarantees valid JSON");
+    let arr = match val {
+        Value::Array(arr) => Ok(arr),
+        _ => {
+            Err(RpcError(format!(
+                "RPC Params are expected to be an array but got {params}"
+            )))
+        }
+    }?;
+    Ok(ParamsSer::Array(arr))
+}

subxt/src/rpc/mod.rs

@@ -0,0 +1,76 @@
+// Copyright 2019-2022 Parity Technologies (UK) Ltd.
+// This file is dual-licensed as Apache-2.0 or GPL-3.0.
+// see LICENSE for license details.
+
+//! RPC types and client for interacting with a substrate node.
+//!
+//! These is used behind the scenes by various `subxt` APIs, but can
+//! also be used directly.
+//!
+//! - [`Rpc`] is the highest level wrapper, and the one you will run into
+//!   first. It contains the higher level methods for interacting with a node.
+//! - [`RpcClient`] is what [`Rpc`] uses to actually talk to a node, offering
+//!   a [`RpcClient::request`] and [`RpcClient::subscribe`] method to do so.
+//! - [`RpcClientT`] is the underlying dynamic RPC implementation. This provides
+//!   the low level [`RpcClientT::request_raw`] and [`RpcClientT::subscribe_raw`]
+//!   methods. This can be swapped out for a custom implementation, but by default
+//!   we'll rely on `jsonrpsee` for this.
+//!
+//! # Example
+//!
+//! Fetching storage keys
+//!
+//! ```no_run
+//! use subxt::{ PolkadotConfig, OnlineClient, storage::StorageKey };
+//!
+//! #[subxt::subxt(runtime_metadata_path = "../artifacts/polkadot_metadata.scale")]
+//! pub mod polkadot {}
+//!
+//! # #[tokio::main]
+//! # async fn main() {
+//! let api = OnlineClient::<PolkadotConfig>::new().await.unwrap();
+//!
+//! let key = polkadot::storage()
+//!     .xcm_pallet()
+//!     .version_notifiers_root()
+//!     .to_bytes();
+//!
+//! // Fetch up to 10 keys.
+//! let keys = api
+//!     .rpc()
+//!     .storage_keys_paged(&key, 10, None, None)
+//!     .await
+//!     .unwrap();
+//!
+//! for key in keys.iter() {
+//!     println!("Key: 0x{}", hex::encode(&key));
+//! }
+//! # }
+//! ```
+
+// Allow an `rpc.rs` file in the `rpc` folder to align better
+// with other file names for their types.
+#![allow(clippy::module_inception)]
+
+#[cfg(feature = "jsonrpsee")]
+mod jsonrpsee_impl;
+
+mod rpc;
+mod rpc_client;
+mod rpc_client_t;
+
+// Expose the `Rpc` struct and any associated types.
+pub use rpc::*;
+
+pub use rpc_client_t::{
+    RpcClientT,
+    RpcFuture,
+    RpcSubscription,
+};
+
+pub use rpc_client::{
+    rpc_params,
+    RpcClient,
+    RpcParams,
+    Subscription,
+};