Skip to content

Commit

Permalink
Bump hyper version to v1 (#184)
Browse files Browse the repository at this point in the history 
* major update hyper version

* make subway server builder compatible with hyper v1

* update jsonrpsee dep to v0.23.x

* remove unncessary clone

* cargo fmt

* fix format issue

* overwrite addr to the one with actual allocated port
  • Loading branch information
@shunsukew
shunsukew authored Jun 9, 2024
1 parent 7d63591 commit a7e6469
Show file tree
Hide file tree
Showing 8 changed files with 490 additions and 333 deletions.
466 changes: 287 additions & 179 deletions Cargo.lock
View file

Large diffs are not rendered by default.

10 changes: 6 additions & 4 deletions Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,10 @@ chrono = "0.4.24"
clap = { version = "4.1.1", features = ["derive"] }
enumflags2 = "0.7.7"
futures = "0.3.25"
http = "0.2"
hyper = "0.14"
http = "1"
http-body = "1"
http-body-util = "0.1"
hyper = "1.3"
moka = { version = "0.12", features = ["future"] }
opentelemetry = { version = "0.21.0" }
opentelemetry-datadog = { version = "0.9.0", features = ["reqwest-client"] }
Expand All @@ -39,13 +41,13 @@ serde_yaml = "0.9.17"
substrate-prometheus-endpoint = "0.17.0"
tokio = { version = "1.24.2", features = ["full"] }
tower = { version = "0.4.13", features = ["full"] }
tower-http = { version = "0.4", features = ["full"] }
tower-http = { version = "0.5.2", features = ["full"] }
tracing = "0.1.40"
tracing-serde = "0.1.3"
tracing-subscriber = { version = "0.3.18", features = ["env-filter", "json"] }
garde = { version = "0.18", features = ["full"] }

jsonrpsee = { version = "0.22.1",features = ["full"] }
jsonrpsee = { version = "0.23", features = ["full"] }
governor = { path = "./vendor/governor/governor" }

[dev-dependencies]
Expand Down
22 changes: 11 additions & 11 deletions benches/bench/helpers.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ pub async fn ws_server(handle: tokio::runtime::Handle, url: &str) -> (String, js
SUB_METHOD_NAME,
SUB_METHOD_NAME,
UNSUB_METHOD_NAME,
|_params, pending, _ctx| async move {
|_params, pending, _ctx, _| async move {
let sink = pending.accept().await?;
let msg = SubscriptionMessage::from_json(&"Hello")?;
sink.send(msg).await?;
Expand All @@ -53,7 +53,7 @@ pub async fn ws_server(handle: tokio::runtime::Handle, url: &str) -> (String, js
"chain_subscribeNewHeads",
"chain_newHead",
"chain_unsubscribeNewHeads",
|_params, pending, _ctx| async move {
|_params, pending, _ctx, _| async move {
let sink = pending.accept().await?;
let msg = SubscriptionMessage::from_json(&serde_json::json!({ "number": "0x4321" }))?;
sink.send(msg).await?;
Expand All @@ -66,7 +66,7 @@ pub async fn ws_server(handle: tokio::runtime::Handle, url: &str) -> (String, js
"chain_subscribeFinalizedHeads",
"chain_finalizedHead",
"chain_unsubscribeFinalizedHeads",
|_params, pending, _ctx| async move {
|_params, pending, _ctx, _| async move {
let sink = pending.accept().await?;
let msg = SubscriptionMessage::from_json(&serde_json::json!({ "number": "0x4321" }))?;
sink.send(msg).await?;
Expand All @@ -84,47 +84,47 @@ fn gen_rpc_module() -> jsonrpsee::RpcModule<()> {
let mut module = jsonrpsee::RpcModule::new(());

module
.register_method(SYNC_FAST_CALL, |_, _| Ok::<_, ErrorObjectOwned>("lo"))
.register_method(SYNC_FAST_CALL, |_, _, _| Ok::<_, ErrorObjectOwned>("lo"))
.unwrap();
module
.register_async_method(ASYNC_FAST_CALL, |_, _| async {
.register_async_method(ASYNC_FAST_CALL, |_, _, _| async {
Result::<_, ErrorObjectOwned>::Ok("lo")
})
.unwrap();

module
.register_method(SYNC_MEM_CALL, |_, _| Ok::<_, ErrorObjectOwned>("A".repeat(MIB)))
.register_method(SYNC_MEM_CALL, |_, _, _| Ok::<_, ErrorObjectOwned>("A".repeat(MIB)))
.unwrap();

module
.register_async_method(ASYNC_MEM_CALL, |_, _| async move {
.register_async_method(ASYNC_MEM_CALL, |_, _, _| async move {
Result::<_, ErrorObjectOwned>::Ok("A".repeat(MIB))
})
.unwrap();

module
.register_method(SYNC_SLOW_CALL, |_, _| {
.register_method(SYNC_SLOW_CALL, |_, _, _| {
std::thread::sleep(SLOW_CALL);
Ok::<_, ErrorObjectOwned>("slow call")
})
.unwrap();

module
.register_async_method(ASYNC_SLOW_CALL, |_, _| async move {
.register_async_method(ASYNC_SLOW_CALL, |_, _, _| async move {
tokio::time::sleep(SLOW_CALL).await;
Result::<_, ErrorObjectOwned>::Ok("slow call async")
})
.unwrap();

module
.register_async_method(ASYNC_INJECT_CALL, |_, _| async move {
.register_async_method(ASYNC_INJECT_CALL, |_, _, _| async move {
tokio::time::sleep(SLOW_CALL).await;
Result::<_, ErrorObjectOwned>::Ok("inject call async")
})
.unwrap();

module
.register_async_method("chain_getBlockHash", |_, _| async move {
.register_async_method("chain_getBlockHash", |_, _, _| async move {
tokio::time::sleep(SLOW_CALL).await;
Result::<_, ErrorObjectOwned>::Ok("0x42")
})
Expand Down
6 changes: 3 additions & 3 deletions src/extensions/client/mock.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ impl TestServerBuilder {
pub fn register_method(&mut self, name: &'static str) -> mpsc::Receiver<MockRequest> {
let (tx, rx) = mpsc::channel::<MockRequest>(100);
self.module
.register_async_method(name, move |params, _| {
.register_async_method(name, move |params, _, _| {
let tx = tx.clone();
let params = params.parse::<JsonValue>().unwrap();
async move {
Expand All @@ -56,7 +56,7 @@ impl TestServerBuilder {
) -> mpsc::Receiver<MockSubscription> {
let (tx, rx) = mpsc::channel::<MockSubscription>(100);
self.module
.register_subscription(sub_name, method_name, unsub_name, move |params, sink, _| {
.register_subscription(sub_name, method_name, unsub_name, move |params, sink, _, _| {
let tx = tx.clone();
let params = params.parse::<JsonValue>().unwrap();
tokio::spawn(async move {
Expand All @@ -76,7 +76,7 @@ impl TestServerBuilder {
unsub_name: &'static str,
) {
self.module
.register_subscription(sub_name, method_name, unsub_name, move |_, sink, _| async {
.register_subscription(sub_name, method_name, unsub_name, move |_, sink, _, _| async {
sink.reject(errors::map_error(Error::Call(ErrorObject::owned(
1010,
"Invalid Transaction",
Expand Down
10 changes: 2 additions & 8 deletions src/extensions/client/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -250,10 +250,7 @@ impl Client {
Err(err) => {
tracing::debug!("Request failed: {:?}", err);
match err {
Error::RequestTimeout
| Error::Transport(_)
| Error::RestartNeeded(_)
| Error::MaxSlotsExceeded => {
Error::RequestTimeout | Error::Transport(_) | Error::RestartNeeded(_) => {
tokio::time::sleep(get_backoff_time(&request_backoff_counter)).await;

// make sure it's still connected
Expand Down Expand Up @@ -329,10 +326,7 @@ impl Client {
Err(err) => {
tracing::debug!("Subscribe failed: {:?}", err);
match err {
Error::RequestTimeout
| Error::Transport(_)
| Error::RestartNeeded(_)
| Error::MaxSlotsExceeded => {
Error::RequestTimeout | Error::Transport(_) | Error::RestartNeeded(_) => {
tokio::time::sleep(get_backoff_time(&request_backoff_counter)).await;

// make sure it's still connected
Expand Down
Loading

0 comments on commit a7e6469

Please sign in to comment.