data-center/exchanges/src/bitget_spot_ws.rs

// use std::sync::Arc;
// use std::sync::atomic::AtomicBool;
// use std::time::Duration;
//
// use chrono::Utc;
// use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
// use ring::hmac;
// use serde_json::{json, Value};
// use tokio::sync::Mutex;
// use tokio_tungstenite::tungstenite::{Error, Message};
// use tracing::{info, trace};
//
// use crate::response_base::ResponseData;
// use crate::socket_tool::{AbstractWsMode, HeartbeatType};
//
// //类型
// pub enum BitgetSpotWsType {
//     Public,
//     Private,
// }
//
// //订阅频道
// #[derive(Clone)]
// pub enum BitgetSpotSubscribeType {
//     PuTicker,
//     PuCandle1m,
//     PuTrade,
//     PuBooks5,
//
//     PrAccount,
//     PrOrders,
// }
//
// //账号信息
// #[derive(Clone)]
// #[allow(dead_code)]
// pub struct BitgetSpotLogin {
//     pub api_key: String,
//     pub secret_key: String,
//     pub passphrase_key: String,
// }
//
//
// #[derive(Clone)]
// pub struct BitgetSpotWs {
//     //类型
//     label: String,
//     //地址
//     address_url: String,
//     //账号
//     login_param: Option<BitgetSpotLogin>,
//     //币对
//     symbol_s: Vec<String>,
//     //订阅
//     subscribe_types: Vec<BitgetSpotSubscribeType>,
//     //心跳间隔
//     heartbeat_time: u64,
// }
//
// impl BitgetSpotWs {
//     /*******************************************************************************************************/
//     /*****************************************获取一个对象****************************************************/
//     /*******************************************************************************************************/
//     pub fn new(is_colo: bool, login_param: Option<BitgetSpotLogin>, ws_type: BitgetSpotWsType) -> BitgetSpotWs {
//         return BitgetSpotWs::new_label("default-BitgetSpotWs".to_string(), is_colo, login_param, ws_type);
//     }
//     pub fn new_label(label: String, is_colo: bool, login_param: Option<BitgetSpotLogin>, ws_type: BitgetSpotWsType) -> BitgetSpotWs
//     {
//         /*******公共频道-私有频道数据组装*/
//         let address_url = match ws_type {
//             BitgetSpotWsType::Public => {
//                 format!("wss://ws.bitget.com/v2/ws/public")
//             }
//             BitgetSpotWsType::Private => {
//                 format!("wss://ws.bitget.com/v2/ws/private")
//             }
//         };
//
//         if is_colo {
//             info!("开启高速(未配置，走普通：{})通道",address_url);
//         } else {
//             info!("走普通通道：{}",address_url);
//         }
//
//         BitgetSpotWs {
//             label,
//             address_url,
//             login_param,
//             symbol_s: vec![],
//             subscribe_types: vec![],
//             heartbeat_time: 1000 * 10,
//         }
//     }
//
//     /*******************************************************************************************************/
//     /*****************************************订阅函数********************************************************/
//     /*******************************************************************************************************/
//     //手动添加订阅信息
//     pub fn set_subscribe(&mut self, subscribe_types: Vec<BitgetSpotSubscribeType>) {
//         self.subscribe_types.extend(subscribe_types);
//     }
//     //手动添加币对
//     pub fn set_symbols(&mut self, mut b_array: Vec<String>) {
//         for symbol in b_array.iter_mut() {
//             // 小写
//             *symbol = symbol.to_uppercase();
//             // 字符串替换
//             *symbol = symbol.replace("-", "");
//             *symbol = symbol.replace("_", "");
//         }
//         self.symbol_s = b_array;
//     }
//     //频道是否需要登录
//     fn contains_pr(&self) -> bool {
//         for t in self.subscribe_types.clone() {
//             if match t {
//                 BitgetSpotSubscribeType::PuTicker => false,
//                 BitgetSpotSubscribeType::PuCandle1m => false,
//                 BitgetSpotSubscribeType::PuTrade => false,
//                 BitgetSpotSubscribeType::PuBooks5 => false,
//
//                 BitgetSpotSubscribeType::PrAccount => false,
//                 BitgetSpotSubscribeType::PrOrders => false
//             } {
//                 return true;
//             }
//         }
//         false
//     }
//     /*******************************************************************************************************/
//     /*****************************************工具函数********************************************************/
//     /*******************************************************************************************************/
//     //订阅枚举解析
//     pub fn enum_to_string(symbol: String, subscribe_type: BitgetSpotSubscribeType) -> Value {
//         match subscribe_type {
//             BitgetSpotSubscribeType::PuTicker => {
//                 json!({
//                     "instType": "SPOT",
//                     "channel": "ticker",
//                     "instId": symbol,
//                 })
//             }
//             BitgetSpotSubscribeType::PuCandle1m => {
//                 json!({
//                     "instType": "SPOT",
//                     "channel": "candle1m",
//                     "instId": symbol,
//                 })
//             }
//             BitgetSpotSubscribeType::PuTrade => {
//                 json!({
//                     "instType": "SPOT",
//                     "channel": "trade",
//                     "instId": symbol,
//                 })
//             }
//             BitgetSpotSubscribeType::PuBooks5 => {
//                 json!({
//                     "instType": "SPOT",
//                     "channel": "books5",
//                     "instId": symbol,
//                 })
//             }
//             BitgetSpotSubscribeType::PrAccount => {
//                 json!({
//                     "instType": "SPOT",
//                     "channel": "account",
//                     "coin": "default"
//                 })
//             }
//             BitgetSpotSubscribeType::PrOrders => {
//                 json!({
//                     "instType": "SPOT",
//                     "channel": "orders",
//                     "instId": symbol
//                 })
//             }
//         }
//     }
//     //订阅信息生成
//     pub fn get_subscription(&self) -> String {
//         let mut params = vec![];
//         for symbol in &self.symbol_s {
//             for subscribe_type in &self.subscribe_types {
//                 let ty_str = Self::enum_to_string(symbol.clone(), subscribe_type.clone());
//                 params.push(ty_str);
//             }
//         }
//         let str = json!({
//             "op": "subscribe",
//             "args": params
//         });
//
//         str.to_string()
//     }
//     //登录数据组装
//     fn log_in_to_str(&self) -> String {
//         let mut login_json_str = "".to_string();
//
//         let mut access_key: String = "".to_string();
//         let mut secret_key: String = "".to_string();
//         let mut passphrase: String = "".to_string();
//         match self.login_param.clone() {
//             None => {}
//             Some(param) => {
//                 access_key = param.api_key;
//                 secret_key = param.secret_key;
//                 passphrase = param.passphrase_key;
//             }
//         }
//         if access_key.len() > 0 || secret_key.len() > 0 || passphrase.len() > 0 {
//             let timestamp = Utc::now().timestamp().to_string();
//             // 时间戳 + 请求类型+ 请求参数字符串
//             let message = format!("{}GET{}", timestamp, "/user/verify");
//             let hmac_key = ring::hmac::Key::new(hmac::HMAC_SHA256, secret_key.as_bytes());
//             let result = ring::hmac::sign(&hmac_key, &message.as_bytes());
//             let sign = base64::encode(result);
//
//             let login_json = json!({
//                               "op": "login",
//                               "args": [{
//                                 "apiKey": access_key,
//                                 "passphrase": passphrase,
//                                 "timestamp": timestamp,
//                                 "sign": sign
//                               }]
//                         });
//
//             trace!("---login_json：{0}", login_json.to_string());
//             trace!("--登录：{}", login_json.to_string());
//             login_json_str = login_json.to_string();
//         }
//         login_json_str
//     }
//     /*******************************************************************************************************/
//     /*****************************************socket基本*****************************************************/
//     /*******************************************************************************************************/
//     //链接
//     pub async fn ws_connect_async(&mut self,
//                                   is_shutdown_arc: Arc<AtomicBool>,
//                                   write_tx_am: &Arc<Mutex<UnboundedSender<Message>>>,
//                                   write_rx: UnboundedReceiver<Message>,
//                                   read_tx: UnboundedSender<ResponseData>) -> Result<(), Error>
//     {
//         let login_is = self.contains_pr();
//         let subscription = self.get_subscription();
//         let address_url = self.address_url.clone();
//         let label = self.label.clone();
//         let heartbeat_time = self.heartbeat_time.clone();
//
//
//         //心跳-- 方法内部线程启动
//         let write_tx_clone1 = Arc::clone(write_tx_am);
//         tokio::spawn(async move {
//             trace!("线程-异步心跳-开始");
//             AbstractWsMode::ping_or_pong(write_tx_clone1, HeartbeatType::Ping, heartbeat_time).await;
//             trace!("线程-异步心跳-结束");
//         });
//
//         //设置订阅
//         let mut subscribe_array = vec![];
//         if login_is {
//             //登录相关
//             let login_str = self.log_in_to_str();
//             let write_tx_clone2 = Arc::clone(write_tx_am);
//             AbstractWsMode::send_subscribe(write_tx_clone2, Message::Text(login_str)).await;
//             tokio::time::sleep(Duration::from_millis(1000 * 3)).await;
//         }
//         subscribe_array.push(subscription.to_string());
//
//         //链接
//         let t2 = tokio::spawn(async move {
//             trace!("线程-异步链接-开始");
//             match AbstractWsMode::ws_connect_async(is_shutdown_arc, address_url.clone(),
//                                                    label.clone(), subscribe_array,
//                                                    write_rx, read_tx,
//                                                    Self::message_text,
//                                                    Self::message_ping,
//                                                    Self::message_pong,
//             ).await {
//                 Ok(_) => { trace!("线程-异步链接-结束"); }
//                 Err(e) => { trace!("发生异常：bitget-现货链接关闭-{:?}",e); }
//             }
//             trace!("线程-异步链接-结束");
//         });
//         tokio::try_join!(t2).unwrap();
//         trace!("线程-心跳与链接-结束");
//
//         Ok(())
//     }
//     /*******************************************************************************************************/
//     /*****************************************数据解析*****************************************************/
//     /*******************************************************************************************************/
//     //数据解析-Text
//     pub fn message_text(text: String) -> Option<ResponseData> {
//         let response_data = Self::ok_text(text);
//         Option::from(response_data)
//     }
//     //数据解析-ping
//     pub fn message_ping(_pi: Vec<u8>) -> Option<ResponseData> {
//         return Option::from(ResponseData::new("".to_string(), "-300".to_string(), "success".to_string(), "".to_string()));
//     }
//     //数据解析-pong
//     pub fn message_pong(_po: Vec<u8>) -> Option<ResponseData> {
//         return Option::from(ResponseData::new("".to_string(), "-301".to_string(), "success".to_string(), "".to_string()));
//     }
//     //数据解析
//     pub fn ok_text(text: String) -> ResponseData
//     {
//         // trace!("原始数据：{}", text);
//         // trace!("大小：{}", text.capacity());
//         let mut res_data = ResponseData::new("".to_string(), "200".to_string(), "success".to_string(), "".to_string());
//         let json_value: serde_json::Value = serde_json::from_str(&text).unwrap();
//
//         // {"event":"login","code":0}
//         if json_value.get("event").is_some() && json_value["event"].as_str() == Option::from("login") {
//             if json_value.get("code").is_some() && json_value["code"] == 0 {
//                 res_data.message = format!("登录成功");
//             } else {
//                 res_data.message = format!("登录失败:({},{})", json_value.get("code").as_ref().unwrap(), json_value.get("msg").as_ref().unwrap());
//             }
//             res_data.channel = format!("login");
//             res_data.code = "-200".to_string();
//             res_data
//         } else if json_value.get("event").is_some() && json_value["event"].as_str() == Option::from("subscribe") {
//             // trace!("解析-订阅反馈：{}", text);
//             res_data.code = "-201".to_string();
//             res_data.data = text;
//             res_data.channel = format!("{}", json_value["arg"]["channel"].as_str().unwrap());
//             res_data
//         } else if json_value.get("action").is_some() {
//             // trace!("解析-推送数据：{}", text);
//             res_data.data = json_value["data"].to_string();
//             if res_data.data == "[]" {
//                 res_data.code = "".to_string();
//             } else {
//                 res_data.code = "200".to_string();
//             }
//             res_data.channel = format!("{}", json_value["arg"]["channel"].as_str().unwrap());
//             res_data.reach_time = json_value["ts"].as_i64().unwrap() * 1000;
//             res_data
//         } else {
//             res_data
//         }
//     }
// }