From 2243cacf70046aff615242b0a6a0f569cca3289d Mon Sep 17 00:00:00 2001 From: Bryan Stitt Date: Sun, 5 Jun 2022 22:39:44 +0000 Subject: [PATCH] new heads subscription works --- web3-proxy/src/app.rs | 13 ++++++++----- web3-proxy/src/config.rs | 2 +- web3-proxy/src/frontend/errors.rs | 6 ++---- web3-proxy/src/frontend/ws_proxy.rs | 2 +- web3-proxy/src/jsonrpc.rs | 4 +++- 5 files changed, 15 insertions(+), 12 deletions(-) diff --git a/web3-proxy/src/app.rs b/web3-proxy/src/app.rs index 1bd1eb72..c7d898d3 100644 --- a/web3-proxy/src/app.rs +++ b/web3-proxy/src/app.rs @@ -158,13 +158,12 @@ impl Web3ProxyApp { let f = { let subscription_id = subscription_id.clone(); - let id = payload.id; let params = payload.params.as_deref().unwrap().get(); if params == r#"["newHeads"]"# { let head_block_receiver = self.head_block_receiver.clone(); - info!("received new heads subscription"); + trace!(?subscription_id, "new heads subscription"); async move { let mut head_block_receiver = Abortable::new( WatchStream::new(head_block_receiver), @@ -172,9 +171,8 @@ impl Web3ProxyApp { ); while let Some(new_head) = head_block_receiver.next().await { - // TODO: make a struct for this? using JsonRpcForwardedResponse + // TODO: make a struct for this? using JsonRpcForwardedResponse won't work because it needs an id let msg = json!({ - "id": id, "jsonrpc": "2.0", "method":"eth_subscription", "params": { @@ -185,8 +183,13 @@ impl Web3ProxyApp { let msg = Message::Text(serde_json::to_string(&msg).unwrap()); - subscription_tx.send_async(msg).await.unwrap(); + if subscription_tx.send_async(msg).await.is_err() { + // TODO: cancel this subscription earlier? select on head_block_receiver.next() and an abort handle? + break; + }; } + + trace!(?subscription_id, "closed new heads subscription"); } } else { return Err(anyhow::anyhow!("unimplemented")); diff --git a/web3-proxy/src/config.rs b/web3-proxy/src/config.rs index a2415d30..e08be760 100644 --- a/web3-proxy/src/config.rs +++ b/web3-proxy/src/config.rs @@ -13,7 +13,7 @@ pub struct CliConfig { #[argh(option, default = "8544")] pub port: u16, - /// number of worker threads + /// number of worker threads. Defaults to the number of logical processors #[argh(option, default = "0")] pub workers: usize, diff --git a/web3-proxy/src/frontend/errors.rs b/web3-proxy/src/frontend/errors.rs index 2fe25d6d..6ac9353c 100644 --- a/web3-proxy/src/frontend/errors.rs +++ b/web3-proxy/src/frontend/errors.rs @@ -12,12 +12,12 @@ pub async fn handler_404() -> impl IntoResponse { /// handle errors by converting them into something that implements `IntoResponse` /// TODO: use this. i can't get https://docs.rs/axum/latest/axum/error_handling/index.html to work +/// TODO: i think we want a custom result type instead. put the anyhow result inside. then `impl IntoResponse for CustomResult` pub async fn handle_anyhow_error( err: anyhow::Error, code: Option, ) -> impl IntoResponse { - // TODO: what id can we use? how do we make sure the incoming id gets attached to this? - let id = RawValue::from_string("0".to_string()).unwrap(); + let id = RawValue::from_string("null".to_string()).unwrap(); let err = JsonRpcForwardedResponse::from_anyhow_error(err, id); @@ -27,5 +27,3 @@ pub async fn handle_anyhow_error( (code, Json(err)) } - -// i think we want a custom result type. it has an anyhow result inside. if it impl IntoResponse I think we'll get this for free diff --git a/web3-proxy/src/frontend/ws_proxy.rs b/web3-proxy/src/frontend/ws_proxy.rs index 1f842cdf..6d2eb356 100644 --- a/web3-proxy/src/frontend/ws_proxy.rs +++ b/web3-proxy/src/frontend/ws_proxy.rs @@ -83,7 +83,7 @@ async fn handle_socket_payload( } Err(err) => { // TODO: what should this id be? - let id = RawValue::from_string("0".to_string()).unwrap(); + let id = RawValue::from_string("null".to_string()).unwrap(); (id, Err(err.into())) } }; diff --git a/web3-proxy/src/jsonrpc.rs b/web3-proxy/src/jsonrpc.rs index f3809c1b..e0d82d49 100644 --- a/web3-proxy/src/jsonrpc.rs +++ b/web3-proxy/src/jsonrpc.rs @@ -193,7 +193,9 @@ impl JsonRpcForwardedResponse { pub fn from_string(partial_response: String, id: Box) -> Self { trace!("partial_response: {}", partial_response); - let partial_response = RawValue::from_string(partial_response).unwrap(); + // TODO: anyhow result on this + // TODO: proper escaping. this feels wrong. probably refactor to not need this at all + let partial_response = RawValue::from_string(format!(r#""{}""#, partial_response)).unwrap(); Self::from_response(partial_response, id) }