|
|
@@ -1,501 +1,523 @@
|
|
|
-// use std::collections::{BTreeMap};
|
|
|
-// use std::net::{IpAddr, Ipv4Addr, SocketAddr};
|
|
|
-// use std::sync::Arc;
|
|
|
-// use std::sync::atomic::{AtomicBool, Ordering};
|
|
|
-// use std::thread;
|
|
|
-// use std::time::Duration;
|
|
|
-// use chrono::Utc;
|
|
|
-// use tokio::sync::mpsc::Sender;
|
|
|
-// use serde_json::{json, Value};
|
|
|
-// use tracing::{error, info, trace};
|
|
|
-// use crate::{proxy};
|
|
|
-// use url::Url;
|
|
|
-// use crate::proxy::ParsingDetail;
|
|
|
-// use crate::response_base::ResponseData;
|
|
|
-// use crate::utils::get_time_microsecond;
|
|
|
-//
|
|
|
-// use ring::hmac;
|
|
|
-//
|
|
|
-// pub enum BitgetWsType {
|
|
|
-// //频道类型
|
|
|
-// Public,
|
|
|
-// Private,
|
|
|
-// }
|
|
|
-//
|
|
|
-//
|
|
|
-// #[derive(Clone)] //订阅枚举
|
|
|
-// pub enum BitgetSubscribeType {
|
|
|
-// PuTicker,
|
|
|
-// PuCandle1m,
|
|
|
-// PuTrade,
|
|
|
-// PuBooks5,
|
|
|
-//
|
|
|
-// PrAccount,
|
|
|
-// PrOrders,
|
|
|
-// }
|
|
|
-//
|
|
|
-// #[derive(Clone)]
|
|
|
-// pub struct BitgetSpotWs {
|
|
|
-// pub label: String,
|
|
|
-// request_url: String,
|
|
|
-// //实际ws 链接地址
|
|
|
-// proxy: ParsingDetail,
|
|
|
-// //账号信息
|
|
|
-// login_param: BTreeMap<String, String>,
|
|
|
-// //kuconis特殊参数
|
|
|
-// symbol_s: Vec<String>,
|
|
|
-// //订阅币对
|
|
|
-// subscribe_types: Vec<BitgetSubscribeType>,
|
|
|
-// //订阅信息
|
|
|
-// sender: Sender<ResponseData>, //数据通道
|
|
|
-// }
|
|
|
-//
|
|
|
-// impl BitgetSpotWs {
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************获取一个对象****************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// pub fn new(is_colo: bool,
|
|
|
-// login_param: BTreeMap<String, String>,
|
|
|
-// ws_type: BitgetWsType,
|
|
|
-// sender: Sender<ResponseData>,
|
|
|
-// ) -> BitgetSpotWs
|
|
|
-// {
|
|
|
-// return BitgetSpotWs::new_label("default-BitgetSpotWs".to_string(), is_colo, login_param, ws_type, sender);
|
|
|
-// }
|
|
|
-// pub fn new_label(label: String, is_colo: bool,
|
|
|
-// login_param: BTreeMap<String, String>,
|
|
|
-// ws_type: BitgetWsType,
|
|
|
-// sender: Sender<ResponseData>,
|
|
|
-// ) -> BitgetSpotWs
|
|
|
-// {
|
|
|
-// /*******走代理：根据环境变量配置来决定,如果配置了走代理，没有配置不走*******/
|
|
|
-// let parsing_detail = proxy::ParsingDetail::parsing_environment_variables();
|
|
|
-//
|
|
|
-//
|
|
|
-// let request_url = match ws_type {
|
|
|
-// BitgetWsType::Public => {
|
|
|
-// format!("wss://ws.bitget.com/v2/ws/public")
|
|
|
-// }
|
|
|
-// BitgetWsType::Private => {
|
|
|
-// format!("wss://ws.bitget.com/v2/ws/private")
|
|
|
-// }
|
|
|
-// };
|
|
|
-//
|
|
|
-// if is_colo {
|
|
|
-// info!("开启高速(未配置，走普通：{})通道",request_url);
|
|
|
-// } else {
|
|
|
-// info!("走普通通道：{}",request_url);
|
|
|
-// }
|
|
|
-// /*****返回结构体*******/
|
|
|
-// BitgetSpotWs {
|
|
|
-// label,
|
|
|
-// request_url,
|
|
|
-// proxy: parsing_detail,
|
|
|
-// login_param,
|
|
|
-// symbol_s: vec![],
|
|
|
-// subscribe_types: vec![],
|
|
|
-// sender,
|
|
|
-// }
|
|
|
-// }
|
|
|
-//
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************订阅函数********************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// //手动添加订阅信息
|
|
|
-// pub fn set_subscribe(&mut self, subscribe_types: Vec<BitgetSubscribeType>) {
|
|
|
-// self.subscribe_types.extend(subscribe_types);
|
|
|
-// }
|
|
|
-// //自定义
|
|
|
-// pub async fn custom_subscribe(&mut self, bool_v1: Arc<AtomicBool>, b_array: Vec<String>)
|
|
|
-// {
|
|
|
-// let mut symbol_s = b_array.clone();
|
|
|
-// for symbol in symbol_s.iter_mut() {
|
|
|
-// // 大写
|
|
|
-// *symbol = symbol.to_uppercase();
|
|
|
-// // 字符串替换
|
|
|
-// *symbol = symbol.replace("-", "");
|
|
|
-// *symbol = symbol.replace("_", "");
|
|
|
-// }
|
|
|
-// self.symbol_s = symbol_s;
|
|
|
-// self.run(bool_v1).await;
|
|
|
-// }
|
|
|
-//
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************工具函数********************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// //订阅枚举解析
|
|
|
-// pub fn enum_to_string(symbol: String, subscribe_type: BitgetSubscribeType) -> Value {
|
|
|
-// match subscribe_type {
|
|
|
-// BitgetSubscribeType::PuTicker => {
|
|
|
-// json!({
|
|
|
-// "instType": "SPOT",
|
|
|
-// "channel": "ticker",
|
|
|
-// "instId": symbol,
|
|
|
-// })
|
|
|
-// }
|
|
|
-// BitgetSubscribeType::PuCandle1m => {
|
|
|
-// json!({
|
|
|
-// "instType": "SPOT",
|
|
|
-// "channel": "candle1m",
|
|
|
-// "instId": symbol,
|
|
|
-// })
|
|
|
-// }
|
|
|
-// BitgetSubscribeType::PuTrade => {
|
|
|
-// json!({
|
|
|
-// "instType": "SPOT",
|
|
|
-// "channel": "trade",
|
|
|
-// "instId": symbol,
|
|
|
-// })
|
|
|
-// }
|
|
|
-// BitgetSubscribeType::PuBooks5 => {
|
|
|
-// json!({
|
|
|
-// "instType": "SPOT",
|
|
|
-// "channel": "books5",
|
|
|
-// "instId": symbol,
|
|
|
-// })
|
|
|
-// }
|
|
|
-// BitgetSubscribeType::PrAccount => {
|
|
|
-// json!({
|
|
|
-// "instType": "SPOT",
|
|
|
-// "channel": "account",
|
|
|
-// "coin": "default"
|
|
|
-// })
|
|
|
-// }
|
|
|
-// BitgetSubscribeType::PrOrders => {
|
|
|
-// json!({
|
|
|
-// "instType": "SPOT",
|
|
|
-// "channel": "orders",
|
|
|
-// "instId": symbol
|
|
|
-// })
|
|
|
-// }
|
|
|
-// }
|
|
|
-// }
|
|
|
-// //组装订阅数据
|
|
|
-// pub fn get_subscription(&self) -> String {
|
|
|
-// let mut args = 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());
|
|
|
-// args.push(ty_str);
|
|
|
-// }
|
|
|
-// }
|
|
|
-// let str = json!({
|
|
|
-// "op": "subscribe",
|
|
|
-// "args": args
|
|
|
-// });
|
|
|
-//
|
|
|
-// str.to_string()
|
|
|
-// }
|
|
|
-//
|
|
|
-// // fn sign(secret_key: String, channel: String, event: String, time: String) -> String {
|
|
|
-// // let message = format!("channel={}&event={}&time={}", channel, event, time);
|
|
|
-// // let mut mac = Hmac::<Sha512>::new_varkey(secret_key.as_bytes()).expect("Failed to create HMAC");
|
|
|
-// // mac.update(message.as_bytes());
|
|
|
-// // let result = mac.finalize().into_bytes();
|
|
|
-// // let sign = hex::encode(result);
|
|
|
-// // sign
|
|
|
-// // }
|
|
|
-//
|
|
|
-// //组装登陆数据
|
|
|
-// 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();
|
|
|
-// for (key, value) in &self.login_param {
|
|
|
-// if key == "access_key" {
|
|
|
-// access_key = value.parse().unwrap();
|
|
|
-// } else if key == "secret_key" {
|
|
|
-// secret_key = value.parse().unwrap();
|
|
|
-// } else if key == "pass_key" {
|
|
|
-// passphrase = value.parse().unwrap();
|
|
|
-// }
|
|
|
-// }
|
|
|
-// 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基本*****************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// async fn run(&mut self, bool_v1: Arc<AtomicBool>)
|
|
|
-// {
|
|
|
-// //订阅信息组装
|
|
|
-// let subscription = self.get_subscription();
|
|
|
-// loop {
|
|
|
-// tokio::time::sleep(Duration::from_millis(5000)).await;
|
|
|
-// trace!("要连接咯~~！！{}", self.request_url);
|
|
|
-//
|
|
|
-// let request_url = Url::parse(self.request_url.as_str()).unwrap();
|
|
|
-// //1. 判断是否需要代理,根据代理地址是否存来选择
|
|
|
-// if self.proxy.ip_address.len() > 0 {
|
|
|
-// let ip_array: Vec<&str> = self.proxy.ip_address.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.proxy.port.parse().unwrap());
|
|
|
-// 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) => {
|
|
|
-// let bool_v1_clone = Arc::clone(&bool_v1);
|
|
|
-// self.proxy_subscription(bool_v1_clone, ws.0, subscription.clone()).await;
|
|
|
-// }
|
|
|
-// Err(err) => {
|
|
|
-// error!("Can't connect(无法连接): {}", err);
|
|
|
-// }
|
|
|
-// };
|
|
|
-// } else {
|
|
|
-// match connect(request_url.clone()) {
|
|
|
-// Ok(ws) => {
|
|
|
-// let bool_v1_clone = Arc::clone(&bool_v1);
|
|
|
-// self.subscription(bool_v1_clone, ws.0, subscription.clone()).await;
|
|
|
-// }
|
|
|
-// Err(err) => {
|
|
|
-// // 连接失败时执行的操作
|
|
|
-// error!("Can't connect(无法连接): {}", err);
|
|
|
-// // 返回一个默认的 WebSocket 对象或其他适当的值
|
|
|
-// // 或者根据需要触发 panic 或返回错误信息
|
|
|
-// }
|
|
|
-// };
|
|
|
-// }
|
|
|
-// trace!("退出来咯");
|
|
|
-//
|
|
|
-// let bool_v1_clone = Arc::clone(&bool_v1);
|
|
|
-// let bool_v1_v = bool_v1_clone.load(Ordering::SeqCst);
|
|
|
-// if !bool_v1_v {
|
|
|
-// break;
|
|
|
-// }
|
|
|
-// }
|
|
|
-// }
|
|
|
-//
|
|
|
-// //代理
|
|
|
-// async fn proxy_subscription(&self, bool_v1: Arc<AtomicBool>, mut web_socket: WebSocket<ProxyAutoStream>, subscription: String)
|
|
|
-// {
|
|
|
-// info!("走代理-链接成功！开始数据读取");
|
|
|
-// let label = self.label.clone();
|
|
|
-// /*****登陆***/
|
|
|
-// let login_str = self.log_in_to_str();
|
|
|
-// if login_str != "" {
|
|
|
-// let _ = web_socket.write_message(Message::Text(login_str));
|
|
|
-// tokio::time::sleep(Duration::from_secs(3)).await;
|
|
|
-// }
|
|
|
-// /*****订阅***/
|
|
|
-// if subscription.len() > 0 {
|
|
|
-// info!("订阅信息：{}", subscription);
|
|
|
-// web_socket.write_message(Message::Text(subscription.clone()))
|
|
|
-// .unwrap();
|
|
|
-// }
|
|
|
-// /*****消息溜***/
|
|
|
-// let mut ping_timeout = chrono::Utc::now().timestamp_millis();
|
|
|
-// loop {
|
|
|
-// tokio::time::sleep(Duration::from_millis(1)).await;
|
|
|
-// let msg = web_socket.read_message();
|
|
|
-// match msg {
|
|
|
-// Ok(Message::Text(text)) => {
|
|
|
-// let get_time = chrono::Utc::now().timestamp_millis();
|
|
|
-// if (get_time - ping_timeout) >= (1000 * 30) {
|
|
|
-// trace!("30s 一次主动发送心跳包!");
|
|
|
-// let _ = web_socket.write_message(Message::Ping(Vec::from("ping")));
|
|
|
-// ping_timeout = get_time;
|
|
|
-// }
|
|
|
-//
|
|
|
-// // let st_time = get_time_microsecond();
|
|
|
-// let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
-// // trace!("解析数据耗时：：{}", st_time - get_time_microsecond());
|
|
|
-// res_data.time = get_time_microsecond();
|
|
|
-// if res_data.code == "-200" {
|
|
|
-// trace!("订阅成功：{:?}", res_data.data);
|
|
|
-// } else if res_data.code == "-201" {
|
|
|
-// trace!("登陆：{:?}", res_data);
|
|
|
-// } else if res_data.code == "-202" {
|
|
|
-// trace!("空数据不予返回：{:?}", res_data);
|
|
|
-// } else {
|
|
|
-// let sender = self.sender.clone();
|
|
|
-// // let prev_time = Utc::now().timestamp_millis();
|
|
|
-// tokio::spawn(async move {
|
|
|
-// // info!("{:04} {}", Utc::now().timestamp_millis() - prev_time, res_data.channel);
|
|
|
-// sender.send(res_data).await.unwrap();
|
|
|
-// });
|
|
|
-// tokio::spawn(async move {});
|
|
|
-// }
|
|
|
-// }
|
|
|
-// Ok(Message::Ping(s)) => {
|
|
|
-// trace!( "Ping-响应--{:?}", String::from_utf8(s.clone()));
|
|
|
-// let _ = web_socket.write_message(Message::Pong(Vec::from("pong")));
|
|
|
-// trace!( "回应-pong---{:?}", String::from_utf8(s.clone()));
|
|
|
-// }
|
|
|
-// Ok(Message::Pong(s)) => {
|
|
|
-// // trace!("Pong-响应--{:?}", String::from_utf8(s));
|
|
|
-// trace!("Pong-响应--{:?}", String::from_utf8(s.clone()));
|
|
|
-// }
|
|
|
-// Ok(Message::Close(_)) => {
|
|
|
-// // trace!("socket 关闭: ");
|
|
|
-// trace!( "Close-响应");
|
|
|
-// break;
|
|
|
-// }
|
|
|
-// Err(error) => {
|
|
|
-// // trace!("Error receiving message: {}", error);
|
|
|
-// error!( "Err-响应{}", error);
|
|
|
-// break;
|
|
|
-// }
|
|
|
-// _ => {}
|
|
|
-// }
|
|
|
-//
|
|
|
-// let bool_v1_v = bool_v1.load(Ordering::SeqCst);
|
|
|
-// if !bool_v1_v {
|
|
|
-// break;
|
|
|
-// }
|
|
|
-// }
|
|
|
-// web_socket.close(None).unwrap();
|
|
|
-// }
|
|
|
-//
|
|
|
-// //非代理
|
|
|
-// async fn subscription(&self, bool_v1: Arc<AtomicBool>, mut web_socket: WebSocket<AutoStream>,
|
|
|
-// subscription: String)
|
|
|
-// {
|
|
|
-// info!("链接成功！开始数据读取");
|
|
|
-// let label = self.label.clone();
|
|
|
-// /*****登陆***/
|
|
|
-// let login_str = self.log_in_to_str();
|
|
|
-// if login_str != "" {
|
|
|
-// let _ = web_socket.write_message(Message::Text(login_str));
|
|
|
-// thread::sleep(Duration::from_secs(3));
|
|
|
-// }
|
|
|
-// /*****订阅***/
|
|
|
-// if subscription.len() > 0 {
|
|
|
-// trace!("订阅信息：{}",subscription);
|
|
|
-// web_socket.write_message(Message::Text(subscription))
|
|
|
-// .unwrap();
|
|
|
-// }
|
|
|
-// /*****消息溜***/
|
|
|
-// let mut ping_timeout = chrono::Utc::now().timestamp_millis();
|
|
|
-// loop {
|
|
|
-// tokio::time::sleep(Duration::from_millis(1)).await;
|
|
|
-// let msg = web_socket.read_message();
|
|
|
-// match msg {
|
|
|
-// Ok(Message::Text(text)) => {
|
|
|
-// let get_time = chrono::Utc::now().timestamp_millis();
|
|
|
-// if (get_time - ping_timeout) >= (1000 * 30) {
|
|
|
-// trace!("30s 一次主动发送心跳包!");
|
|
|
-// let _ = web_socket.write_message(Message::Ping(Vec::from("ping")));
|
|
|
-// ping_timeout = get_time;
|
|
|
-// }
|
|
|
-//
|
|
|
-// // let st_time = get_time_microsecond();
|
|
|
-// let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
-// // trace!("解析数据耗时：：{}", st_time - get_time_microsecond());
|
|
|
-// res_data.time = get_time_microsecond();
|
|
|
-// if res_data.code == "-200" {
|
|
|
-// trace!("订阅成功：{:?}", res_data.data);
|
|
|
-// } else if res_data.code == "-201" {
|
|
|
-// trace!("登陆：{:?}", res_data);
|
|
|
-// } else if res_data.code == "-202" {
|
|
|
-// trace!("空数据不予返回：{:?}", res_data);
|
|
|
-// } else {
|
|
|
-// let sender = self.sender.clone();
|
|
|
-// tokio::spawn(async move {
|
|
|
-// sender.send(res_data).await.unwrap();
|
|
|
-// });
|
|
|
-// tokio::spawn(async move {});
|
|
|
-// }
|
|
|
-// }
|
|
|
-// Ok(Message::Ping(s)) => {
|
|
|
-// trace!( "Ping-响应--{:?}", String::from_utf8(s.clone()));
|
|
|
-// let _ = web_socket.write_message(Message::Pong(Vec::from("pong")));
|
|
|
-// trace!( "回应-pong---{:?}", String::from_utf8(s.clone()));
|
|
|
-// }
|
|
|
-// Ok(Message::Pong(s)) => {
|
|
|
-// // trace!("Pong-响应--{:?}", String::from_utf8(s));
|
|
|
-// trace!("Pong-响应--{:?}", String::from_utf8(s.clone()));
|
|
|
-// }
|
|
|
-// Ok(Message::Close(_)) => {
|
|
|
-// // trace!("socket 关闭: ");
|
|
|
-// trace!( "Close-响应");
|
|
|
-// break;
|
|
|
-// }
|
|
|
-// Err(error) => {
|
|
|
-// // trace!("Error receiving message: {}", error);
|
|
|
-// error!( "Err-响应{}", error);
|
|
|
-// break;
|
|
|
-// }
|
|
|
-// _ => {}
|
|
|
-// }
|
|
|
-//
|
|
|
-// let bool_v1_v = bool_v1.load(Ordering::SeqCst);
|
|
|
-// if !bool_v1_v {
|
|
|
-// break;
|
|
|
-// }
|
|
|
-// }
|
|
|
-// web_socket.close(None).unwrap();
|
|
|
-// }
|
|
|
-//
|
|
|
-// //数据解析
|
|
|
-// pub fn ok_text(label: String, text: String) -> ResponseData
|
|
|
-// {
|
|
|
-// // trace!("原始数据：{}", text);
|
|
|
-// // trace!("大小：{}", text.capacity());
|
|
|
-// 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();
|
|
|
-//
|
|
|
-// // {"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 = "-201".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 = "-200".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 = "-202".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
|
|
|
-// }
|
|
|
-// }
|
|
|
-// }
|
|
|
+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};
|
|
|
+use crate::utils::get_time_microsecond;
|
|
|
+
|
|
|
+//类型
|
|
|
+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 * 20,
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************订阅函数********************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ //手动添加订阅信息
|
|
|
+ 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,
|
|
|
+ bool_v1: 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::Pong, 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!("线程-异步链接-开始");
|
|
|
+ AbstractWsMode::ws_connect_async(bool_v1, address_url.clone(),
|
|
|
+ label.clone(), subscribe_array,
|
|
|
+ write_rx, read_tx,
|
|
|
+ Self::message_text,
|
|
|
+ Self::message_ping,
|
|
|
+ Self::message_pong,
|
|
|
+ ).await.expect("币安-期货");
|
|
|
+ trace!("线程-异步链接-结束");
|
|
|
+ });
|
|
|
+ tokio::try_join!(t2).unwrap();
|
|
|
+ trace!("线程-心跳与链接-结束");
|
|
|
+
|
|
|
+ Ok(())
|
|
|
+ }
|
|
|
+
|
|
|
+ // //代理
|
|
|
+ // async fn proxy_subscription(&self, bool_v1: Arc<AtomicBool>, mut web_socket: WebSocket<ProxyAutoStream>, subscription: String)
|
|
|
+ // {
|
|
|
+ // info!("走代理-链接成功！开始数据读取");
|
|
|
+ // let label = self.label.clone();
|
|
|
+ // /*****登陆***/
|
|
|
+ // let login_str = self.log_in_to_str();
|
|
|
+ // if login_str != "" {
|
|
|
+ // let _ = web_socket.write_message(Message::Text(login_str));
|
|
|
+ // tokio::time::sleep(Duration::from_secs(3)).await;
|
|
|
+ // }
|
|
|
+ // /*****订阅***/
|
|
|
+ // if subscription.len() > 0 {
|
|
|
+ // info!("订阅信息：{}", subscription);
|
|
|
+ // web_socket.write_message(Message::Text(subscription.clone()))
|
|
|
+ // .unwrap();
|
|
|
+ // }
|
|
|
+ // /*****消息溜***/
|
|
|
+ // let mut ping_timeout = chrono::Utc::now().timestamp_millis();
|
|
|
+ // loop {
|
|
|
+ // tokio::time::sleep(Duration::from_millis(1)).await;
|
|
|
+ // let msg = web_socket.read_message();
|
|
|
+ // match msg {
|
|
|
+ // Ok(Message::Text(text)) => {
|
|
|
+ // let get_time = chrono::Utc::now().timestamp_millis();
|
|
|
+ // if (get_time - ping_timeout) >= (1000 * 30) {
|
|
|
+ // trace!("30s 一次主动发送心跳包!");
|
|
|
+ // let _ = web_socket.write_message(Message::Ping(Vec::from("ping")));
|
|
|
+ // ping_timeout = get_time;
|
|
|
+ // }
|
|
|
+ //
|
|
|
+ // // let st_time = get_time_microsecond();
|
|
|
+ // let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
+ // // trace!("解析数据耗时：：{}", st_time - get_time_microsecond());
|
|
|
+ // res_data.time = get_time_microsecond();
|
|
|
+ // if res_data.code == "-200" {
|
|
|
+ // trace!("登陆：{:?}", res_data);
|
|
|
+ // } else if res_data.code == "-201" {
|
|
|
+ // trace!("订阅成功：{:?}", res_data.data);
|
|
|
+ // } else if res_data.code == "-202" {
|
|
|
+ // trace!("空数据不予返回：{:?}", res_data);
|
|
|
+ // } else {
|
|
|
+ // let sender = self.sender.clone();
|
|
|
+ // // let prev_time = Utc::now().timestamp_millis();
|
|
|
+ // tokio::spawn(async move {
|
|
|
+ // // info!("{:04} {}", Utc::now().timestamp_millis() - prev_time, res_data.channel);
|
|
|
+ // sender.send(res_data).await.unwrap();
|
|
|
+ // });
|
|
|
+ // tokio::spawn(async move {});
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // Ok(Message::Ping(s)) => {
|
|
|
+ // trace!( "Ping-响应--{:?}", String::from_utf8(s.clone()));
|
|
|
+ // let _ = web_socket.write_message(Message::Pong(Vec::from("pong")));
|
|
|
+ // trace!( "回应-pong---{:?}", String::from_utf8(s.clone()));
|
|
|
+ // }
|
|
|
+ // Ok(Message::Pong(s)) => {
|
|
|
+ // // trace!("Pong-响应--{:?}", String::from_utf8(s));
|
|
|
+ // trace!("Pong-响应--{:?}", String::from_utf8(s.clone()));
|
|
|
+ // }
|
|
|
+ // Ok(Message::Close(_)) => {
|
|
|
+ // // trace!("socket 关闭: ");
|
|
|
+ // trace!( "Close-响应");
|
|
|
+ // break;
|
|
|
+ // }
|
|
|
+ // Err(error) => {
|
|
|
+ // // trace!("Error receiving message: {}", error);
|
|
|
+ // error!( "Err-响应{}", error);
|
|
|
+ // break;
|
|
|
+ // }
|
|
|
+ // _ => {}
|
|
|
+ // }
|
|
|
+ //
|
|
|
+ // let bool_v1_v = bool_v1.load(Ordering::SeqCst);
|
|
|
+ // if !bool_v1_v {
|
|
|
+ // break;
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // web_socket.close(None).unwrap();
|
|
|
+ // }
|
|
|
+ //
|
|
|
+ // //非代理
|
|
|
+ // async fn subscription(&self, bool_v1: Arc<AtomicBool>, mut web_socket: WebSocket<AutoStream>,
|
|
|
+ // subscription: String)
|
|
|
+ // {
|
|
|
+ // info!("链接成功！开始数据读取");
|
|
|
+ // let label = self.label.clone();
|
|
|
+ // /*****登陆***/
|
|
|
+ // let login_str = self.log_in_to_str();
|
|
|
+ // if login_str != "" {
|
|
|
+ // let _ = web_socket.write_message(Message::Text(login_str));
|
|
|
+ // thread::sleep(Duration::from_secs(3));
|
|
|
+ // }
|
|
|
+ // /*****订阅***/
|
|
|
+ // if subscription.len() > 0 {
|
|
|
+ // trace!("订阅信息：{}",subscription);
|
|
|
+ // web_socket.write_message(Message::Text(subscription))
|
|
|
+ // .unwrap();
|
|
|
+ // }
|
|
|
+ // /*****消息溜***/
|
|
|
+ // let mut ping_timeout = chrono::Utc::now().timestamp_millis();
|
|
|
+ // loop {
|
|
|
+ // tokio::time::sleep(Duration::from_millis(1)).await;
|
|
|
+ // let msg = web_socket.read_message();
|
|
|
+ // match msg {
|
|
|
+ // Ok(Message::Text(text)) => {
|
|
|
+ // let get_time = chrono::Utc::now().timestamp_millis();
|
|
|
+ // if (get_time - ping_timeout) >= (1000 * 30) {
|
|
|
+ // trace!("30s 一次主动发送心跳包!");
|
|
|
+ // let _ = web_socket.write_message(Message::Ping(Vec::from("ping")));
|
|
|
+ // ping_timeout = get_time;
|
|
|
+ // }
|
|
|
+ //
|
|
|
+ // // let st_time = get_time_microsecond();
|
|
|
+ // let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
+ // // trace!("解析数据耗时：：{}", st_time - get_time_microsecond());
|
|
|
+ // res_data.time = get_time_microsecond();
|
|
|
+ // if res_data.code == "-200" {
|
|
|
+ // trace!("订阅成功：{:?}", res_data.data);
|
|
|
+ // } else if res_data.code == "-201" {
|
|
|
+ // trace!("登陆：{:?}", res_data);
|
|
|
+ // } else if res_data.code == "-202" {
|
|
|
+ // trace!("空数据不予返回：{:?}", res_data);
|
|
|
+ // } else {
|
|
|
+ // let sender = self.sender.clone();
|
|
|
+ // tokio::spawn(async move {
|
|
|
+ // sender.send(res_data).await.unwrap();
|
|
|
+ // });
|
|
|
+ // tokio::spawn(async move {});
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // Ok(Message::Ping(s)) => {
|
|
|
+ // trace!( "Ping-响应--{:?}", String::from_utf8(s.clone()));
|
|
|
+ // let _ = web_socket.write_message(Message::Pong(Vec::from("pong")));
|
|
|
+ // trace!( "回应-pong---{:?}", String::from_utf8(s.clone()));
|
|
|
+ // }
|
|
|
+ // Ok(Message::Pong(s)) => {
|
|
|
+ // // trace!("Pong-响应--{:?}", String::from_utf8(s));
|
|
|
+ // trace!("Pong-响应--{:?}", String::from_utf8(s.clone()));
|
|
|
+ // }
|
|
|
+ // Ok(Message::Close(_)) => {
|
|
|
+ // // trace!("socket 关闭: ");
|
|
|
+ // trace!( "Close-响应");
|
|
|
+ // break;
|
|
|
+ // }
|
|
|
+ // Err(error) => {
|
|
|
+ // // trace!("Error receiving message: {}", error);
|
|
|
+ // error!( "Err-响应{}", error);
|
|
|
+ // break;
|
|
|
+ // }
|
|
|
+ // _ => {}
|
|
|
+ // }
|
|
|
+ //
|
|
|
+ // let bool_v1_v = bool_v1.load(Ordering::SeqCst);
|
|
|
+ // if !bool_v1_v {
|
|
|
+ // break;
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // web_socket.close(None).unwrap();
|
|
|
+ // }
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************数据解析*****************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ //数据解析-Text
|
|
|
+ pub fn message_text(text: String) -> Option<ResponseData> {
|
|
|
+ let mut response_data = Self::ok_text(text);
|
|
|
+ response_data.time = get_time_microsecond();
|
|
|
+ match response_data.code.as_str() {
|
|
|
+ "200" => Option::from(response_data),
|
|
|
+ "-201" => {
|
|
|
+ trace!("订阅成功：{:?}", response_data);
|
|
|
+ None
|
|
|
+ }
|
|
|
+ "-200" => {
|
|
|
+ trace!("登录成功：{:?}", response_data);
|
|
|
+ None
|
|
|
+ }
|
|
|
+ "-202" => {
|
|
|
+ trace!("未知解析：{:?}", response_data);
|
|
|
+ None
|
|
|
+ }
|
|
|
+ _ => None
|
|
|
+ }
|
|
|
+ }
|
|
|
+ //数据解析-ping
|
|
|
+ pub fn message_ping(_pi: Vec<u8>) -> Option<ResponseData> {
|
|
|
+ trace!("服务器响应-ping");
|
|
|
+ return None;
|
|
|
+ }
|
|
|
+ //数据解析-pong
|
|
|
+ pub fn message_pong(_po: Vec<u8>) -> Option<ResponseData> {
|
|
|
+ trace!("服务器响应-pong");
|
|
|
+ return None;
|
|
|
+ }
|
|
|
+ //数据解析
|
|
|
+ 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 = "-202".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
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
hl