代码优化，兼并channel 方式回调方式

exchanges/src/binance_usdt_swap_ws.rs

@@ -3,7 +3,7 @@ use std::future::Future;
 use std::{io};
 use std::io::{Write};
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
-use std::sync::Arc;
+use std::sync::{Arc, mpsc};
 use serde_json::{json, Value};
 use tokio::sync::Mutex;
 use crate::proxy;
@@ -26,6 +26,8 @@ pub struct BinanceUsdtSwapWs {
     is_login: bool,
     //登陆所需参数
     login_param: BTreeMap<String, String>,
+
+    // sender: mpsc::Sender<ResponseData>,
 }
 
 impl BinanceUsdtSwapWs {
@@ -34,7 +36,9 @@ impl BinanceUsdtSwapWs {
     /*******************************************************************************************************/
     pub fn new(is_colo: bool,
                is_login: bool,
-               login_param: BTreeMap<String, String>) -> BinanceUsdtSwapWs
+               login_param: BTreeMap<String, String>,
+               // sender: mpsc::Sender<ResponseData>,
+    ) -> BinanceUsdtSwapWs
     {
         let mut base_url = String::from("");
         if is_colo {
@@ -60,6 +64,7 @@ impl BinanceUsdtSwapWs {
             port: port_v,
             is_login,
             login_param,
+            // sender,
         }
     }
 
@@ -242,7 +247,9 @@ impl BinanceUsdtSwapWs {
             match msg {
                 Ok(Message::Text(text)) => {
                     let req_data = Self::ok_text(text);
+                    // self.sender.send(req_data).unwrap();
                     // writeln!(stdout, "Pong-响应--{:?}", req_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;
@@ -252,6 +259,8 @@ impl BinanceUsdtSwapWs {
                 }
                 Ok(Message::Ping(s)) => {
                     println!("Ping-响应--{:?}", String::from_utf8(s));
+                    let _ = web_socket.write_message(Message::Pong(Vec::from("pong")));
+                    println!("pong-回应")
                 }
                 Ok(Message::Pong(s)) => {
                     println!("Pong-响应--{:?}", String::from_utf8(s));

exchanges/src/gate_swap_rest.rs

@@ -139,10 +139,64 @@ impl GateSwapRest {
         data
     }
 
+
+    // » contract	body	string	是	合约标识
+    // » size	body	integer(int64)	是	必选。交易数量，正数为买入，负数为卖出。平仓委托则设置为0。
+    // » iceberg	body	integer(int64)	否	冰山委托显示数量。0为完全不隐藏。注意，隐藏部分成交按照taker收取手续费。
+    // » price	body	string	否	委托价。价格为0并且tif为ioc，代表市价委托。
+    // » close	body	boolean	否	设置为 true 的时候执行平仓操作，并且size应设置为0
+    // » reduce_only	body	boolean	否	设置为 true 的时候，为只减仓委托
+    // » tif	body	string	否	Time in force 策略，市价单当前只支持 ioc 模式
+    // » text	body	string	否	订单自定义信息，用户可以用该字段设置自定义 ID，用户自定义字段必须满足以下条件：
+    // » auto_size	body	string	否	双仓模式下用于设置平仓的方向，close_long 平多头， close_short 平空头，需要同时设置 size 为 0
+    // » stp_act	body	string	否	Self-Trading Prevention Action,用户可以用该字段设置自定义限制自成交策略。
+    //合约交易：市价价下单
+    pub async fn swap_bazaar_order(&self,
+                             text: String,
+                             origin_side: String,
+                             settle: String,
+                             contract: String,
+                             size: i64) -> ResponseData {
+        let mut reduce_only = false;
+        let mut param = serde_json::json!({
+            "text":text,
+            "contract":contract,
+            "price":"0",
+            "size":size,
+        });
+
+        let req = match origin_side.as_str() {
+            "kd" => {
+                reduce_only = false;
+                true
+            }
+            "pd" => {
+                reduce_only = true;
+                true
+            }
+            "kk" => {
+                reduce_only = false;
+                true
+            }
+            "pk" => {
+                reduce_only = true;
+                true
+            }
+            _ => { false } // 处理未知请求类型
+        };
+        if req {
+            param.as_object_mut().unwrap().insert("reduce_only".to_string(), serde_json::json!(reduce_only));
+        }
+
+        let data = self.swap_order(settle, param).await;
+        data
+    }
+
+
     //合约交易下单
-    pub async fn swap_order(&self,
-                            settle: String,
-                            params: serde_json::Value,
+    async fn swap_order(&self,
+                        settle: String,
+                        params: serde_json::Value,
     ) -> ResponseData {
         let data = self.request("POST".to_string(),
                                 "/api/v4".to_string(),
@@ -152,6 +206,8 @@ impl GateSwapRest {
         ).await;
         data
     }
+
+
     //设置持仓模式
     pub async fn setting_dual_mode(&self,
                                    settle: String,
@@ -187,12 +243,12 @@ impl GateSwapRest {
     }
     //交易账户互转
     pub async fn wallet_transfers(&self,
-                                  currency:String,
-                                  from:String,
-                                  to:String,
-                                  amount:String,
-                                  currency_pair:String,
-                                  settle:String,
+                                  currency: String,
+                                  from: String,
+                                  to: String,
+                                  amount: String,
+                                  currency_pair: String,
+                                  settle: String,
     ) -> ResponseData {
         let params = serde_json::json!({
                 "currency":settle,

exchanges/src/http_tool.rs

@@ -1,4 +1,3 @@
-use std::collections::{BTreeMap};
 use reqwest::{Client};
 use reqwest::header::HeaderMap;
 use crate::response_base::ResponseData;
@@ -36,14 +35,6 @@ impl RestTool {
         let url = format!("{}{}", self.base_url.to_string(), request_path);
         let request_type = request_type.clone().to_uppercase();
         let addrs_url = format!("{}?{}", url.clone(), RestTool::parse_params_to_str(params.clone()));
-        // println!("------{}", addrs_url);
-
-        // let body = params.clone();
-        let body: serde_json::Value = serde_json::from_str(&params).unwrap();
-        // println!("---params：{}", params);
-        // println!("---http_toll-body：{}", body);
-        // let z = BTreeMap::new();
-
 
         let req = match request_type.as_str() {
             "GET" => self.client.get(addrs_url.clone()).headers(headers),
@@ -67,56 +58,6 @@ impl RestTool {
         Ok(req_data)
     }
 
-    pub async fn get(&self, request_path: &str, params: BTreeMap<&str, &str>) -> Result<ResponseData, reqwest::Error> {
-        let req_data: ResponseData;
-        /*请求接口与 地址*/
-        let url = format!("{}{}", self.base_url.to_string(), request_path);
-        let req = self.client
-            .get(url)
-            .json(&params)
-            ;
-
-        let response = req.send()
-            .await?;
-        // 检查响应是否成功
-        if response.status().is_success() {
-            // 读取响应的内容
-            let body = response.text().await?;
-            req_data = ResponseData::new("200".to_string(), "success".to_string(), body);
-        } else {
-            let body = response.text().await?;
-            req_data = ResponseData::error(body.to_string())
-        }
-        Ok(req_data)
-    }
-
-    pub async fn post_v(&self, request_path: String, params: BTreeMap<&str, &str>) -> Result<ResponseData, reqwest::Error> {
-        let req_data: ResponseData;
-        /*请求接口与 地址*/
-        let url = format!("{}{}", self.base_url.to_string(), request_path);
-        let req = self.client
-            .post(url)
-            .json(&params)
-            ;
-
-        let response = req.send()
-            .await?;
-        // 检查响应是否成功
-        //println!("---状态：{:?},{}", response.status(), response.status().is_success());
-        if response.status().is_success() {
-            // 读取响应的内容
-            let body = response.text().await?;
-            //println!("okx_order-Response body:\n{}", body);
-            req_data = ResponseData::new("200".to_string(), "success".to_string(), body);
-        } else {
-            let body = response.text().await?;
-            //println!("okx_order-Request failed with status: {}", body);
-            req_data = ResponseData::error(body.to_string())
-        }
-        Ok(req_data)
-    }
-
-
     //map数据转 get请求参数
     pub fn parse_params_to_str(parameters: String) -> String {
         let mut params_str = String::from("");
@@ -141,16 +82,6 @@ impl RestTool {
         }
         // println!("---json-转字符串拼接:{}",params_str);
         params_str.to_string()
-        // for (key, value) in json_obj.as_object().unwrap().iter() {
-        //     // println!("Key: {}", key);
-        //     // println!("Value: {}", value);
-        // }
-
-        // json_obj
-        //     .into_iter()
-        //     .map(|(key, value)| format!("{}={}", key, value))
-        //     .collect::<Vec<String>>()
-        //     .join("&")
     }
     //req_data 解析
     pub fn req_data_analysis(result: Result<ResponseData, reqwest::Error>) -> ResponseData {

exchanges/src/lib.rs

@@ -9,5 +9,6 @@ pub mod gate_spot_rest;
 pub mod gate_spot_ws;
 pub mod gate_swap_ws;
 pub mod gate_swap_rest;
+pub mod socket_tool;
 
 

exchanges/src/proxy.rs

@@ -3,8 +3,8 @@ use std::env;
 /**代理工具*/
 #[derive(Debug)]
 pub struct ParsingDetail {
-    pub(crate) ip_address: String,
-    pub(crate) port: String,
+    pub ip_address: String,
+    pub port: String,
 }
 
 impl ParsingDetail {