Support for light clients.

.github/workflows/rust.yml

@@ -23,7 +23,7 @@ jobs:
       run: cargo +nightly fmt --all -- --check
 
     - name: build
-      run: cargo build --verbose
+      run: cargo build --workspace --verbose
 
     - name: test
-      run: cargo test --verbose
+      run: cargo test --workspace --verbose

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = [".", "proc-macro"]
+members = [".", "light-client", "proc-macro"]
 
 [package]
 name = "substrate-subxt"

light-client/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "substrate-subxt-light-client"
+version = "0.1.0"
+authors = ["David Craven <david@craven.ch>"]
+edition = "2018"
+
+[dependencies]
+async-std = "1.5.0"
+futures = { version = "0.3.4", features = ["compat"] }
+futures01 = { package = "futures", version = "0.1.29" }
+jsonrpsee = "0.1.0"
+log = "0.4.8"
+sc-informant = "0.8.0-alpha.6"
+sc-network = "0.8.0-alpha.6"
+sc-service = "0.8.0-alpha.6"
+serde_json = "1.0.51"
+thiserror = "1.0.15"

light-client/src/lib.rs

@@ -0,0 +1,243 @@
+// Copyright 2019-2020 Parity Technologies (UK) Ltd.
+// This file is part of substrate-subxt.
+//
+// subxt is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// subxt is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with substrate-subxt.  If not, see <http://www.gnu.org/licenses/>.
+
+use async_std::task;
+use futures::{
+    compat::{
+        Compat01As03,
+        Compat01As03Sink,
+        Sink01CompatExt,
+        Stream01CompatExt,
+    },
+    future::poll_fn,
+    sink::SinkExt,
+    stream::{
+        Stream,
+        StreamExt,
+    },
+};
+use futures01::sync::mpsc;
+use jsonrpsee::{
+    common::{
+        Request,
+        Response,
+    },
+    transport::TransportClient,
+};
+use sc_network::config::TransportConfig;
+pub use sc_service::Error as ServiceError;
+use sc_service::{
+    config::{
+        DatabaseConfig,
+        KeystoreConfig,
+        NetworkConfiguration,
+    },
+    AbstractService,
+    ChainSpec,
+    Configuration,
+    Role,
+    RpcSession,
+};
+use std::{
+    future::Future,
+    path::PathBuf,
+    pin::Pin,
+    sync::Arc,
+    task::Poll,
+};
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum LightClientError {
+    #[error("{0}")]
+    Json(#[from] serde_json::Error),
+    #[error("{0}")]
+    Mpsc(#[from] mpsc::SendError<String>),
+}
+
+pub struct LightClientConfig<C: ChainSpec + 'static, S: AbstractService> {
+    pub impl_name: &'static str,
+    pub impl_version: &'static str,
+    pub author: &'static str,
+    pub copyright_start_year: i32,
+    pub db_path: PathBuf,
+    pub builder: fn(Configuration) -> Result<S, sc_service::Error>,
+    pub chain_spec: C,
+}
+
+pub struct LightClient {
+    to_back: Compat01As03Sink<mpsc::Sender<String>, String>,
+    from_back: Compat01As03<mpsc::Receiver<String>>,
+}
+
+impl LightClient {
+    pub fn new<C: ChainSpec + 'static, S: AbstractService>(
+        config: LightClientConfig<C, S>,
+    ) -> Result<Self, ServiceError> {
+        let (to_back, from_front) = mpsc::channel(4);
+        let (to_front, from_back) = mpsc::channel(4);
+        start_light_client(config, from_front, to_front)?;
+        Ok(LightClient {
+            to_back: to_back.sink_compat(),
+            from_back: from_back.compat(),
+        })
+    }
+}
+
+impl TransportClient for LightClient {
+    type Error = LightClientError;
+
+    fn send_request<'a>(
+        &'a mut self,
+        request: Request,
+    ) -> Pin<Box<dyn Future<Output = Result<(), Self::Error>> + Send + 'a>> {
+        Box::pin(async move {
+            let request = serde_json::to_string(&request)?;
+            self.to_back.send(request).await?;
+            Ok(())
+        })
+    }
+
+    fn next_response<'a>(
+        &'a mut self,
+    ) -> Pin<Box<dyn Future<Output = Result<Response, Self::Error>> + Send + 'a>> {
+        Box::pin(async move {
+            let response = self
+                .from_back
+                .next()
+                .await
+                .expect("channel shouldn't close")
+                .unwrap();
+            Ok(serde_json::from_str(&response)?)
+        })
+    }
+}
+
+impl From<LightClient> for jsonrpsee::Client {
+    fn from(client: LightClient) -> Self {
+        let client = jsonrpsee::raw::RawClient::new(client);
+        jsonrpsee::Client::new(client)
+    }
+}
+
+fn start_light_client<C: ChainSpec + 'static, S: AbstractService>(
+    config: LightClientConfig<C, S>,
+    from_front: mpsc::Receiver<String>,
+    to_front: mpsc::Sender<String>,
+) -> Result<(), ServiceError> {
+    let mut network = NetworkConfiguration::new(
+        format!("{} (light client)", config.chain_spec.name()),
+        "unknown",
+        Default::default(),
+        &PathBuf::new(),
+    );
+    network.boot_nodes = config.chain_spec.boot_nodes().to_vec();
+    network.transport = TransportConfig::Normal {
+        enable_mdns: true,
+        allow_private_ipv4: true,
+        wasm_external_transport: None,
+        use_yamux_flow_control: true,
+    };
+    let service_config = Configuration {
+        network,
+        impl_name: config.impl_name,
+        impl_version: config.impl_version,
+        chain_spec: Box::new(config.chain_spec),
+        role: Role::Light,
+        task_executor: Arc::new(move |fut| {
+            task::spawn(fut);
+        }),
+        database: DatabaseConfig::Path {
+            path: config.db_path,
+            cache_size: 64,
+        },
+        keystore: KeystoreConfig::InMemory,
+        max_runtime_instances: 8,
+        announce_block: true,
+
+        telemetry_endpoints: Default::default(),
+        telemetry_external_transport: Default::default(),
+        default_heap_pages: Default::default(),
+        dev_key_seed: Default::default(),
+        disable_grandpa: Default::default(),
+        execution_strategies: Default::default(),
+        force_authoring: Default::default(),
+        offchain_worker: Default::default(),
+        prometheus_config: Default::default(),
+        pruning: Default::default(),
+        rpc_cors: Default::default(),
+        rpc_http: Default::default(),
+        rpc_ws: Default::default(),
+        rpc_ws_max_connections: Default::default(),
+        state_cache_child_ratio: Default::default(),
+        state_cache_size: Default::default(),
+        tracing_receiver: Default::default(),
+        tracing_targets: Default::default(),
+        transaction_pool: Default::default(),
+        wasm_method: Default::default(),
+    };
+
+    log::info!("{}", service_config.impl_name);
+    log::info!("✌️  version {}", service_config.impl_version);
+    log::info!("❤️  by {}, {}", config.author, config.copyright_start_year);
+    log::info!(
+        "📋 Chain specification: {}",
+        service_config.chain_spec.name()
+    );
+    log::info!("🏷  Node name: {}", service_config.network.node_name);
+    log::info!("👤 Role: {:?}", service_config.role);
+
+    // Create the service. This is the most heavy initialization step.
+    let mut service = (config.builder)(service_config)?;
+
+    // Spawn informant.
+    task::spawn(sc_informant::build(
+        &service,
+        sc_informant::OutputFormat::Plain,
+    ));
+
+    // Spawn background task.
+    let session = RpcSession::new(to_front.clone());
+    let mut from_front = from_front.compat();
+    task::spawn(poll_fn(move |cx| {
+        loop {
+            match Pin::new(&mut from_front).poll_next(cx) {
+                Poll::Ready(Some(message)) => {
+                    let mut to_front = to_front.clone().sink_compat();
+                    let message = message.unwrap();
+                    let fut = service.rpc_query(&session, &message);
+                    task::spawn(async move {
+                        if let Some(response) = fut.await {
+                            to_front.send(response).await.ok();
+                        }
+                    });
+                }
+                Poll::Pending => break,
+                Poll::Ready(None) => return Poll::Ready(()),
+            }
+        }
+
+        loop {
+            match Pin::new(&mut service).poll(cx) {
+                Poll::Ready(Ok(())) => return Poll::Ready(()),
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Err(e)) => log::error!("{}", e),
+            }
+        }
+    }));
+
+    Ok(())
+}