fix SubscriptionRpc

nervosnetwork · Dec 12, 2023 · 6868b40 · 6868b40
1 parent f9c0092
commit 6868b40
Show file tree

Hide file tree

Showing 4 changed files with 6 additions and 11 deletions.
diff --git a/rpc/src/module/subscription.rs b/rpc/src/module/subscription.rs
@@ -196,8 +196,9 @@ impl SubscriptionRpc for SubscriptionRpcImpl {
             Topic::ProposedTransaction => self.proposed_transaction_sender.clone(),
             Topic::RejectedTransaction => self.new_reject_transaction_sender.clone(),
         };
+        let mut rx = tx.subscribe();
         Ok(Box::pin(async_stream::stream! {
-               while let Ok(msg) = tx.subscribe().recv().await {
+               while let Ok(msg) = rx.recv().await {
                     yield msg;
                }
         }))

diff --git a/rpc/src/server.rs b/rpc/src/server.rs
@@ -101,8 +101,7 @@ impl RpcServer {
                 stream_config
                     .with_keep_alive(true)
                     .with_shutdown(async move {
-                        let exit = new_tokio_exit_rx();
-                        exit.cancelled().await;
+                        new_tokio_exit_rx().cancelled().await;
                     });
             app = app.layer(Extension(ws_config));
         }
@@ -121,8 +120,7 @@ impl RpcServer {
 
             let _ = tx_addr.send(server.local_addr());
             let graceful = server.with_graceful_shutdown(async move {
-                let exit = new_tokio_exit_rx();
-                exit.cancelled().await;
+                new_tokio_exit_rx().cancelled().await;
             });
             drop(graceful.await);
         });
@@ -144,8 +142,7 @@ impl RpcServer {
                 .with_channel_size(4)
                 .with_pipeline_size(4)
                 .with_shutdown(async move {
-                    let exit = new_tokio_exit_rx();
-                    exit.cancelled().await;
+                    new_tokio_exit_rx().cancelled().await;
                 });
 
             let exit_signal: CancellationToken = new_tokio_exit_rx();

diff --git a/rpc/src/service_builder.rs b/rpc/src/service_builder.rs
@@ -49,7 +49,7 @@ impl<'a> ServiceBuilder<'a> {
     pub fn new(config: &'a RpcConfig) -> Self {
         Self {
             config,
-            io_handler: IoHandler::with_compatibility(jsonrpc_core::Compatibility::V2),
+            io_handler: IoHandler::default(),
         }
     }
 

diff --git a/rpc/src/tests/setup.rs b/rpc/src/tests/setup.rs
@@ -35,9 +35,6 @@ const CELLBASE_MATURITY: u64 = 0;
 const ALERT_UNTIL_TIMESTAMP: u64 = 2_524_579_200;
 
 // Construct `Consensus` with an always-success cell
-//
-// It is similar to `util::test-chain-utils::always_success_consensus`, but with hard-code
-// genesis timestamp.
 pub(crate) fn always_success_consensus() -> Consensus {
     let always_success_tx = always_success_transaction();
     let dao = genesis_dao_data(vec![&always_success_tx]).unwrap();