wip

Cargo.toml

@@ -17,7 +17,7 @@ log = { version = "0.4.11" }
 simple_logger = { version = "1.11.0", default-features = false }
 tokio-postgres = { version = "0.7.0", features = ["with-chrono-0_4", "runtime"] }
 ipinfo = { version = "0.2.0" }
-ckb-testkit = { git = "https://github.com/keroro520/ckb-integration-test", rev = "485fde9d108f49bcdb86806eae8fb3c9df0388c0", features = ["with_subscribe"] }
+ckb-testkit = { git = "https://github.com/keroro520/ckb-integration-test", rev = "f915a032fe356d173c170c4b188d1675786dd776", features = ["with_subscribe"] }
 clap = { version = "2" }
 url = "1.7.2"
 futures = "0.3"

docker/Dockerfile

@@ -3,7 +3,7 @@ FROM ubuntu:20.04
 RUN apt update \
     && apt install -y wget
 
-RUN wget https://github.com/cryptape/ckb-analyzer/releases/download/rc-v0.1.3/ckb-analyzer-linux-x86_64.tar.gz \
+RUN wget https://github.com/cryptape/ckb-analyzer/releases/download/rc-v0.1.3-2/ckb-analyzer-linux-x86_64.tar.gz \
     && tar xzf ckb-analyzer-linux-x86_64.tar.gz \
     && mv target/release/ckb-analyzer /bin/ckb-analyzer \
     && chmod +x /bin/ckb-analyzer \

src/main.rs

@@ -7,7 +7,6 @@ use crate::util::crossbeam_channel_to_tokio_channel;
 use ckb_testkit::{connector::SharedState, ConnectorBuilder, Node};
 use clap::{crate_version, values_t_or_exit, App, Arg};
 use std::env;
-use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
 use std::time::{Duration, Instant};
@@ -39,22 +38,14 @@ async fn main() {
             .map_err(|err| panic!("Invalid CKB RPC url, url: \"{}\", error: {:?}", raw, err));
         raw
     };
-    let subscription_addr = {
-        let raw = match matches.value_of("ckb-subscription-addr") {
-            Some(raw) => raw.to_string(),
-            None => match env::var_os("CKB_SUBSCRIPTION_ADDR") {
-                Some(raw) => raw.to_string_lossy().to_string(),
-                None => {
-                    panic!("Miss CKB subscription addr via neither --ckb-subscription-addr nor environment variable \"CKB_SUBSCRIPTION_ADDR\"");
-                }
-            },
-        };
-        raw.parse::<SocketAddr>().unwrap_or_else(|err| {
-            panic!(
-                "Invalid CKB subscription addr, addr: \"{}\", error: {:?}",
-                raw, err
-            )
-        })
+    let subscription_addr = match matches.value_of("ckb-subscription-addr") {
+        Some(raw) => raw.to_string(),
+        None => match env::var_os("CKB_SUBSCRIPTION_ADDR") {
+            Some(raw) => raw.to_string_lossy().to_string(),
+            None => {
+                panic!("Miss CKB subscription addr via neither --ckb-subscription-addr nor environment variable \"CKB_SUBSCRIPTION_ADDR\"");
+            }
+        },
     };
     let topics = values_t_or_exit!(matches, "topics", String);
     log::info!("CKB CKB RPC: \"{}\"", rpc_url);
@@ -205,18 +196,21 @@ async fn main() {
             }
             "SubscribeNewTransaction" => {
                 let mut handler = SubscribeNewTransaction::new(node.clone(), query_sender.clone());
+                let subscription_addr = subscription_addr.clone();
                 tokio::spawn(async move {
                     handler.run(subscription_addr).await;
                 });
             }
             "SubscribeProposedTransaction" => {
+                let subscription_addr = subscription_addr.clone();
                 let mut handler =
                     SubscribeProposedTransaction::new(node.clone(), query_sender.clone());
                 tokio::spawn(async move {
                     handler.run(subscription_addr).await;
                 });
             }
             "SubscribeRejectedTransaction" => {
+                let subscription_addr = subscription_addr.clone();
                 let mut handler =
                     SubscribeRejectedTransaction::new(node.clone(), query_sender.clone());
                 tokio::spawn(async move {

src/topic/subscribe_new_transaction.rs

@@ -2,7 +2,7 @@ use crate::ckb_types::{packed, prelude::Pack};
 use crate::entry;
 use ckb_testkit::Node;
 use futures::stream::StreamExt;
-use std::net::SocketAddr;
+use tokio::net::ToSocketAddrs;
 
 pub struct SubscribeNewTransaction {
     node: Node,
@@ -14,7 +14,7 @@ impl SubscribeNewTransaction {
         Self { node, query_sender }
     }
 
-    pub async fn run(&mut self, subscription_addr: SocketAddr) {
+    pub async fn run<A: ToSocketAddrs>(&mut self, subscription_addr: A) {
         self.node.subscribe_new_transaction(subscription_addr).await;
         while let Some(Ok((_topic, pool_tx_entry))) =
             self.node.new_transaction_subscriber().next().await

src/topic/subscribe_proposed_transaction.rs

@@ -2,7 +2,7 @@ use crate::ckb_types::{packed, prelude::Pack};
 use crate::entry;
 use ckb_testkit::Node;
 use futures::stream::StreamExt;
-use std::net::SocketAddr;
+use tokio::net::ToSocketAddrs;
 
 pub struct SubscribeProposedTransaction {
     node: Node,
@@ -14,7 +14,7 @@ impl SubscribeProposedTransaction {
         Self { node, query_sender }
     }
 
-    pub async fn run(&mut self, subscription_addr: SocketAddr) {
+    pub async fn run<A: ToSocketAddrs>(&mut self, subscription_addr: A) {
         self.node
             .subscribe_proposed_transaction(subscription_addr)
             .await;

src/topic/subscribe_rejected_transaction.rs

@@ -3,7 +3,7 @@ use crate::ckb_types::{packed, prelude::Pack};
 use crate::entry;
 use ckb_testkit::Node;
 use futures::stream::StreamExt;
-use std::net::SocketAddr;
+use tokio::net::ToSocketAddrs;
 
 pub struct SubscribeRejectedTransaction {
     node: Node,
@@ -15,7 +15,7 @@ impl SubscribeRejectedTransaction {
         Self { node, query_sender }
     }
 
-    pub async fn run(&mut self, subscription_addr: SocketAddr) {
+    pub async fn run<A: ToSocketAddrs>(&mut self, subscription_addr: A) {
         self.node
             .subscribe_rejected_transaction(subscription_addr)
             .await;