优化

exchanges/src/binance_swap_rest.rs

@@ -287,7 +287,7 @@ impl BinanceSwapRest {
         let mut params_json: serde_json::Value = serde_json::from_str(params.clone().as_str()).unwrap();
         // let mut body = "".to_string();
         let timestamp = chrono::Utc::now().timestamp_millis().to_string();
-        params_json.as_object_mut().unwrap().insert("timestamp".to_string(), timestamp.parse().unwrap());
+        params_json.as_object_mut().unwrap().insert("timestamp".to_string(), serde_json::Value::from(timestamp));
 
 
         let mut headers = HeaderMap::new();
@@ -391,11 +391,14 @@ impl BinanceSwapRest {
     //res_data 解析
     pub fn res_data_analysis(result: Result<ResponseData, reqwest::Error>, params: String) -> ResponseData {
         match result {
-            Ok(res_data) => {
+            Ok(res_data) => { 
                 if res_data.code != "200" {
                     let message = res_data.message;
-                    let mut  error = ResponseData::error(res_data.label, "".to_string());
-                    error.message = format!("{},相关参数:{}", message, params);
+                    let json_value: serde_json::Value = serde_json::from_str(&message).unwrap();
+                    let msg = json_value["msg"].as_str().unwrap();
+                    let mut error = ResponseData::error(res_data.label, msg.parse().unwrap());
+                    error.code = format!("{}", json_value["code"]);
+                    error.data = format!("参数：{}", params);
                     error
                 } else {
                     res_data

exchanges/src/binance_swap_ws.rs

@@ -3,7 +3,7 @@ use std::{io};
 use std::io::{Write};
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use tokio::sync::mpsc::Sender;
-use tracing::{ trace};
+use tracing::{info, trace};
 use crate::{proxy};
 use tungstenite::client::{AutoStream, connect_with_proxy, ProxyAutoStream};
 use tungstenite::{connect, Message, WebSocket};
@@ -221,6 +221,7 @@ impl BinanceSwapWs {
     //代理
     async fn proxy_subscription(&self, mut web_socket: WebSocket<ProxyAutoStream>, _subscription: String)
     {
+        info!("走代理-链接成功！开始数据读取");
         let lable = self.label.clone();
         /*****消息溜***/
         let mut stdout = io::stdout();
@@ -228,6 +229,7 @@ impl BinanceSwapWs {
             let msg = web_socket.read_message();
             match msg {
                 Ok(Message::Text(text)) => {
+                    info!("获取推送：{}",text.clone());
                     let res_data = Self::ok_text(lable.to_string(), text);
                     if res_data.code == "-200" {
                         writeln!(stdout, "订阅成功：{:?}", res_data.data).unwrap();
@@ -267,6 +269,7 @@ impl BinanceSwapWs {
     async fn subscription(&self, mut web_socket: WebSocket<AutoStream>,
                           subscription: String)
     {
+        info!("链接成功！开始数据读取");
         let lable = self.label.clone();
         /*****订阅***/
         web_socket.write_message(Message::Text(subscription))
@@ -277,6 +280,7 @@ impl BinanceSwapWs {
             let msg = web_socket.read_message();
             match msg {
                 Ok(Message::Text(text)) => {
+                    info!("获取推送：{}",text.clone());
                     let res_data = Self::ok_text(lable.to_string(), text);
                     if res_data.code == "-200" {
                         writeln!(stdout, "订阅成功：{:?}", res_data.data).unwrap();

exchanges/src/binance_usdt_swap_rest.rs

@@ -1,119 +0,0 @@
-use std::collections::BTreeMap;
-use reqwest::header::HeaderMap;
-use ring::hmac;
-use hex;
-use tracing::trace;
-use crate::http_tool::RestTool;
-use crate::response_base::ResponseData;
-
-#[derive(Clone)]
-pub struct BinanceUsdtSwapRest {
-    pub label: String,
-    login_param: BTreeMap<String, String>,
-    rest: RestTool,
-}
-
-impl BinanceUsdtSwapRest {
-    /*******************************************************************************************************/
-    /*****************************************获取一个对象****************************************************/
-    /*******************************************************************************************************/
-    pub fn new(is_colo: bool, login_param: BTreeMap<String, String>) -> BinanceUsdtSwapRest
-    {
-        return BinanceUsdtSwapRest::new_label("default-BinanceSpotWs".to_string(), is_colo, login_param);
-    }
-    pub fn new_label(label: String, is_colo: bool, login_param: BTreeMap<String, String>) -> BinanceUsdtSwapRest {
-        let base_url = if is_colo {
-            trace!("不支持colo高速线路");
-            "https://fapi.binance.com".to_string()
-        } else {
-            "https://fapi.binance.com".to_string()
-        };
-
-        /*****返回结构体*******/
-        BinanceUsdtSwapRest {
-            label,
-            login_param,
-            rest: RestTool::new(base_url.to_string()),
-        }
-    }
-
-    /*******************************************************************************************************/
-    /*****************************************rest请求函数********************************************************/
-    /*******************************************************************************************************/
-    //账户信息
-    pub async fn get_account(&self) -> ResponseData {
-        let btree_map = serde_json::json!({
-            "recvWindow":"2000"
-         });
-
-        //组装登陆认证信息
-        let params = self.get_sign(btree_map.to_string());
-        let get_response = self.rest.http_toll(
-            "/fapi/v2/balance".to_string(),
-            "get".to_string(),
-            params,
-            self.get_headers(true),
-        ).await;
-        let mut res_data = RestTool::res_data_analysis(get_response);
-        res_data.channel = "account".to_string();
-        res_data
-    }
-    //持仓信息
-    pub fn get_position(&self) {}
-    /*******************************************************************************************************/
-    /*****************************************工具函数********************************************************/
-    /*******************************************************************************************************/
-    //组装heander
-    fn get_headers(&self, is_login: bool) -> HeaderMap {
-        let mut headers = HeaderMap::new();
-        headers.insert("Content-Type", "application/json; charset=UTF-8".parse().unwrap());
-        if is_login {
-            let access_key = self.login_param.get("access_key");
-            match access_key {
-                None => {}
-                Some(v) => {
-                    headers.insert("X-MBX-APIKEY", v.clone().parse().unwrap());
-                }
-            }
-        }
-        // trace!("----组装headers:{:?}", headers);
-        headers
-    }
-
-    //生成sign,重新组装param
-    fn get_sign(&self, params_str: String) -> String {
-        let mut params: serde_json::Value = serde_json::from_str(params_str.clone().as_str()).unwrap();
-        if let Some(map) = params.as_object_mut() {
-            map.insert("timestamp".to_owned(), chrono::Utc::now().timestamp_millis().to_string().parse().unwrap());
-        }
-        // params.insert("timestamp".to_string(), chrono::Utc::now().timestamp_millis().to_string());
-
-        // trace!("login_param参数：{:?}", self.login_param);
-        let mut secret_key = "";
-        if self.login_param.contains_key("secret_key") {
-            secret_key = self.login_param.get("secret_key").unwrap()
-        }
-
-        /*签名生成*/
-        let params_str = RestTool::parse_params_to_str(params.to_string());
-        // trace!("---参数:{}", params_str);
-        let message = format!("{}", params_str);
-
-        // let secret_key2 = "2b5eb11e18796d12d88f13dc27dbbd02c2cc51ff7059765ed9821957d82bb4d9";
-        // let message2 = "symbol=BTCUSDT&side=BUY&type=LIMIT&quantity=1&price=9000&timeInForce=GTC&recvWindow=5000&timestamp=1591702613943";
-        let signed_key = hmac::Key::new(hmac::HMAC_SHA256, secret_key.as_ref());
-        let signature = hex::encode(hmac::sign(&signed_key, message.as_bytes()).as_ref());
-        // trace!("---signature:{}", signature);
-
-        // $ echo -n "symbol=BTCUSDT&side=BUY&type=LIMIT&quantity=1&price=9000&timeInForce=GTC&recvWindow=5000&timestamp=1591702613943" | openssl dgst -sha256 -hmac "2b5eb11e18796d12d88f13dc27dbbd02c2cc51ff7059765ed9821957d82bb4d9"
-        //     (stdin)= 3c661234138461fcc7a7d8746c6558c9842d4e10870d2ecbedf7777cad694af9
-
-        // params.insert("signature".to_string(), signature.clone());
-        if let Some(map) = params.as_object_mut() {
-            map.insert("signature".to_owned(), serde_json::Value::from(signature.clone()));
-        }
-
-        // trace!("----组装params:{:?}", params);
-        params.to_string()
-    }
-}

exchanges/src/binance_usdt_swap_ws.rs

@@ -1,343 +0,0 @@
-use std::collections::BTreeMap;
-use std::io;
-use std::io::{Write};
-use std::net::{IpAddr, Ipv4Addr, SocketAddr};
-use serde_json::{json, Value};
-use tokio::sync::mpsc::Sender;
-use tracing::trace;
-use crate::proxy;
-use crate::response_base::ResponseData;
-use tungstenite::client::{AutoStream, connect_with_proxy, ProxyAutoStream};
-use tungstenite::{connect, Message, WebSocket};
-use tungstenite::protocol::WebSocketConfig;
-use url::Url;
-
-#[derive(Clone)]
-pub struct BinanceUsdtSwapWs {
-    pub label: String,
-    //连接地址
-    request_url: String,
-    //ip
-    ip: String,
-    //端口
-    port: u16,
-    //登陆所需参数
-    // login_param: BTreeMap<String, String>,
-
-    sender: Sender<ResponseData>,
-}
-
-impl BinanceUsdtSwapWs {
-    /*******************************************************************************************************/
-    /*****************************************获取一个对象****************************************************/
-    /*******************************************************************************************************/
-    pub fn new(is_colo: bool,
-               _login_param: BTreeMap<String, String>,
-               sender: Sender<ResponseData>) -> BinanceUsdtSwapWs
-    {
-        return BinanceUsdtSwapWs::new_label("default-BinanceUsdtSwapWs".to_string(), is_colo, _login_param, sender);
-    }
-    pub fn new_label(label: String, is_colo: bool,
-                     _login_param: BTreeMap<String, String>,
-                     sender: Sender<ResponseData>) -> BinanceUsdtSwapWs
-    {
-        let base_url = if is_colo {
-            trace!("不支持colo高速线路");
-            "wss://stream.binance.com:443/ws".to_string()
-        } else {
-            "wss://stream.binance.com:443/ws".to_string()
-        };
-        let mut ip_v = "";
-        let mut port_v = 8080;
-
-        /*******走代理：根据环境变量配置来决定,如果配置了走代理，没有配置不走*******/
-        let parsing_detail = proxy::ParsingDetail::parsing_environment_variables();
-        if parsing_detail.ip_address.len() > 0 && parsing_detail.port.len() > 0 {
-            ip_v = parsing_detail.ip_address.as_str();
-            port_v = parsing_detail.port.parse().unwrap();
-        }
-
-        /*****返回结构体*******/
-        BinanceUsdtSwapWs {
-            label,
-            request_url: base_url,
-            ip: ip_v.clone().to_string(),
-            port: port_v,
-            // login_param,
-            sender,
-        }
-    }
-
-
-    /*******************************************************************************************************/
-    /*****************************************订阅函数********************************************************/
-    /*******************************************************************************************************/
-    //K线-不需要认证
-    pub async fn kline(&self, b_array: Vec<&str>)
-    {
-        //订阅信息拼接
-        let mut params = vec![];
-        for item in &b_array {
-            let mut b_name = item.to_lowercase();
-            b_name = format!("{}@{}", b_name, "kline_1s");
-            params.push(b_name);
-        }
-        self.run(params).await;
-    }
-
-    //自定义-不需要认证
-    pub async fn custom_subscribe(&self, b_array: Vec<String>, sub_trade: u8, sub_fast: u8)
-    {
-        let mut params = vec![];
-
-        for item in &b_array {
-            let b_name = item.to_lowercase();
-            if sub_trade > 0 {
-                params.push(format!("{}@aggTrade", b_name));
-            }
-            if sub_fast > 0 {
-                params.push(format!("{}@bookTicker", b_name));
-            } else {
-                params.push(format!("{}@depth20@100ms", b_name));
-            }
-        }
-        self.run(params).await;
-    }
-
-    /*******************************************************************************************************/
-    /*****************************************工具函数********************************************************/
-    /*******************************************************************************************************/
-
-    /*******************************************************************************************************/
-    /*****************************************socket基本*****************************************************/
-    /*******************************************************************************************************/
-    async fn run(&self, params: Vec<String>)
-    {
-        //订阅信息组装
-        let subscription = json!({
-        "method": "SUBSCRIBE",
-        "params":params,
-                "id": 1
-        });
-        // let parse_fn_arc = Arc::new(Mutex::new(parse_fn)); // Wrap the closure in an Arc<Mutex<_>>
-        loop {
-            trace!("要连接咯~~！！");
-            //币安-登陆流程-rest请求获取k然后拿到 key 拼接地址
-            // if self.is_login { //暂时没看到有订阅的频道需要登陆 所以暂时不做
-            // }
-
-            let request_url = Url::parse(self.request_url.as_str()).unwrap();
-            //1. 判断是否需要代理,根据代理地址是否存来选择
-            if self.ip.len() > 0 {
-                let ip_array: Vec<&str> = self.ip.split(".").collect();
-                let proxy_address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(
-                    ip_array[0].parse().unwrap(),
-                    ip_array[1].parse().unwrap(),
-                    ip_array[2].parse().unwrap(),
-                    ip_array[3].parse().unwrap())
-                ), self.port);
-                let websocket_config = Some(WebSocketConfig {
-                    max_send_queue: Some(16),
-                    max_message_size: Some(16 * 1024 * 1024),
-                    max_frame_size: Some(16 * 1024 * 1024),
-                    accept_unmasked_frames: false,
-                });
-                let max_redirects = 5;
-                match connect_with_proxy(request_url.clone(), proxy_address, websocket_config, max_redirects) {
-                    Ok(ws) => {
-                        self.proxy_subscription(ws.0, &subscription).await;
-                    }
-                    Err(err) => {
-                        trace!("Can't connect(无法连接): {}", err);
-                    }
-                };
-            } else {
-                match connect(request_url.clone()) {
-                    Ok(ws) => {
-                        self.subscription(ws.0, &subscription).await;
-                    }
-                    Err(err) => {
-                        // 连接失败时执行的操作
-                        trace!("Can't connect(无法连接): {}", err);
-                        // 返回一个默认的 WebSocket 对象或其他适当的值
-                        // 或者根据需要触发 panic 或返回错误信息
-                    }
-                };
-            }
-            trace!("退出来咯")
-        }
-    }
-
-    //代理
-    async fn proxy_subscription(&self, mut web_socket: WebSocket<ProxyAutoStream>,
-                                subscription: &serde_json::Value,
-    )
-    {
-        let label = self.label.clone();
-        /*****消息溜***/
-        let mut stdout = io::stdout();
-        // let mut stderr = io::stderr();
-        // /*****是否需要登陆****/
-        // if self.is_login {
-        //     trace!("----需要登陆");
-        // } else {
-        //     trace!("----no longin(不需要登陆)");
-        // }
-        /******订阅信息********/
-        let sub_json = subscription.clone();
-        trace!("--订阅内容：{:?}", sub_json.to_string());
-        let sub_json_str = sub_json.to_string();
-        web_socket.write_message(Message::Text(sub_json_str))
-            .unwrap();
-        loop {
-            let msg = web_socket.read_message();
-            match msg {
-                Ok(Message::Text(text)) => {
-                    let res_data = Self::ok_text(label.to_string(), text);
-
-                    let sender = self.sender.clone();
-                    tokio::spawn(async move {
-                        sender.send(res_data).await.unwrap();
-                    });
-                    tokio::spawn(async move {});
-                }
-                Ok(Message::Ping(s)) => {
-                    writeln!(stdout, "Ping-响应--{:?}", String::from_utf8(s.clone())).unwrap();
-                    let _ = web_socket.write_message(Message::Pong(Vec::from("pong")));
-                    writeln!(stdout, "回应-pong---{:?}", String::from_utf8(s.clone())).unwrap();
-                }
-                Ok(Message::Pong(s)) => {
-                    // trace!("Pong-响应--{:?}", String::from_utf8(s));
-                    writeln!(stdout, "Pong-响应--{:?}", String::from_utf8(s.clone())).unwrap();
-                }
-                Ok(Message::Close(_)) => {
-                    // trace!("socket 关闭: ");
-                    writeln!(stdout, "Close-响应").unwrap();
-                }
-                Err(error) => {
-                    // trace!("Error receiving message: {}", error);
-                    writeln!(stdout, "Err-响应{}", error).unwrap();
-                    break;
-                }
-                _ => {}
-            }
-        }
-        web_socket.close(None).unwrap();
-    }
-    //非代理
-    async fn subscription(&self, mut web_socket: WebSocket<AutoStream>,
-                          subscription: &serde_json::Value,
-    )
-    {
-        let label = self.label.clone();
-        /*****消息溜***/
-        // let  stdout = io::stdout();
-        // let mut stderr = io::stderr();
-        /*****是否需要登陆****/
-        // if self.is_login {
-        //     trace!("----需要登陆");
-        //     // let login_json_str = self.log_in_to_str();
-        //     // web_socket.write_message(Message::Text(login_json_str)).unwrap();
-        //     // thread::sleep(Duration::from_secs(1));
-        // } else {
-        //     trace!("----no longin(不需要登陆)");
-        // }
-        /******订阅信息********/
-        let sub_json = subscription.clone();
-        trace!("--订阅内容：{:?}", sub_json);
-        let sub_json_str = sub_json.to_string();
-        web_socket.write_message(Message::Text(sub_json_str))
-            .unwrap();
-        loop {
-            let msg = web_socket.read_message();
-            match msg {
-                Ok(Message::Text(text)) => {
-                    let res_data = Self::ok_text(label.to_string(), text);
-                    self.sender.send(res_data).await.unwrap();
-                    // writeln!(stdout, "Pong-响应--{:?}", res_data).unwrap();
-                    // let parse_fn_clone = Arc::clone(parse_fn); // Clone the Arc for each iteration
-                    // tokio::spawn(async move {
-                    //     let parse_fn_lock = parse_fn_clone.lock().await;
-                    //     parse_fn_lock(res_data).await;
-                    // });
-                }
-                Ok(Message::Ping(s)) => {
-                    trace!("Ping-响应--{:?}", String::from_utf8(s));
-                }
-                Ok(Message::Pong(s)) => {
-                    trace!("Pong-响应--{:?}", String::from_utf8(s));
-                }
-                Ok(Message::Close(_)) => {
-                    trace!("socket 关闭: ");
-                }
-                Err(error) => {
-                    trace!("Error receiving message: {}", error);
-                    break;
-                }
-                _ => {}
-            }
-        }
-        web_socket.close(None).unwrap();
-    }
-
-    //数据解析
-    pub fn ok_text(label: String, text: String) -> ResponseData
-    {
-        let mut res_data = ResponseData::new(label.to_string(), "200".to_string(), "success".to_string(), "".to_string());
-        let json_value: serde_json::Value = serde_json::from_str(&text).unwrap();
-        if json_value.get("result").is_some() {
-            //订阅反馈
-        } else {
-            // trace!("????-----:{}", json_value);
-            //数据解析
-            if let Some(json_obj) = json_value.as_object() {
-                if Self::channel_agg_trade(json_obj.clone()) {
-                    res_data.channel = "aggTrade".to_string();
-                } else if Self::channel_book_ticker(json_obj.clone()) {
-                    res_data.channel = "bookTicker".to_string();
-                } else if Self::channel_depth(json_obj.clone()) {
-                    res_data.channel = "depth".to_string();
-                } else {
-                    res_data.channel = "未解析频道".to_string();
-                }
-                res_data.data = text
-            }
-        }
-        res_data
-    }
-
-    pub fn channel_agg_trade(json_obj: serde_json::Map<String, Value>) -> bool {
-        let mut is_agg_trade = false;
-        if let Some(name) = json_obj.get("e") {
-            if let Some(name_str) = name.as_str() {
-                if name_str == "aggTrade" {
-                    is_agg_trade = true;
-                }
-            }
-        }
-        is_agg_trade
-    }
-    pub fn channel_book_ticker(json_obj: serde_json::Map<String, Value>) -> bool {
-        let mut is_book_ticker = false;
-        if json_obj.contains_key("A") &&
-            json_obj.contains_key("B") &&
-            json_obj.contains_key("a") &&
-            json_obj.contains_key("b") &&
-            json_obj.contains_key("s") &&
-            json_obj.contains_key("u")
-        {
-            is_book_ticker = true;
-        }
-        is_book_ticker
-    }
-    pub fn channel_depth(json_obj: serde_json::Map<String, Value>) -> bool {
-        let mut is_book_ticker = false;
-        if json_obj.contains_key("asks") &&
-            json_obj.contains_key("bids") &&
-            json_obj.contains_key("lastUpdateId")
-        {
-            is_book_ticker = true;
-        }
-        is_book_ticker
-    }
-}

exchanges/src/gate_swap_rest.rs

@@ -523,8 +523,11 @@ impl GateSwapRest {
         match result {
             Ok(res_data) => {
                 if res_data.code != "200" {
-                    let mut error = res_data;
-                    error.message = format!("错误：{},url：{},相关参数：{}", error.message, base_url, params);
+                    let message = res_data.message;
+                    let json_value: serde_json::Value = serde_json::from_str(&message).unwrap();
+                    let label = json_value["label"].as_str().unwrap();
+                    let mut error = ResponseData::error(res_data.label, label.parse().unwrap());
+                    error.data = format!("请求地址：{},请求参数：{}",base_url,params);
                     error
                 } else {
                     res_data

exchanges/src/gate_swap_ws.rs

@@ -5,7 +5,7 @@ use serde_json::{json, Value};
 use hex;
 use hmac::{Hmac, Mac, NewMac};
 use sha2::Sha512;
-use tracing::trace;
+use tracing::{info, trace};
 use crate::{proxy};
 use tungstenite::client::{AutoStream, connect_with_proxy, ProxyAutoStream};
 use tungstenite::{connect, Message, WebSocket};
@@ -283,6 +283,7 @@ impl GateSwapWs {
     //代理
     async fn proxy_subscription(&self, mut web_socket: WebSocket<ProxyAutoStream>, subscription: Vec<Value>)
     {
+        info!("走代理-链接成功！开始数据读取");
         let label = self.label.clone();
         /*****订阅***/
         for sub in &subscription {
@@ -296,6 +297,7 @@ impl GateSwapWs {
             let msg = web_socket.read_message();
             match msg {
                 Ok(Message::Text(text)) => {
+                    info!("获取推送：{}",text.clone());
                     let res_data = Self::ok_text(label.to_string(), text);
                     if res_data.code == "-200" {
                         trace!("订阅成功：{:?}", res_data.data);
@@ -351,6 +353,7 @@ impl GateSwapWs {
     async fn subscription(&self, mut web_socket: WebSocket<AutoStream>,
                           subscription: Vec<Value>)
     {
+        info!("链接成功！开始数据读取");
         let label = self.label.clone();
         /*****订阅***/
         for sub in &subscription {
@@ -362,6 +365,7 @@ impl GateSwapWs {
             let msg = web_socket.read_message();
             match msg {
                 Ok(Message::Text(text)) => {
+                    info!("获取推送：{}",text.clone());
                     let res_data = Self::ok_text(label.to_string(), text);
                     if res_data.code == "-200" {
                         trace!( "订阅成功：{:?}", res_data.data);

exchanges/src/kucoin_swap_rest.rs

@@ -494,9 +494,14 @@ impl KucoinSwapRest {
 
                     if code != "200000" {
                         let msg = json_value["msg"].as_str().unwrap();
-                        let error = ResponseData::new("".to_string(), code.to_string(),
-                                                      format!("错误：{},url：{},相关参数：{}", msg, base_url, params),
-                                                      "".parse().unwrap());
+                        // let error = ResponseData::new("".to_string(), code.to_string(),
+                        //                               format!("错误：{},url：{},相关参数：{}", msg, base_url, params),
+                        //                               "".parse().unwrap());
+                        // error
+
+                        let mut error = ResponseData::error(res_data.label, msg.parse().unwrap());
+                        error.code = code.parse().unwrap();
+                        error.data = format!("请求地址：{},请求参数：{}",base_url,params);
                         error
                     } else {
                         let data = serde_json::to_string(&json_value["data"]).unwrap();

exchanges/src/kucoin_swap_ws.rs

@@ -4,7 +4,7 @@ use std::io::{Write};
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use tokio::sync::mpsc::Sender;
 use serde_json::{json};
-use tracing::trace;
+use tracing::{info, trace};
 use crate::{proxy};
 use tungstenite::client::{AutoStream, connect_with_proxy, ProxyAutoStream};
 use tungstenite::{connect, Message, WebSocket};
@@ -271,6 +271,7 @@ impl KucoinSwapWs {
     async fn proxy_subscription(&self, mut web_socket: WebSocket<ProxyAutoStream>,
                                 subscription: Vec<String>)
     {
+        info!("走代理-链接成功！开始数据读取");
         let label = self.label.clone();
         /*****消息溜***/
         let mut stdout = io::stdout();
@@ -280,6 +281,7 @@ impl KucoinSwapWs {
             let msg = web_socket.read_message();
             match msg {
                 Ok(Message::Text(text)) => {
+                    info!("获取推送：{}",text.clone());
                     // writeln!(stdout, "Text-响应--{:?}", text.clone()).unwrap();
                     let res_data = Self::ok_text(label.to_string(), text);
 
@@ -338,6 +340,7 @@ impl KucoinSwapWs {
     async fn subscription(&self, mut web_socket: WebSocket<AutoStream>,
                           subscription: Vec<String>)
     {
+        info!("链接成功！开始数据读取");
         let label = self.label.clone();
         /*****消息溜***/
         let mut stdout = io::stdout();
@@ -347,6 +350,7 @@ impl KucoinSwapWs {
             let msg = web_socket.read_message();
             match msg {
                 Ok(Message::Text(text)) => {
+                    info!("获取推送：{}",text.clone());
                     // writeln!(stdout, "Text-响应--{:?}", text.clone()).unwrap();
                     let res_data = Self::ok_text(label.to_string(), text);
 

exchanges/src/lib.rs

@@ -1,8 +1,6 @@
 pub mod proxy;
 pub mod response_base;
 pub mod http_tool;
-pub mod binance_usdt_swap_ws;
-pub mod binance_usdt_swap_rest;
 pub mod binance_spot_rest;
 pub mod binance_spot_ws;
 pub mod gate_spot_rest;
@@ -17,4 +15,3 @@ pub mod binance_swap_ws;
 pub mod okx_swap_rest;
 pub mod binance_swap_rest;
 
-