|
|
@@ -1,472 +1,490 @@
|
|
|
-// use std::collections::BTreeMap;
|
|
|
-// use std::net::{IpAddr, Ipv4Addr, SocketAddr};
|
|
|
-// use std::sync::Arc;
|
|
|
-// use std::sync::atomic::{AtomicBool, Ordering};
|
|
|
-// use std::time::Duration;
|
|
|
-//
|
|
|
-// use hex;
|
|
|
-// use hmac::{Hmac, Mac, NewMac};
|
|
|
-// use serde_json::{json, Value};
|
|
|
-// use sha2::Sha512;
|
|
|
-// use tokio::sync::mpsc::Sender;
|
|
|
-// use tracing::{error, info, trace};
|
|
|
-// use url::Url;
|
|
|
-//
|
|
|
-//
|
|
|
-// use crate::proxy;
|
|
|
-// use crate::proxy::ParsingDetail;
|
|
|
-// use crate::response_base::ResponseData;
|
|
|
-// use crate::utils::get_time_microsecond;
|
|
|
-//
|
|
|
-// pub enum GateWsType {
|
|
|
-// //频道类型
|
|
|
-// PublicAndPrivate(String),
|
|
|
-// }
|
|
|
-//
|
|
|
-//
|
|
|
-// #[derive(Clone)] //订阅枚举
|
|
|
-// pub enum GateSubscribeType {
|
|
|
-// PuFuturesOrderBook,
|
|
|
-// PuFuturesCandlesticks,
|
|
|
-// PuFuturesTrades,
|
|
|
-// //
|
|
|
-// PrFuturesOrders(String),
|
|
|
-// PrFuturesPositions(String),
|
|
|
-// PrFuturesBalances(String),
|
|
|
-//
|
|
|
-// }
|
|
|
-//
|
|
|
-// #[derive(Clone)]
|
|
|
-// pub struct GateSwapWs {
|
|
|
-// pub label: String,
|
|
|
-// request_url: String,
|
|
|
-// //实际ws 链接地址
|
|
|
-// proxy: ParsingDetail,
|
|
|
-// //账号信息
|
|
|
-// login_param: BTreeMap<String, String>,
|
|
|
-// //kuconis特殊参数
|
|
|
-// symbol_s: Vec<String>,
|
|
|
-// //订阅币对
|
|
|
-// subscribe_types: Vec<GateSubscribeType>,
|
|
|
-// //订阅信息
|
|
|
-// sender: Sender<ResponseData>, //数据通道
|
|
|
-// }
|
|
|
-//
|
|
|
-// impl GateSwapWs {
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************获取一个对象****************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// pub fn new(is_colo: bool,
|
|
|
-// login_param: BTreeMap<String, String>,
|
|
|
-// ws_type: GateWsType,
|
|
|
-// sender: Sender<ResponseData>,
|
|
|
-// ) -> GateSwapWs
|
|
|
-// {
|
|
|
-// return GateSwapWs::new_label("default-GateSwapWs".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: GateWsType,
|
|
|
-// sender: Sender<ResponseData>,
|
|
|
-// ) -> GateSwapWs
|
|
|
-// {
|
|
|
-// /*******走代理：根据环境变量配置来决定,如果配置了走代理，没有配置不走*******/
|
|
|
-// let parsing_detail = proxy::ParsingDetail::parsing_environment_variables();
|
|
|
-//
|
|
|
-//
|
|
|
-// let request_url = match ws_type {
|
|
|
-// GateWsType::PublicAndPrivate(name) => {
|
|
|
-// format!("wss://fx-ws.gateio.ws/v4/ws/{}", name.to_string())
|
|
|
-// }
|
|
|
-// };
|
|
|
-//
|
|
|
-// if is_colo {
|
|
|
-// info!("开启高速(未配置，走普通：{})通道",request_url);
|
|
|
-// } else {
|
|
|
-// info!("走普通通道：{}",request_url);
|
|
|
-// }
|
|
|
-// /*****返回结构体*******/
|
|
|
-// GateSwapWs {
|
|
|
-// label,
|
|
|
-// request_url,
|
|
|
-// proxy: parsing_detail,
|
|
|
-// login_param,
|
|
|
-// symbol_s: vec![],
|
|
|
-// subscribe_types: vec![],
|
|
|
-// sender,
|
|
|
-// }
|
|
|
-// }
|
|
|
-//
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************订阅函数********************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// //手动添加订阅信息
|
|
|
-// pub fn set_subscribe(&mut self, subscribe_types: Vec<GateSubscribeType>) {
|
|
|
-// 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("-", "_");
|
|
|
-// }
|
|
|
-// self.symbol_s = symbol_s;
|
|
|
-// self.run(bool_v1).await;
|
|
|
-// }
|
|
|
-//
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************工具函数********************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// //订阅枚举解析
|
|
|
-// pub fn enum_to_string(symbol: String, subscribe_type: GateSubscribeType, login_param: BTreeMap<String, String>) -> Value {
|
|
|
-// let time = chrono::Utc::now().timestamp();
|
|
|
-// let mut access_key = "".to_string();
|
|
|
-// if login_param.contains_key("access_key") {
|
|
|
-// access_key = login_param.get("access_key").unwrap().to_string();
|
|
|
-// }
|
|
|
-// let mut secret_key = "".to_string();
|
|
|
-// if login_param.contains_key("secret_key") {
|
|
|
-// secret_key = login_param.get("secret_key").unwrap().to_string();
|
|
|
-// }
|
|
|
-//
|
|
|
-// match subscribe_type {
|
|
|
-// GateSubscribeType::PuFuturesOrderBook => {
|
|
|
-// json!({
|
|
|
-// "time": time,
|
|
|
-// "channel": "futures.order_book",
|
|
|
-// "event": "subscribe",
|
|
|
-// "payload": [symbol, "20", "0"]
|
|
|
-// })
|
|
|
-// }
|
|
|
-// GateSubscribeType::PuFuturesCandlesticks => {
|
|
|
-// json!({
|
|
|
-// "time": time,
|
|
|
-// "channel": "futures.candlesticks",
|
|
|
-// "event": "subscribe",
|
|
|
-// "payload": ["1m", symbol]
|
|
|
-// })
|
|
|
-// }
|
|
|
-// GateSubscribeType::PrFuturesOrders(user_id) => {
|
|
|
-// json!({
|
|
|
-// "time": time,
|
|
|
-// "channel": "futures.orders",
|
|
|
-// "event": "subscribe",
|
|
|
-// "payload": [user_id, symbol],
|
|
|
-// "auth": {
|
|
|
-// "method": "api_key",
|
|
|
-// "KEY": access_key,
|
|
|
-// "SIGN":Self::sign(secret_key.to_string(),
|
|
|
-// "futures.orders".to_string(),
|
|
|
-// "subscribe".to_string(),
|
|
|
-// time.to_string())
|
|
|
-// }
|
|
|
-// })
|
|
|
-// }
|
|
|
-// GateSubscribeType::PrFuturesPositions(user_id) => {
|
|
|
-// json!({
|
|
|
-// "time": time,
|
|
|
-// "channel": "futures.positions",
|
|
|
-// "event": "subscribe",
|
|
|
-// "payload": [user_id, symbol],
|
|
|
-// "auth": {
|
|
|
-// "method": "api_key",
|
|
|
-// "KEY": access_key,
|
|
|
-// "SIGN":Self::sign(secret_key.to_string(),
|
|
|
-// "futures.positions".to_string(),
|
|
|
-// "subscribe".to_string(),
|
|
|
-// time.to_string())
|
|
|
-// }
|
|
|
-// })
|
|
|
-// }
|
|
|
-// GateSubscribeType::PrFuturesBalances(user_id) => {
|
|
|
-// json!({
|
|
|
-// "time": time,
|
|
|
-// "channel": "futures.balances",
|
|
|
-// "event": "subscribe",
|
|
|
-// "payload": [user_id],
|
|
|
-// "auth": {
|
|
|
-// "method": "api_key",
|
|
|
-// "KEY": access_key,
|
|
|
-// "SIGN":Self::sign(secret_key.to_string(),
|
|
|
-// "futures.balances".to_string(),
|
|
|
-// "subscribe".to_string(),
|
|
|
-// time.to_string())
|
|
|
-// }
|
|
|
-// })
|
|
|
-// }
|
|
|
-// GateSubscribeType::PuFuturesTrades => {
|
|
|
-// json!({
|
|
|
-// "time": time,
|
|
|
-// "channel": "futures.trades",
|
|
|
-// "event": "subscribe",
|
|
|
-// "payload": [symbol]
|
|
|
-// })
|
|
|
-// }
|
|
|
-// }
|
|
|
-// }
|
|
|
-// //组装订阅数据
|
|
|
-// pub fn get_subscription(&self) -> Vec<Value> {
|
|
|
-// 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(),
|
|
|
-// self.login_param.clone(),
|
|
|
-// );
|
|
|
-// args.push(ty_str);
|
|
|
-// }
|
|
|
-// }
|
|
|
-// args
|
|
|
-// }
|
|
|
-//
|
|
|
-// 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
|
|
|
-// }
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************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: Vec<Value>)
|
|
|
-// {
|
|
|
-// info!("走代理-链接成功！开始数据读取");
|
|
|
-// let label = self.label.clone();
|
|
|
-// /*****订阅***/
|
|
|
-// for sub in &subscription {
|
|
|
-// trace!("订阅信息：{}", sub.to_string());
|
|
|
-// web_socket.write_message(Message::Text(sub.to_string()))
|
|
|
-// .unwrap();
|
|
|
-// }
|
|
|
-// /*****消息溜***/
|
|
|
-// let mut start_ping = 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)) => {
|
|
|
-// // trace!("获取推送：{}",text.clone());
|
|
|
-// let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
-// res_data.time = get_time_microsecond();
|
|
|
-// if res_data.code == "-200" {
|
|
|
-// trace!("订阅成功：{:?}", res_data.data);
|
|
|
-// } else {
|
|
|
-// // self.sender.send(res_data).await.unwrap();
|
|
|
-// let sender = self.sender.clone();
|
|
|
-// tokio::spawn(async move {
|
|
|
-// sender.send(res_data).await.unwrap();
|
|
|
-// });
|
|
|
-// tokio::spawn(async move {});
|
|
|
-//
|
|
|
-// //主动ping 服务器
|
|
|
-// let new_time = chrono::Utc::now().timestamp_millis();
|
|
|
-// // let tr = format!("判断-ping {}--{},{},{}",new_time,start_ping,(new_time - start_ping), (new_time - start_ping) > 10000);
|
|
|
-// // trace!(tr);
|
|
|
-// if (new_time - start_ping) > 10000 {
|
|
|
-// let t = chrono::Utc::now().timestamp();
|
|
|
-// let ping_str = serde_json::json!({
|
|
|
-// "time" :t, "channel" : "futures.ping"
|
|
|
-// });
|
|
|
-// let _ = web_socket.write_message(Message::Ping(Vec::from(ping_str.to_string())));
|
|
|
-// start_ping = new_time;
|
|
|
-// }
|
|
|
-// }
|
|
|
-// }
|
|
|
-// 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: Vec<Value>)
|
|
|
-// {
|
|
|
-// info!("链接成功！开始数据读取");
|
|
|
-// let label = self.label.clone();
|
|
|
-// /*****订阅***/
|
|
|
-// for sub in &subscription {
|
|
|
-// trace!("订阅信息：{}", sub.to_string());
|
|
|
-// web_socket.write_message(Message::Text(sub.to_string()))
|
|
|
-// .unwrap();
|
|
|
-// }
|
|
|
-// /*****消息溜***/
|
|
|
-// let mut start_ping = 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)) => {
|
|
|
-// // trace!("获取推送：{}",text.clone());
|
|
|
-// let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
-// res_data.time = get_time_microsecond();
|
|
|
-// if res_data.code == "-200" {
|
|
|
-// trace!("订阅成功：{:?}", res_data.data);
|
|
|
-// } else {
|
|
|
-// // self.sender.send(res_data).await.unwrap();
|
|
|
-// let sender = self.sender.clone();
|
|
|
-// tokio::spawn(async move {
|
|
|
-// sender.send(res_data).await.unwrap();
|
|
|
-// });
|
|
|
-// tokio::spawn(async move {});
|
|
|
-//
|
|
|
-// //主动ping 服务器
|
|
|
-// let new_time = chrono::Utc::now().timestamp_millis();
|
|
|
-// // let tr = format!("判断-ping {}--{},{},{}",new_time,start_ping,(new_time - start_ping), (new_time - start_ping) > 10000);
|
|
|
-// // trace!(tr);
|
|
|
-// if (new_time - start_ping) > 10000 {
|
|
|
-// let t = chrono::Utc::now().timestamp();
|
|
|
-// let ping_str = serde_json::json!({
|
|
|
-// "time" :t, "channel" : "futures.ping"
|
|
|
-// });
|
|
|
-// let _ = web_socket.write_message(Message::Ping(Vec::from(ping_str.to_string())));
|
|
|
-// start_ping = new_time;
|
|
|
-// }
|
|
|
-// }
|
|
|
-// }
|
|
|
-// 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);
|
|
|
-// 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("error").is_some() {
|
|
|
-// let message = json_value["error"]["message"].as_str().unwrap().to_string();
|
|
|
-// let mes = message.trim_end_matches('\n');
|
|
|
-//
|
|
|
-// // let mes_json_value: serde_json::Value = serde_json::from_str(mes).unwrap();
|
|
|
-// // // trace!("错误信息：{}", mes_json_value.to_string());
|
|
|
-// res_data.code = json_value["error"]["code"].to_string();
|
|
|
-// res_data.message = mes.to_string();
|
|
|
-// } else if json_value["result"]["status"].as_str() == Option::from("success") {//订阅返回
|
|
|
-// res_data.code = "-200".to_string();
|
|
|
-// res_data.data = text;
|
|
|
-// } else {
|
|
|
-// res_data.channel = format!("{}", json_value["channel"].as_str().unwrap());
|
|
|
-// res_data.code = "200".to_string();
|
|
|
-// res_data.data = json_value["result"].to_string();
|
|
|
-// }
|
|
|
-// res_data
|
|
|
-// }
|
|
|
-// }
|
|
|
+use std::sync::Arc;
|
|
|
+use std::sync::atomic::AtomicBool;
|
|
|
+
|
|
|
+use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
|
|
|
+use hex;
|
|
|
+use hmac::{Hmac, Mac, NewMac};
|
|
|
+use serde_json::{json, Value};
|
|
|
+use sha2::Sha512;
|
|
|
+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 GateSwapWsType {
|
|
|
+ PublicAndPrivate(String),
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+//订阅频道
|
|
|
+#[derive(Clone)]
|
|
|
+pub enum GateSwapSubscribeType {
|
|
|
+ PuFuturesOrderBook,
|
|
|
+ PuFuturesCandlesticks,
|
|
|
+ PuFuturesTrades,
|
|
|
+
|
|
|
+ PrFuturesOrders(String),
|
|
|
+ PrFuturesPositions(String),
|
|
|
+ PrFuturesBalances(String),
|
|
|
+}
|
|
|
+
|
|
|
+//账号信息
|
|
|
+#[derive(Clone)]
|
|
|
+#[allow(dead_code)]
|
|
|
+pub struct GateSwapLogin {
|
|
|
+ pub api_key: String,
|
|
|
+ pub secret: String,
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+#[derive(Clone)]
|
|
|
+pub struct GateSwapWs {
|
|
|
+ //类型
|
|
|
+ label: String,
|
|
|
+ //地址
|
|
|
+ address_url: String,
|
|
|
+ //账号信息
|
|
|
+ login_param: Option<GateSwapLogin>,
|
|
|
+ //币对
|
|
|
+ symbol_s: Vec<String>,
|
|
|
+ //订阅
|
|
|
+ subscribe_types: Vec<GateSwapSubscribeType>,
|
|
|
+ //心跳间隔
|
|
|
+ heartbeat_time: u64,
|
|
|
+}
|
|
|
+
|
|
|
+impl GateSwapWs {
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************获取一个对象****************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ pub fn new(is_colo: bool, login_param: Option<GateSwapLogin>, ws_type: GateSwapWsType) -> GateSwapWs {
|
|
|
+ return GateSwapWs::new_label("default-GateSwapWs".to_string(), is_colo, login_param, ws_type);
|
|
|
+ }
|
|
|
+ pub fn new_label(label: String, is_colo: bool, login_param: Option<GateSwapLogin>, ws_type: GateSwapWsType) -> GateSwapWs
|
|
|
+ {
|
|
|
+ /*******公共频道-私有频道数据组装*/
|
|
|
+ let address_url = match ws_type {
|
|
|
+ GateSwapWsType::PublicAndPrivate(name) => {
|
|
|
+ format!("wss://fx-ws.gateio.ws/v4/ws/{}", name.to_string())
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ if is_colo {
|
|
|
+ info!("开启高速(未配置，走普通：{})通道",address_url);
|
|
|
+ } else {
|
|
|
+ info!("走普通通道：{}",address_url);
|
|
|
+ }
|
|
|
+ GateSwapWs {
|
|
|
+ label,
|
|
|
+ address_url,
|
|
|
+ login_param,
|
|
|
+ symbol_s: vec![],
|
|
|
+ subscribe_types: vec![],
|
|
|
+ heartbeat_time: 1000 * 20,
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************订阅函数********************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ //手动添加订阅信息
|
|
|
+ pub fn set_subscribe(&mut self, subscribe_types: Vec<GateSwapSubscribeType>) {
|
|
|
+ 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("-", "_");
|
|
|
+ }
|
|
|
+ self.symbol_s = b_array;
|
|
|
+ }
|
|
|
+ //频道是否需要登录
|
|
|
+ fn contains_pr(&self) -> bool {
|
|
|
+ for t in self.subscribe_types.clone() {
|
|
|
+ if match t {
|
|
|
+ GateSwapSubscribeType::PuFuturesOrderBook => false,
|
|
|
+ GateSwapSubscribeType::PuFuturesCandlesticks => false,
|
|
|
+ GateSwapSubscribeType::PuFuturesTrades => false,
|
|
|
+
|
|
|
+ GateSwapSubscribeType::PrFuturesOrders(_) => true,
|
|
|
+ GateSwapSubscribeType::PrFuturesPositions(_) => true,
|
|
|
+ GateSwapSubscribeType::PrFuturesBalances(_) => true,
|
|
|
+ } {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ false
|
|
|
+ }
|
|
|
+
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************工具函数********************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ //订阅枚举解析
|
|
|
+ pub fn enum_to_string(symbol: String, subscribe_type: GateSwapSubscribeType, login_param: Option<GateSwapLogin>) -> Value {
|
|
|
+ let time = chrono::Utc::now().timestamp();
|
|
|
+ let mut access_key = "".to_string();
|
|
|
+ let mut secret_key = "".to_string();
|
|
|
+ match login_param {
|
|
|
+ None => {}
|
|
|
+ Some(param) => {
|
|
|
+ access_key = param.api_key.clone();
|
|
|
+ secret_key = param.secret.clone();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ match subscribe_type {
|
|
|
+ GateSwapSubscribeType::PuFuturesOrderBook => {
|
|
|
+ json!({
|
|
|
+ "time": time,
|
|
|
+ "channel": "futures.order_book",
|
|
|
+ "event": "subscribe",
|
|
|
+ "payload": [symbol, "20", "0"]
|
|
|
+ })
|
|
|
+ }
|
|
|
+ GateSwapSubscribeType::PuFuturesCandlesticks => {
|
|
|
+ json!({
|
|
|
+ "time": time,
|
|
|
+ "channel": "futures.candlesticks",
|
|
|
+ "event": "subscribe",
|
|
|
+ "payload": ["1m", symbol]
|
|
|
+ })
|
|
|
+ }
|
|
|
+ GateSwapSubscribeType::PrFuturesOrders(user_id) => {
|
|
|
+ json!({
|
|
|
+ "time": time,
|
|
|
+ "channel": "futures.orders",
|
|
|
+ "event": "subscribe",
|
|
|
+ "payload": [user_id, symbol],
|
|
|
+ "auth": {
|
|
|
+ "method": "api_key",
|
|
|
+ "KEY": access_key,
|
|
|
+ "SIGN":Self::sign(secret_key.to_string(),
|
|
|
+ "futures.orders".to_string(),
|
|
|
+ "subscribe".to_string(),
|
|
|
+ time.to_string())
|
|
|
+ }
|
|
|
+ })
|
|
|
+ }
|
|
|
+ GateSwapSubscribeType::PrFuturesPositions(user_id) => {
|
|
|
+ json!({
|
|
|
+ "time": time,
|
|
|
+ "channel": "futures.positions",
|
|
|
+ "event": "subscribe",
|
|
|
+ "payload": [user_id, symbol],
|
|
|
+ "auth": {
|
|
|
+ "method": "api_key",
|
|
|
+ "KEY": access_key,
|
|
|
+ "SIGN":Self::sign(secret_key.to_string(),
|
|
|
+ "futures.positions".to_string(),
|
|
|
+ "subscribe".to_string(),
|
|
|
+ time.to_string())
|
|
|
+ }
|
|
|
+ })
|
|
|
+ }
|
|
|
+ GateSwapSubscribeType::PrFuturesBalances(user_id) => {
|
|
|
+ json!({
|
|
|
+ "time": time,
|
|
|
+ "channel": "futures.balances",
|
|
|
+ "event": "subscribe",
|
|
|
+ "payload": [user_id],
|
|
|
+ "auth": {
|
|
|
+ "method": "api_key",
|
|
|
+ "KEY": access_key,
|
|
|
+ "SIGN":Self::sign(secret_key.to_string(),
|
|
|
+ "futures.balances".to_string(),
|
|
|
+ "subscribe".to_string(),
|
|
|
+ time.to_string())
|
|
|
+ }
|
|
|
+ })
|
|
|
+ }
|
|
|
+ GateSwapSubscribeType::PuFuturesTrades => {
|
|
|
+ json!({
|
|
|
+ "time": time,
|
|
|
+ "channel": "futures.trades",
|
|
|
+ "event": "subscribe",
|
|
|
+ "payload": [symbol]
|
|
|
+ })
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ //订阅信息生成
|
|
|
+ pub fn get_subscription(&self) -> Vec<Value> {
|
|
|
+ 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(),
|
|
|
+ self.login_param.clone(),
|
|
|
+ );
|
|
|
+ args.push(ty_str);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ args
|
|
|
+ }
|
|
|
+ //生成签名
|
|
|
+ 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
|
|
|
+ }
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************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 {
|
|
|
+ //登录相关
|
|
|
+ }
|
|
|
+ for s in subscription {
|
|
|
+ subscribe_array.push(s.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: Vec<Value>)
|
|
|
+ // {
|
|
|
+ // info!("走代理-链接成功！开始数据读取");
|
|
|
+ // let label = self.label.clone();
|
|
|
+ // /*****订阅***/
|
|
|
+ // for sub in &subscription {
|
|
|
+ // trace!("订阅信息：{}", sub.to_string());
|
|
|
+ // web_socket.write_message(Message::Text(sub.to_string()))
|
|
|
+ // .unwrap();
|
|
|
+ // }
|
|
|
+ // /*****消息溜***/
|
|
|
+ // let mut start_ping = 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)) => {
|
|
|
+ // // trace!("获取推送：{}",text.clone());
|
|
|
+ // let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
+ // res_data.time = get_time_microsecond();
|
|
|
+ // if res_data.code == "-200" {
|
|
|
+ // trace!("订阅成功：{:?}", res_data.data);
|
|
|
+ // } else {
|
|
|
+ // // self.sender.send(res_data).await.unwrap();
|
|
|
+ // let sender = self.sender.clone();
|
|
|
+ // tokio::spawn(async move {
|
|
|
+ // sender.send(res_data).await.unwrap();
|
|
|
+ // });
|
|
|
+ // tokio::spawn(async move {});
|
|
|
+ //
|
|
|
+ // //主动ping 服务器
|
|
|
+ // let new_time = chrono::Utc::now().timestamp_millis();
|
|
|
+ // // let tr = format!("判断-ping {}--{},{},{}",new_time,start_ping,(new_time - start_ping), (new_time - start_ping) > 10000);
|
|
|
+ // // trace!(tr);
|
|
|
+ // if (new_time - start_ping) > 10000 {
|
|
|
+ // let t = chrono::Utc::now().timestamp();
|
|
|
+ // let ping_str = serde_json::json!({
|
|
|
+ // "time" :t, "channel" : "futures.ping"
|
|
|
+ // });
|
|
|
+ // let _ = web_socket.write_message(Message::Ping(Vec::from(ping_str.to_string())));
|
|
|
+ // start_ping = new_time;
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // 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: Vec<Value>)
|
|
|
+ // {
|
|
|
+ // info!("链接成功！开始数据读取");
|
|
|
+ // let label = self.label.clone();
|
|
|
+ // /*****订阅***/
|
|
|
+ // for sub in &subscription {
|
|
|
+ // trace!("订阅信息：{}", sub.to_string());
|
|
|
+ // web_socket.write_message(Message::Text(sub.to_string()))
|
|
|
+ // .unwrap();
|
|
|
+ // }
|
|
|
+ // /*****消息溜***/
|
|
|
+ // let mut start_ping = 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)) => {
|
|
|
+ // // trace!("获取推送：{}",text.clone());
|
|
|
+ // let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
+ // res_data.time = get_time_microsecond();
|
|
|
+ // if res_data.code == "-200" {
|
|
|
+ // trace!("订阅成功：{:?}", res_data.data);
|
|
|
+ // } else {
|
|
|
+ // // self.sender.send(res_data).await.unwrap();
|
|
|
+ // let sender = self.sender.clone();
|
|
|
+ // tokio::spawn(async move {
|
|
|
+ // sender.send(res_data).await.unwrap();
|
|
|
+ // });
|
|
|
+ // tokio::spawn(async move {});
|
|
|
+ //
|
|
|
+ // //主动ping 服务器
|
|
|
+ // let new_time = chrono::Utc::now().timestamp_millis();
|
|
|
+ // // let tr = format!("判断-ping {}--{},{},{}",new_time,start_ping,(new_time - start_ping), (new_time - start_ping) > 10000);
|
|
|
+ // // trace!(tr);
|
|
|
+ // if (new_time - start_ping) > 10000 {
|
|
|
+ // let t = chrono::Utc::now().timestamp();
|
|
|
+ // let ping_str = serde_json::json!({
|
|
|
+ // "time" :t, "channel" : "futures.ping"
|
|
|
+ // });
|
|
|
+ // let _ = web_socket.write_message(Message::Ping(Vec::from(ping_str.to_string())));
|
|
|
+ // start_ping = new_time;
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // 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
|
|
|
+ }
|
|
|
+ _ => 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);
|
|
|
+ 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();
|
|
|
+
|
|
|
+ if json_value.get("error").is_some() {
|
|
|
+ let message = json_value["error"]["message"].as_str().unwrap().to_string();
|
|
|
+ let mes = message.trim_end_matches('\n');
|
|
|
+
|
|
|
+ res_data.code = json_value["error"]["code"].to_string();
|
|
|
+ res_data.message = mes.to_string();
|
|
|
+ } else if json_value["result"]["status"].as_str() == Option::from("success") {//订阅返回
|
|
|
+ res_data.code = "-201".to_string();
|
|
|
+ res_data.data = text;
|
|
|
+ } else {
|
|
|
+ res_data.channel = format!("{}", json_value["channel"].as_str().unwrap());
|
|
|
+ res_data.code = "200".to_string();
|
|
|
+ res_data.data = json_value["result"].to_string();
|
|
|
+ }
|
|
|
+ res_data
|
|
|
+ }
|
|
|
+}
|
hl