mirror of
https://github.com/pezkuwichain/pezkuwi-subxt.git
synced 2026-06-12 03:01:07 +00:00
Rework light client (#1475)
* WIP second pass over light client code for simpler API * First pass new light client * pub(crate) LightClientRpc::new_raw(), and fmt * Update examples and add back a way to configure boot nodes and fetch chainspec from a URL * Fix light client examples * remove unused deps and tidy lightclient feature flags * fix wasm error * LightClientRpc can be cloned * update light client tests * Other small fixes * exclude mod unless jsonrpsee * Fix wasm-lightclient-tests * add back docsrs bit and web+native feature flag compile error * update book and light client example names * fix docs
This commit is contained in:
James Wilson
+330
-363
@@ -1,43 +1,47 @@
|
||||
// Copyright 2019-2023 Parity Technologies (UK) Ltd.
|
||||
// Copyright 2019-2024 Parity Technologies (UK) Ltd.
|
||||
// This file is dual-licensed as Apache-2.0 or GPL-3.0.
|
||||
// see LICENSE for license details.
|
||||
|
||||
use futures::stream::StreamExt;
|
||||
use futures_util::future::{self, Either};
|
||||
use serde::Deserialize;
|
||||
use crate::rpc::RpcResponse;
|
||||
use crate::shared_client::SharedClient;
|
||||
use crate::{JsonRpcError, LightClientRpcError};
|
||||
use futures::{stream::StreamExt, FutureExt};
|
||||
use serde_json::value::RawValue;
|
||||
use smoldot_light::platform::PlatformRef;
|
||||
use std::{collections::HashMap, str::FromStr};
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
use tokio_stream::wrappers::UnboundedReceiverStream;
|
||||
|
||||
use crate::client::AddedChain;
|
||||
const LOG_TARGET: &str = "subxt-light-client-background-task";
|
||||
|
||||
use super::LightClientRpcError;
|
||||
use smoldot_light::ChainId;
|
||||
|
||||
const LOG_TARGET: &str = "subxt-light-client-background";
|
||||
|
||||
/// The response of an RPC method.
|
||||
/// Response from [`BackgroundTaskHandle::request()`].
|
||||
pub type MethodResponse = Result<Box<RawValue>, LightClientRpcError>;
|
||||
|
||||
/// Response from [`BackgroundTaskHandle::subscribe()`].
|
||||
pub type SubscriptionResponse = Result<
|
||||
(
|
||||
SubscriptionId,
|
||||
mpsc::UnboundedReceiver<Result<Box<RawValue>, JsonRpcError>>,
|
||||
),
|
||||
LightClientRpcError,
|
||||
>;
|
||||
|
||||
/// Type of subscription IDs we can get back.
|
||||
pub type SubscriptionId = String;
|
||||
|
||||
/// Message protocol between the front-end client that submits the RPC requests
|
||||
/// and the backend handler that produces responses from the chain.
|
||||
///
|
||||
/// The light client uses a single object [`smoldot_light::JsonRpcResponses`] to
|
||||
/// handle all requests and subscriptions from a chain. A background task is spawned
|
||||
/// to multiplex the rpc responses and to provide them back to their rightful submitters.
|
||||
/// and the background task which fetches responses from Smoldot. Hidden behind
|
||||
/// the [`BackgroundTaskHandle`].
|
||||
#[derive(Debug)]
|
||||
pub enum FromSubxt {
|
||||
enum Message {
|
||||
/// The RPC method request.
|
||||
Request {
|
||||
/// The method of the request.
|
||||
method: String,
|
||||
/// The parameters of the request.
|
||||
params: String,
|
||||
/// Channel used to send back the result.
|
||||
params: Option<Box<RawValue>>,
|
||||
/// Channel used to send back the method response.
|
||||
sender: oneshot::Sender<MethodResponse>,
|
||||
/// The ID of the chain used to identify the chain.
|
||||
chain_id: ChainId,
|
||||
},
|
||||
/// The RPC subscription (pub/sub) request.
|
||||
Subscription {
|
||||
@@ -46,29 +50,160 @@ pub enum FromSubxt {
|
||||
/// The method to unsubscribe.
|
||||
unsubscribe_method: String,
|
||||
/// The parameters of the request.
|
||||
params: String,
|
||||
/// Channel used to send back the subscription ID if successful.
|
||||
sub_id: oneshot::Sender<MethodResponse>,
|
||||
/// Channel used to send back the notifications.
|
||||
sender: mpsc::UnboundedSender<Box<RawValue>>,
|
||||
/// The ID of the chain used to identify the chain.
|
||||
chain_id: ChainId,
|
||||
params: Option<Box<RawValue>>,
|
||||
/// Channel used to send back the subscription response.
|
||||
sender: oneshot::Sender<SubscriptionResponse>,
|
||||
},
|
||||
}
|
||||
|
||||
/// Background task data.
|
||||
#[allow(clippy::type_complexity)]
|
||||
pub struct BackgroundTask<TPlatform: PlatformRef, TChain> {
|
||||
/// Smoldot light client implementation that leverages the exposed platform.
|
||||
client: smoldot_light::Client<TPlatform, TChain>,
|
||||
/// Per-chain data.
|
||||
chain_data: HashMap<smoldot_light::ChainId, ChainData>,
|
||||
/// A handle to communicate with the background task.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct BackgroundTaskHandle {
|
||||
to_backend: mpsc::UnboundedSender<Message>,
|
||||
}
|
||||
|
||||
/// The data that we store for each chain.
|
||||
#[derive(Default)]
|
||||
struct ChainData {
|
||||
/// Generates an unique monotonically increasing ID for each chain.
|
||||
impl BackgroundTaskHandle {
|
||||
/// Make an RPC request via the background task.
|
||||
pub async fn request(&self, method: String, params: Option<Box<RawValue>>) -> MethodResponse {
|
||||
let (tx, rx) = oneshot::channel();
|
||||
self.to_backend
|
||||
.send(Message::Request {
|
||||
method,
|
||||
params,
|
||||
sender: tx,
|
||||
})
|
||||
.map_err(|_e| LightClientRpcError::BackgroundTaskDropped)?;
|
||||
|
||||
match rx.await {
|
||||
Err(_e) => Err(LightClientRpcError::BackgroundTaskDropped),
|
||||
Ok(response) => response,
|
||||
}
|
||||
}
|
||||
|
||||
/// Subscribe to some RPC method via the background task.
|
||||
pub async fn subscribe(
|
||||
&self,
|
||||
method: String,
|
||||
params: Option<Box<RawValue>>,
|
||||
unsubscribe_method: String,
|
||||
) -> SubscriptionResponse {
|
||||
let (tx, rx) = oneshot::channel();
|
||||
self.to_backend
|
||||
.send(Message::Subscription {
|
||||
method,
|
||||
params,
|
||||
unsubscribe_method,
|
||||
sender: tx,
|
||||
})
|
||||
.map_err(|_e| LightClientRpcError::BackgroundTaskDropped)?;
|
||||
|
||||
match rx.await {
|
||||
Err(_e) => Err(LightClientRpcError::BackgroundTaskDropped),
|
||||
Ok(response) => response,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A background task which runs with [`BackgroundTask::run()`] and manages messages
|
||||
/// coming to/from Smoldot.
|
||||
#[allow(clippy::type_complexity)]
|
||||
pub struct BackgroundTask<TPlatform: PlatformRef, TChain> {
|
||||
channels: BackgroundTaskChannels,
|
||||
data: BackgroundTaskData<TPlatform, TChain>,
|
||||
}
|
||||
|
||||
impl<TPlatform: PlatformRef, TChain> BackgroundTask<TPlatform, TChain> {
|
||||
/// Constructs a new [`BackgroundTask`].
|
||||
pub(crate) fn new(
|
||||
client: SharedClient<TPlatform, TChain>,
|
||||
chain_id: smoldot_light::ChainId,
|
||||
from_back: smoldot_light::JsonRpcResponses,
|
||||
) -> (BackgroundTask<TPlatform, TChain>, BackgroundTaskHandle) {
|
||||
let (tx, rx) = mpsc::unbounded_channel();
|
||||
|
||||
let bg_task = BackgroundTask {
|
||||
channels: BackgroundTaskChannels {
|
||||
from_front: UnboundedReceiverStream::new(rx),
|
||||
from_back,
|
||||
},
|
||||
data: BackgroundTaskData {
|
||||
client,
|
||||
chain_id,
|
||||
last_request_id: 0,
|
||||
pending_subscriptions: HashMap::new(),
|
||||
requests: HashMap::new(),
|
||||
subscriptions: HashMap::new(),
|
||||
},
|
||||
};
|
||||
|
||||
let bg_handle = BackgroundTaskHandle { to_backend: tx };
|
||||
|
||||
(bg_task, bg_handle)
|
||||
}
|
||||
|
||||
/// Run the background task, which:
|
||||
/// - Forwards messages/subscription requests to Smoldot from the front end.
|
||||
/// - Forwards responses back from Smoldot to the front end.
|
||||
pub async fn run(self) {
|
||||
let chain_id = self.data.chain_id;
|
||||
let mut channels = self.channels;
|
||||
let mut data = self.data;
|
||||
|
||||
loop {
|
||||
tokio::pin! {
|
||||
let from_front_fut = channels.from_front.next().fuse();
|
||||
let from_back_fut = channels.from_back.next().fuse();
|
||||
}
|
||||
|
||||
futures::select! {
|
||||
// Message coming from the front end/client.
|
||||
front_message = from_front_fut => {
|
||||
let Some(message) = front_message else {
|
||||
tracing::trace!(target: LOG_TARGET, "Subxt channel closed");
|
||||
break;
|
||||
};
|
||||
tracing::trace!(
|
||||
target: LOG_TARGET,
|
||||
"Received register message {:?}",
|
||||
message
|
||||
);
|
||||
|
||||
data.handle_requests(message).await;
|
||||
},
|
||||
// Message coming from Smoldot.
|
||||
back_message = from_back_fut => {
|
||||
let Some(back_message) = back_message else {
|
||||
tracing::trace!(target: LOG_TARGET, "Smoldot RPC responses channel closed");
|
||||
break;
|
||||
};
|
||||
tracing::trace!(
|
||||
target: LOG_TARGET,
|
||||
"Received smoldot RPC chain {:?} result {:?}",
|
||||
chain_id, back_message
|
||||
);
|
||||
|
||||
data.handle_rpc_response(back_message);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tracing::trace!(target: LOG_TARGET, "Task closed");
|
||||
}
|
||||
}
|
||||
|
||||
struct BackgroundTaskChannels {
|
||||
/// Messages sent into this background task from the front end.
|
||||
from_front: UnboundedReceiverStream<Message>,
|
||||
/// Messages sent into the background task from Smoldot.
|
||||
from_back: smoldot_light::JsonRpcResponses,
|
||||
}
|
||||
|
||||
struct BackgroundTaskData<TPlatform: PlatformRef, TChain> {
|
||||
/// A smoldot light client that can be shared.
|
||||
client: SharedClient<TPlatform, TChain>,
|
||||
/// Knowing the chain ID helps with debugging, but isn't overwise necessary.
|
||||
chain_id: smoldot_light::ChainId,
|
||||
/// Know which Id to use next for new requests/subscriptions.
|
||||
last_request_id: usize,
|
||||
/// Map the request ID of a RPC method to the frontend `Sender`.
|
||||
requests: HashMap<usize, oneshot::Sender<MethodResponse>>,
|
||||
@@ -78,20 +213,12 @@ struct ChainData {
|
||||
/// The RPC method request is made in the background and the response should
|
||||
/// not be sent back to the user.
|
||||
/// Map the request ID of a RPC method to the frontend `Sender`.
|
||||
id_to_subscription: HashMap<usize, PendingSubscription>,
|
||||
pending_subscriptions: HashMap<usize, PendingSubscription>,
|
||||
/// Map the subscription ID to the frontend `Sender`.
|
||||
///
|
||||
/// The subscription ID is entirely generated by the node (smoldot). Therefore, it is
|
||||
/// possible for two distinct subscriptions of different chains to have the same subscription ID.
|
||||
subscriptions: HashMap<usize, ActiveSubscription>,
|
||||
}
|
||||
|
||||
impl ChainData {
|
||||
/// Fetch and increment the request ID.
|
||||
fn next_id(&mut self) -> usize {
|
||||
self.last_request_id = self.last_request_id.wrapping_add(1);
|
||||
self.last_request_id
|
||||
}
|
||||
subscriptions: HashMap<String, ActiveSubscription>,
|
||||
}
|
||||
|
||||
/// The state needed to resolve the subscription ID and send
|
||||
@@ -100,72 +227,52 @@ struct PendingSubscription {
|
||||
/// Send the method response ID back to the user.
|
||||
///
|
||||
/// It contains the subscription ID if successful, or an JSON RPC error object.
|
||||
sub_id_sender: oneshot::Sender<MethodResponse>,
|
||||
/// The subscription state that is added to the `subscriptions` map only
|
||||
/// if the subscription ID is successfully sent back to the user.
|
||||
subscription_state: ActiveSubscription,
|
||||
}
|
||||
|
||||
impl PendingSubscription {
|
||||
/// Transforms the pending subscription into an active subscription.
|
||||
fn into_parts(self) -> (oneshot::Sender<MethodResponse>, ActiveSubscription) {
|
||||
(self.sub_id_sender, self.subscription_state)
|
||||
}
|
||||
}
|
||||
|
||||
/// The state of the subscription.
|
||||
struct ActiveSubscription {
|
||||
/// Channel to send the subscription notifications back to frontend.
|
||||
sender: mpsc::UnboundedSender<Box<RawValue>>,
|
||||
response_sender: oneshot::Sender<SubscriptionResponse>,
|
||||
/// The unsubscribe method to call when the user drops the receiver
|
||||
/// part of the channel.
|
||||
unsubscribe_method: String,
|
||||
}
|
||||
|
||||
impl<TPlatform: PlatformRef, TChain> BackgroundTask<TPlatform, TChain> {
|
||||
/// Constructs a new [`BackgroundTask`].
|
||||
pub fn new(
|
||||
client: smoldot_light::Client<TPlatform, TChain>,
|
||||
) -> BackgroundTask<TPlatform, TChain> {
|
||||
BackgroundTask {
|
||||
client,
|
||||
chain_data: Default::default(),
|
||||
}
|
||||
}
|
||||
/// The state of the subscription.
|
||||
struct ActiveSubscription {
|
||||
/// Channel to send the subscription notifications back to frontend.
|
||||
notification_sender: mpsc::UnboundedSender<Result<Box<RawValue>, JsonRpcError>>,
|
||||
/// The unsubscribe method to call when the user drops the receiver
|
||||
/// part of the channel.
|
||||
unsubscribe_method: String,
|
||||
}
|
||||
|
||||
fn for_chain_id(
|
||||
&mut self,
|
||||
chain_id: smoldot_light::ChainId,
|
||||
) -> (
|
||||
&mut ChainData,
|
||||
&mut smoldot_light::Client<TPlatform, TChain>,
|
||||
) {
|
||||
let chain_data = self.chain_data.entry(chain_id).or_default();
|
||||
let client = &mut self.client;
|
||||
(chain_data, client)
|
||||
impl<TPlatform: PlatformRef, TChain> BackgroundTaskData<TPlatform, TChain> {
|
||||
/// Fetch and increment the request ID.
|
||||
fn next_id(&mut self) -> usize {
|
||||
self.last_request_id = self.last_request_id.wrapping_add(1);
|
||||
self.last_request_id
|
||||
}
|
||||
|
||||
/// Handle the registration messages received from the user.
|
||||
async fn handle_requests(&mut self, message: FromSubxt) {
|
||||
async fn handle_requests(&mut self, message: Message) {
|
||||
match message {
|
||||
FromSubxt::Request {
|
||||
Message::Request {
|
||||
method,
|
||||
params,
|
||||
sender,
|
||||
chain_id,
|
||||
} => {
|
||||
let (chain_data, client) = self.for_chain_id(chain_id);
|
||||
let id = chain_data.next_id();
|
||||
let id = self.next_id();
|
||||
let chain_id = self.chain_id;
|
||||
|
||||
let params = match ¶ms {
|
||||
Some(params) => params.get(),
|
||||
None => "null",
|
||||
};
|
||||
let request = format!(
|
||||
r#"{{"jsonrpc":"2.0","id":"{}", "method":"{}","params":{}}}"#,
|
||||
id, method, params
|
||||
);
|
||||
|
||||
chain_data.requests.insert(id, sender);
|
||||
self.requests.insert(id, sender);
|
||||
tracing::trace!(target: LOG_TARGET, "Tracking request id={id} chain={chain_id:?}");
|
||||
|
||||
let result = client.json_rpc_request(request, chain_id);
|
||||
let result = self.client.json_rpc_request(request, chain_id);
|
||||
if let Err(err) = result {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
@@ -173,14 +280,14 @@ impl<TPlatform: PlatformRef, TChain> BackgroundTask<TPlatform, TChain> {
|
||||
err.to_string()
|
||||
);
|
||||
|
||||
let sender = chain_data
|
||||
let sender = self
|
||||
.requests
|
||||
.remove(&id)
|
||||
.expect("Channel is inserted above; qed");
|
||||
|
||||
// Send the error back to frontend.
|
||||
if sender
|
||||
.send(Err(LightClientRpcError::Request(err.to_string())))
|
||||
.send(Err(LightClientRpcError::SmoldotError(err.to_string())))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
@@ -192,52 +299,49 @@ impl<TPlatform: PlatformRef, TChain> BackgroundTask<TPlatform, TChain> {
|
||||
tracing::trace!(target: LOG_TARGET, "Submitted to smoldot request with id={id}");
|
||||
}
|
||||
}
|
||||
FromSubxt::Subscription {
|
||||
Message::Subscription {
|
||||
method,
|
||||
unsubscribe_method,
|
||||
params,
|
||||
sub_id,
|
||||
sender,
|
||||
chain_id,
|
||||
} => {
|
||||
let (chain_data, client) = self.for_chain_id(chain_id);
|
||||
let id = chain_data.next_id();
|
||||
let id = self.next_id();
|
||||
let chain_id = self.chain_id;
|
||||
|
||||
// For subscriptions we need to make a plain RPC request to the subscription method.
|
||||
// The server will return as a result the subscription ID.
|
||||
let params = match ¶ms {
|
||||
Some(params) => params.get(),
|
||||
None => "null",
|
||||
};
|
||||
let request = format!(
|
||||
r#"{{"jsonrpc":"2.0","id":"{}", "method":"{}","params":{}}}"#,
|
||||
id, method, params
|
||||
);
|
||||
|
||||
tracing::trace!(target: LOG_TARGET, "Tracking subscription request id={id} chain={chain_id:?}");
|
||||
let subscription_id_state = PendingSubscription {
|
||||
sub_id_sender: sub_id,
|
||||
subscription_state: ActiveSubscription {
|
||||
sender,
|
||||
unsubscribe_method,
|
||||
},
|
||||
let pending_subscription = PendingSubscription {
|
||||
response_sender: sender,
|
||||
unsubscribe_method,
|
||||
};
|
||||
chain_data
|
||||
.id_to_subscription
|
||||
.insert(id, subscription_id_state);
|
||||
self.pending_subscriptions.insert(id, pending_subscription);
|
||||
|
||||
let result = client.json_rpc_request(request, chain_id);
|
||||
let result = self.client.json_rpc_request(request, chain_id);
|
||||
if let Err(err) = result {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send RPC request to lightclient {:?}",
|
||||
err.to_string()
|
||||
);
|
||||
let subscription_id_state = chain_data
|
||||
.id_to_subscription
|
||||
let subscription_id_state = self
|
||||
.pending_subscriptions
|
||||
.remove(&id)
|
||||
.expect("Channels are inserted above; qed");
|
||||
|
||||
// Send the error back to frontend.
|
||||
if subscription_id_state
|
||||
.sub_id_sender
|
||||
.send(Err(LightClientRpcError::Request(err.to_string())))
|
||||
.response_sender
|
||||
.send(Err(LightClientRpcError::SmoldotError(err.to_string())))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
@@ -253,20 +357,75 @@ impl<TPlatform: PlatformRef, TChain> BackgroundTask<TPlatform, TChain> {
|
||||
}
|
||||
|
||||
/// Parse the response received from the light client and sent it to the appropriate user.
|
||||
fn handle_rpc_response(&mut self, chain_id: smoldot_light::ChainId, response: String) {
|
||||
tracing::trace!(target: LOG_TARGET, "Received from smoldot response={response} chain={chain_id:?}");
|
||||
let (chain_data, _client) = self.for_chain_id(chain_id);
|
||||
fn handle_rpc_response(&mut self, response: String) {
|
||||
let chain_id = self.chain_id;
|
||||
tracing::trace!(target: LOG_TARGET, "Received from smoldot response='{response}' chain={chain_id:?}");
|
||||
|
||||
match RpcResponse::from_str(&response) {
|
||||
Ok(RpcResponse::Error { id, error }) => {
|
||||
Ok(RpcResponse::Method { id, result }) => {
|
||||
let Ok(id) = id.parse::<usize>() else {
|
||||
tracing::warn!(target: LOG_TARGET, "Cannot send response. Id={id} chain={chain_id:?} is not a valid number");
|
||||
return;
|
||||
};
|
||||
|
||||
// Send the response back.
|
||||
if let Some(sender) = self.requests.remove(&id) {
|
||||
if sender.send(Ok(result)).is_err() {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send method response to id={id} chain={chain_id:?}",
|
||||
);
|
||||
}
|
||||
} else if let Some(pending_subscription) = self.pending_subscriptions.remove(&id) {
|
||||
let Ok(sub_id) = serde_json::from_str::<SubscriptionId>(result.get()) else {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Subscription id='{result}' chain={chain_id:?} is not a valid string",
|
||||
);
|
||||
return;
|
||||
};
|
||||
|
||||
tracing::trace!(target: LOG_TARGET, "Received subscription id={sub_id} chain={chain_id:?}");
|
||||
|
||||
let (sub_tx, sub_rx) = mpsc::unbounded_channel();
|
||||
|
||||
// Send the method response and a channel to receive notifications back.
|
||||
if pending_subscription
|
||||
.response_sender
|
||||
.send(Ok((sub_id.clone(), sub_rx)))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send subscription ID response to id={id} chain={chain_id:?}",
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
// Store the other end of the notif channel to send future subscription notifications to.
|
||||
self.subscriptions.insert(
|
||||
sub_id,
|
||||
ActiveSubscription {
|
||||
notification_sender: sub_tx,
|
||||
unsubscribe_method: pending_subscription.unsubscribe_method,
|
||||
},
|
||||
);
|
||||
} else {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Response id={id} chain={chain_id:?} is not tracked",
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(RpcResponse::MethodError { id, error }) => {
|
||||
let Ok(id) = id.parse::<usize>() else {
|
||||
tracing::warn!(target: LOG_TARGET, "Cannot send error. Id={id} chain={chain_id:?} is not a valid number");
|
||||
return;
|
||||
};
|
||||
|
||||
if let Some(sender) = chain_data.requests.remove(&id) {
|
||||
if let Some(sender) = self.requests.remove(&id) {
|
||||
if sender
|
||||
.send(Err(LightClientRpcError::Request(error.to_string())))
|
||||
.send(Err(LightClientRpcError::JsonRpcError(JsonRpcError(error))))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
@@ -274,12 +433,10 @@ impl<TPlatform: PlatformRef, TChain> BackgroundTask<TPlatform, TChain> {
|
||||
"Cannot send method response to id={id} chain={chain_id:?}",
|
||||
);
|
||||
}
|
||||
} else if let Some(subscription_id_state) =
|
||||
chain_data.id_to_subscription.remove(&id)
|
||||
{
|
||||
} else if let Some(subscription_id_state) = self.pending_subscriptions.remove(&id) {
|
||||
if subscription_id_state
|
||||
.sub_id_sender
|
||||
.send(Err(LightClientRpcError::Request(error.to_string())))
|
||||
.response_sender
|
||||
.send(Err(LightClientRpcError::JsonRpcError(JsonRpcError(error))))
|
||||
.is_err()
|
||||
{
|
||||
tracing::warn!(
|
||||
@@ -289,93 +446,44 @@ impl<TPlatform: PlatformRef, TChain> BackgroundTask<TPlatform, TChain> {
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(RpcResponse::Method { id, result }) => {
|
||||
let Ok(id) = id.parse::<usize>() else {
|
||||
tracing::warn!(target: LOG_TARGET, "Cannot send response. Id={id} chain={chain_id:?} is not a valid number");
|
||||
return;
|
||||
};
|
||||
|
||||
// Send the response back.
|
||||
if let Some(sender) = chain_data.requests.remove(&id) {
|
||||
if sender.send(Ok(result)).is_err() {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send method response to id={id} chain={chain_id:?}",
|
||||
);
|
||||
}
|
||||
} else if let Some(pending_subscription) = chain_data.id_to_subscription.remove(&id)
|
||||
{
|
||||
let Ok(sub_id) = result
|
||||
.get()
|
||||
.trim_start_matches('"')
|
||||
.trim_end_matches('"')
|
||||
.parse::<usize>()
|
||||
else {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Subscription id={result} chain={chain_id:?} is not a valid number",
|
||||
);
|
||||
return;
|
||||
};
|
||||
|
||||
tracing::trace!(target: LOG_TARGET, "Received subscription id={sub_id} chain={chain_id:?}");
|
||||
|
||||
let (sub_id_sender, active_subscription) = pending_subscription.into_parts();
|
||||
if sub_id_sender.send(Ok(result)).is_err() {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Cannot send method response to id={id} chain={chain_id:?}",
|
||||
);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// Track this subscription ID if send is successful.
|
||||
chain_data.subscriptions.insert(sub_id, active_subscription);
|
||||
} else {
|
||||
Ok(RpcResponse::Notification {
|
||||
method,
|
||||
subscription_id,
|
||||
result,
|
||||
}) => {
|
||||
let Some(active_subscription) = self.subscriptions.get_mut(&subscription_id) else {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Response id={id} chain={chain_id:?} is not tracked",
|
||||
"Subscription response id={subscription_id} chain={chain_id:?} method={method} is not tracked",
|
||||
);
|
||||
return;
|
||||
};
|
||||
if active_subscription
|
||||
.notification_sender
|
||||
.send(Ok(result))
|
||||
.is_err()
|
||||
{
|
||||
self.unsubscribe(&subscription_id, chain_id);
|
||||
}
|
||||
}
|
||||
Ok(RpcResponse::Subscription { method, id, result }) => {
|
||||
let Ok(id) = id.parse::<usize>() else {
|
||||
tracing::warn!(target: LOG_TARGET, "Cannot send subscription. Id={id} chain={chain_id:?} is not a valid number");
|
||||
return;
|
||||
};
|
||||
|
||||
let Some(subscription_state) = chain_data.subscriptions.get_mut(&id) else {
|
||||
Ok(RpcResponse::NotificationError {
|
||||
method,
|
||||
subscription_id,
|
||||
error,
|
||||
}) => {
|
||||
let Some(active_subscription) = self.subscriptions.get_mut(&subscription_id) else {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Subscription response id={id} chain={chain_id:?} method={method} is not tracked",
|
||||
"Subscription error id={subscription_id} chain={chain_id:?} method={method} is not tracked",
|
||||
);
|
||||
return;
|
||||
};
|
||||
if subscription_state.sender.send(result).is_ok() {
|
||||
// Nothing else to do, user is informed about the notification.
|
||||
return;
|
||||
}
|
||||
|
||||
// User dropped the receiver, unsubscribe from the method and remove internal tracking.
|
||||
let Some(subscription_state) = chain_data.subscriptions.remove(&id) else {
|
||||
// State is checked to be some above, so this should never happen.
|
||||
return;
|
||||
};
|
||||
// Make a call to unsubscribe from this method.
|
||||
let unsub_id = chain_data.next_id();
|
||||
let request = format!(
|
||||
r#"{{"jsonrpc":"2.0","id":"{}", "method":"{}","params":["{}"]}}"#,
|
||||
unsub_id, subscription_state.unsubscribe_method, id
|
||||
);
|
||||
|
||||
if let Err(err) = self.client.json_rpc_request(request, chain_id) {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Failed to unsubscribe id={id:?} chain={chain_id:?} method={:?} err={err:?}", subscription_state.unsubscribe_method
|
||||
);
|
||||
} else {
|
||||
tracing::debug!(target: LOG_TARGET,"Unsubscribe id={id:?} chain={chain_id:?} method={:?}", subscription_state.unsubscribe_method);
|
||||
if active_subscription
|
||||
.notification_sender
|
||||
.send(Err(JsonRpcError(error)))
|
||||
.is_err()
|
||||
{
|
||||
self.unsubscribe(&subscription_id, chain_id);
|
||||
}
|
||||
}
|
||||
Err(err) => {
|
||||
@@ -384,169 +492,28 @@ impl<TPlatform: PlatformRef, TChain> BackgroundTask<TPlatform, TChain> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Perform the main background task:
|
||||
/// - receiving requests from subxt RPC method / subscriptions
|
||||
/// - provides the results from the light client back to users.
|
||||
pub async fn start_task(
|
||||
&mut self,
|
||||
from_subxt: mpsc::UnboundedReceiver<FromSubxt>,
|
||||
from_node: Vec<AddedChain>,
|
||||
) {
|
||||
let from_subxt_event = tokio_stream::wrappers::UnboundedReceiverStream::new(from_subxt);
|
||||
// Unsubscribe from a subscription.
|
||||
fn unsubscribe(&mut self, subscription_id: &str, chain_id: smoldot_light::ChainId) {
|
||||
let Some(active_subscription) = self.subscriptions.remove(subscription_id) else {
|
||||
// Subscription doesn't exist so nothing more to do.
|
||||
return;
|
||||
};
|
||||
|
||||
let from_node = from_node.into_iter().map(|rpc| {
|
||||
Box::pin(futures::stream::unfold(rpc, |mut rpc| async move {
|
||||
let response = rpc.rpc_responses.next().await;
|
||||
Some(((response, rpc.chain_id), rpc))
|
||||
}))
|
||||
});
|
||||
let stream_combinator = futures::stream::select_all(from_node);
|
||||
// Build a call to unsubscribe from this method.
|
||||
let unsub_id = self.next_id();
|
||||
let request = format!(
|
||||
r#"{{"jsonrpc":"2.0","id":"{}", "method":"{}","params":["{}"]}}"#,
|
||||
unsub_id, active_subscription.unsubscribe_method, subscription_id
|
||||
);
|
||||
|
||||
tokio::pin!(from_subxt_event, stream_combinator);
|
||||
|
||||
let mut from_subxt_event_fut = from_subxt_event.next();
|
||||
let mut from_node_event_fut = stream_combinator.next();
|
||||
|
||||
loop {
|
||||
match future::select(from_subxt_event_fut, from_node_event_fut).await {
|
||||
// Message received from subxt.
|
||||
Either::Left((subxt_message, previous_fut)) => {
|
||||
let Some(message) = subxt_message else {
|
||||
tracing::trace!(target: LOG_TARGET, "Subxt channel closed");
|
||||
break;
|
||||
};
|
||||
tracing::trace!(
|
||||
target: LOG_TARGET,
|
||||
"Received register message {:?}",
|
||||
message
|
||||
);
|
||||
|
||||
self.handle_requests(message).await;
|
||||
|
||||
from_subxt_event_fut = from_subxt_event.next();
|
||||
from_node_event_fut = previous_fut;
|
||||
}
|
||||
// Message received from rpc handler: lightclient response.
|
||||
Either::Right((node_message, previous_fut)) => {
|
||||
let Some((node_message, chain)) = node_message else {
|
||||
tracing::trace!(target: LOG_TARGET, "Smoldot closed all RPC channels");
|
||||
break;
|
||||
};
|
||||
// Smoldot returns `None` if the chain has been removed (which subxt does not remove).
|
||||
let Some(response) = node_message else {
|
||||
tracing::trace!(target: LOG_TARGET, "Smoldot RPC responses channel closed");
|
||||
break;
|
||||
};
|
||||
tracing::trace!(
|
||||
target: LOG_TARGET,
|
||||
"Received smoldot RPC chain {:?} result {:?}",
|
||||
chain, response
|
||||
);
|
||||
|
||||
self.handle_rpc_response(chain, response);
|
||||
|
||||
// Advance backend, save frontend.
|
||||
from_subxt_event_fut = previous_fut;
|
||||
from_node_event_fut = stream_combinator.next();
|
||||
}
|
||||
}
|
||||
// Submit it.
|
||||
if let Err(err) = self.client.json_rpc_request(request, chain_id) {
|
||||
tracing::warn!(
|
||||
target: LOG_TARGET,
|
||||
"Failed to unsubscribe id={subscription_id} chain={chain_id:?} method={:?} err={err:?}", active_subscription.unsubscribe_method
|
||||
);
|
||||
} else {
|
||||
tracing::debug!(target: LOG_TARGET,"Unsubscribe id={subscription_id} chain={chain_id:?} method={:?}", active_subscription.unsubscribe_method);
|
||||
}
|
||||
|
||||
tracing::trace!(target: LOG_TARGET, "Task closed");
|
||||
}
|
||||
}
|
||||
|
||||
/// The RPC response from the light-client.
|
||||
/// This can either be a response of a method, or a notification from a subscription.
|
||||
#[derive(Debug, Clone)]
|
||||
enum RpcResponse {
|
||||
Method {
|
||||
/// Response ID.
|
||||
id: String,
|
||||
/// The result of the method call.
|
||||
result: Box<RawValue>,
|
||||
},
|
||||
Subscription {
|
||||
/// RPC method that generated the notification.
|
||||
method: String,
|
||||
/// Subscription ID.
|
||||
id: String,
|
||||
/// Result.
|
||||
result: Box<RawValue>,
|
||||
},
|
||||
Error {
|
||||
/// Response ID.
|
||||
id: String,
|
||||
/// Error.
|
||||
error: Box<RawValue>,
|
||||
},
|
||||
}
|
||||
|
||||
impl std::str::FromStr for RpcResponse {
|
||||
type Err = serde_json::Error;
|
||||
|
||||
fn from_str(response: &str) -> Result<Self, Self::Err> {
|
||||
// Helper structures to deserialize from raw RPC strings.
|
||||
#[derive(Deserialize, Debug)]
|
||||
struct Response {
|
||||
/// JSON-RPC version.
|
||||
#[allow(unused)]
|
||||
jsonrpc: String,
|
||||
/// Result.
|
||||
result: Box<RawValue>,
|
||||
/// Request ID
|
||||
id: String,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct NotificationParams {
|
||||
/// The ID of the subscription.
|
||||
subscription: String,
|
||||
/// Result.
|
||||
result: Box<RawValue>,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct ResponseNotification {
|
||||
/// JSON-RPC version.
|
||||
#[allow(unused)]
|
||||
jsonrpc: String,
|
||||
/// RPC method that generated the notification.
|
||||
method: String,
|
||||
/// Result.
|
||||
params: NotificationParams,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct ErrorResponse {
|
||||
/// JSON-RPC version.
|
||||
#[allow(unused)]
|
||||
jsonrpc: String,
|
||||
/// Request ID.
|
||||
id: String,
|
||||
/// Error.
|
||||
error: Box<RawValue>,
|
||||
}
|
||||
|
||||
// Check if the response can be mapped as an RPC method response.
|
||||
let result: Result<Response, _> = serde_json::from_str(response);
|
||||
if let Ok(response) = result {
|
||||
return Ok(RpcResponse::Method {
|
||||
id: response.id,
|
||||
result: response.result,
|
||||
});
|
||||
}
|
||||
|
||||
let result: Result<ResponseNotification, _> = serde_json::from_str(response);
|
||||
if let Ok(notification) = result {
|
||||
return Ok(RpcResponse::Subscription {
|
||||
id: notification.params.subscription,
|
||||
method: notification.method,
|
||||
result: notification.params.result,
|
||||
});
|
||||
}
|
||||
|
||||
let error: ErrorResponse = serde_json::from_str(response)?;
|
||||
Ok(RpcResponse::Error {
|
||||
id: error.id,
|
||||
error: error.error,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user