|
|
@@ -1,516 +1,405 @@
|
|
|
-// use std::collections::{BTreeMap, HashSet};
|
|
|
-// use std::net::{IpAddr, Ipv4Addr, SocketAddr};
|
|
|
-// use std::sync::Arc;
|
|
|
-// use std::sync::atomic::{AtomicBool, Ordering};
|
|
|
-// use std::time::Duration;
|
|
|
-// use tokio::sync::mpsc::Sender;
|
|
|
-// use tracing::{error, info, trace};
|
|
|
-// use crate::{proxy};
|
|
|
-// use url::Url;
|
|
|
-// use crate::kucoin_swap_rest::KucoinSwapRest;
|
|
|
-// use crate::proxy::ParsingDetail;
|
|
|
-// use crate::response_base::ResponseData;
|
|
|
-// use crate::utils::get_time_microsecond;
|
|
|
-//
|
|
|
-//
|
|
|
-// pub enum KucoinWsType {
|
|
|
-// Public,
|
|
|
-// Private,
|
|
|
-// }
|
|
|
-//
|
|
|
-// #[derive(Debug)]
|
|
|
-// #[derive(Clone)]
|
|
|
-// pub struct KucoinWsParam {
|
|
|
-// pub token: String,
|
|
|
-// pub ws_url: String,
|
|
|
-// pub ws_ping_interval: i64,
|
|
|
-// pub ws_ping_timeout: i64,
|
|
|
-// }
|
|
|
-//
|
|
|
-// #[derive(Clone)] //订阅枚举
|
|
|
-// pub enum KucoinSubscribeType {
|
|
|
-// PuContractMarketLevel2Depth50,
|
|
|
-// PuContractMarketExecution,
|
|
|
-// PuContractMarkettickerV2,
|
|
|
-//
|
|
|
-// PrContractAccountWallet,
|
|
|
-// PrContractPosition,
|
|
|
-// PrContractMarketTradeOrdersSys,
|
|
|
-// PrContractMarketTradeOrders,
|
|
|
-// }
|
|
|
-//
|
|
|
-// #[derive(Clone)]
|
|
|
-// pub struct KucoinSwapWs {
|
|
|
-// pub label: String,
|
|
|
-// request_url: String,
|
|
|
-// //实际ws 链接地址
|
|
|
-// proxy: ParsingDetail,
|
|
|
-// //代理信息
|
|
|
-// // login_param: BTreeMap<String, String>,
|
|
|
-// //登陆数据
|
|
|
-// ws_param: KucoinWsParam,
|
|
|
-// //kuconis特殊参数
|
|
|
-// symbol_s: Vec<String>,
|
|
|
-// //订阅币对
|
|
|
-// subscribe_types: Vec<KucoinSubscribeType>,
|
|
|
-// //订阅信息
|
|
|
-// sender: Sender<ResponseData>, //数据通道
|
|
|
-// }
|
|
|
-//
|
|
|
-// impl KucoinSwapWs {
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************获取一个对象****************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// pub async fn new(is_colo: bool,
|
|
|
-// login_param: BTreeMap<String, String>,
|
|
|
-// ws_type: KucoinWsType,
|
|
|
-// sender: Sender<ResponseData>,
|
|
|
-// ) -> KucoinSwapWs {
|
|
|
-// return KucoinSwapWs::new_label("default-KucoinSwapWs".to_string(), is_colo, login_param, ws_type, sender).await;
|
|
|
-// }
|
|
|
-// pub async fn new_label(label: String, _is_colo: bool,
|
|
|
-// login_param: BTreeMap<String, String>,
|
|
|
-// ws_type: KucoinWsType,
|
|
|
-// sender: Sender<ResponseData>,
|
|
|
-// ) -> KucoinSwapWs
|
|
|
-// {
|
|
|
-// /*******走代理:根据环境变量配置来决定,如果配置了走代理,没有配置不走*******/
|
|
|
-// let parsing_detail = proxy::ParsingDetail::parsing_environment_variables();
|
|
|
-//
|
|
|
-// /*******公共频道-私有频道数据组装*/
|
|
|
-// let mut ws_param = KucoinWsParam {
|
|
|
-// token: "".to_string(),
|
|
|
-// ws_url: "".to_string(),
|
|
|
-// ws_ping_interval: 0,
|
|
|
-// ws_ping_timeout: 0,
|
|
|
-// };
|
|
|
-// let res_data = KucoinSwapWs::get_rul_token(ws_type, login_param.clone()).await;
|
|
|
-// match res_data {
|
|
|
-// Ok(param) => {
|
|
|
-// ws_param = param
|
|
|
-// }
|
|
|
-// Err(error) => {
|
|
|
-// error!("-链接地址等参数错误:{:?}", error)
|
|
|
-// }
|
|
|
-// }
|
|
|
-//
|
|
|
-//
|
|
|
-// /*****返回结构体*******/
|
|
|
-// KucoinSwapWs {
|
|
|
-// label,
|
|
|
-// request_url: "".to_string(),
|
|
|
-// proxy: parsing_detail,
|
|
|
-// // login_param,
|
|
|
-// ws_param,
|
|
|
-// symbol_s: vec![],
|
|
|
-// subscribe_types: vec![],
|
|
|
-// sender,
|
|
|
-// }
|
|
|
-// }
|
|
|
-//
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************订阅函数********************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// //手动添加订阅信息
|
|
|
-// pub fn set_subscribe(&mut self, subscribe_types: Vec<KucoinSubscribeType>) {
|
|
|
-// self.subscribe_types.extend(subscribe_types);
|
|
|
-// }
|
|
|
-// //根据当前类型获取对应的频道 地址 与 token
|
|
|
-// async fn get_rul_token(ws_type: KucoinWsType, login_param: BTreeMap<String, String>) -> Result<KucoinWsParam, reqwest::Error> {
|
|
|
-// let mut kucoin_exc = KucoinSwapRest::new(false, login_param.clone());
|
|
|
-// let res_data = match ws_type {
|
|
|
-// KucoinWsType::Public => {
|
|
|
-// kucoin_exc.get_public_token().await
|
|
|
-// }
|
|
|
-// KucoinWsType::Private => {
|
|
|
-// kucoin_exc.get_private_token().await
|
|
|
-// }
|
|
|
-// };
|
|
|
-//
|
|
|
-// trace!("kucoin-swap-rest 获取ws连接地址:{:?}",res_data);
|
|
|
-//
|
|
|
-// if res_data.code == "200" {
|
|
|
-// let mut ws_url = "".to_string();
|
|
|
-// let mut ws_token = "".to_string();
|
|
|
-// let mut ws_ping_interval: i64 = 0;
|
|
|
-// let mut ws_ping_timeout: i64 = 0;
|
|
|
-//
|
|
|
-//
|
|
|
-// //数据解析
|
|
|
-// let parsed_json: serde_json::Value = serde_json::from_str(res_data.data.as_str()).unwrap();
|
|
|
-// if let Some(value) = parsed_json.get("token") {
|
|
|
-// let formatted_value = match value {
|
|
|
-// serde_json::Value::String(s) => s.clone(),
|
|
|
-// _ => value.to_string()
|
|
|
-// };
|
|
|
-// ws_token = format!("{}", formatted_value);
|
|
|
-// }
|
|
|
-// if let Some(endpoint) = parsed_json["instanceServers"][0]["endpoint"].as_str() {
|
|
|
-// ws_url = format!("{}", endpoint);
|
|
|
-// }
|
|
|
-// if let Some(ping_interval) = parsed_json["instanceServers"][0]["pingInterval"].as_i64() {
|
|
|
-// ws_ping_interval = ping_interval;
|
|
|
-// }
|
|
|
-// if let Some(ping_timeout) = parsed_json["instanceServers"][0]["pingTimeout"].as_i64() {
|
|
|
-// ws_ping_timeout = ping_timeout;
|
|
|
-// }
|
|
|
-//
|
|
|
-//
|
|
|
-// Ok(KucoinWsParam { ws_url, token: ws_token, ws_ping_interval, ws_ping_timeout })
|
|
|
-// } else {
|
|
|
-// error!("公共/私有-频道获取失败:{:?}", res_data);
|
|
|
-// panic!("公共/私有-频道获取失败:{:?}", res_data);
|
|
|
-// }
|
|
|
-// }
|
|
|
-// //自定义
|
|
|
-// pub async fn custom_subscribe(&mut self, bool_v1: Arc<AtomicBool>, b_array: Vec<String>)
|
|
|
-// {
|
|
|
-// self.symbol_s = b_array.clone();
|
|
|
-// self.request_url = format!("{}?token={}", self.ws_param.ws_url, self.ws_param.token);
|
|
|
-// info!("走普通通道:{}", self.request_url);
|
|
|
-// self.run(bool_v1).await;
|
|
|
-// }
|
|
|
-//
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************工具函数********************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// //订阅枚举解析
|
|
|
-// pub fn enum_to_string(symbol: String, subscribe_type: KucoinSubscribeType) -> serde_json::Value {
|
|
|
-// match subscribe_type {
|
|
|
-// KucoinSubscribeType::PuContractMarketLevel2Depth50 => {//level2
|
|
|
-// serde_json::json!({
|
|
|
-// "topic": format!("/contractMarket/level2Depth50:{}", symbol),
|
|
|
-// "type": "subscribe",
|
|
|
-// "response": true
|
|
|
-// })
|
|
|
-// }
|
|
|
-// KucoinSubscribeType::PuContractMarketExecution => {//match
|
|
|
-// serde_json::json!({
|
|
|
-// "topic": format!("/contractMarket/execution:{}", symbol),
|
|
|
-// "type": "subscribe",
|
|
|
-// "response": true
|
|
|
-// })
|
|
|
-// }
|
|
|
-// KucoinSubscribeType::PuContractMarkettickerV2 => {//tickerV2
|
|
|
-// serde_json::json!({
|
|
|
-// "topic": format!("/contractMarket/tickerV2:{}", symbol),
|
|
|
-// "type": "subscribe",
|
|
|
-// "response": true
|
|
|
-// })
|
|
|
-// }
|
|
|
-// KucoinSubscribeType::PrContractAccountWallet => {//orderMargin.change
|
|
|
-// serde_json::json!({
|
|
|
-// "type": "subscribe",
|
|
|
-// "topic": "/contractAccount/wallet",
|
|
|
-// "privateChannel":true,
|
|
|
-// "response":true,
|
|
|
-// })
|
|
|
-// }
|
|
|
-// KucoinSubscribeType::PrContractPosition => {//position.change
|
|
|
-// serde_json::json!({
|
|
|
-// "type": "subscribe",
|
|
|
-// "topic": format!("/contract/position:{}", symbol),
|
|
|
-// "privateChannel":true,
|
|
|
-// "response":true,
|
|
|
-// })
|
|
|
-// }
|
|
|
-// KucoinSubscribeType::PrContractMarketTradeOrdersSys => {//orderChange
|
|
|
-// serde_json::json!({
|
|
|
-// "type": "subscribe",
|
|
|
-// "topic": format!("/contractMarket/tradeOrders"),
|
|
|
-// "privateChannel":true,
|
|
|
-// "response":true,
|
|
|
-// })
|
|
|
-// }
|
|
|
-// KucoinSubscribeType::PrContractMarketTradeOrders => {//symbolOrderChange
|
|
|
-// serde_json::json!({
|
|
|
-// "type": "subscribe",
|
|
|
-// "topic": format!("/contractMarket/tradeOrders:{}", symbol),
|
|
|
-// "privateChannel":true,
|
|
|
-// "response":true,
|
|
|
-// })
|
|
|
-// }
|
|
|
-// }
|
|
|
-// }
|
|
|
-// //组装订阅数据
|
|
|
-// pub fn get_subscription(&self) -> Vec<String> {
|
|
|
-// let mut array = 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());
|
|
|
-// array.push(ty_str.to_string());
|
|
|
-// }
|
|
|
-// }
|
|
|
-// array
|
|
|
-// }
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// /*****************************************socket基本*****************************************************/
|
|
|
-// /*******************************************************************************************************/
|
|
|
-// async fn run(&self, bool_v1: Arc<AtomicBool>)
|
|
|
-// {
|
|
|
-// //订阅信息组装
|
|
|
-// let subscription = self.get_subscription();
|
|
|
-// let subscription: Vec<String> = subscription.into_iter().collect::<HashSet<String>>().into_iter().collect();
|
|
|
-// loop {
|
|
|
-// tokio::time::sleep(Duration::from_millis(5000)).await;
|
|
|
-// trace!("要连接咯~~!!{}", self.request_url);
|
|
|
-// //币安-登陆流程-rest请求获取k然后拿到 key 拼接地址
|
|
|
-// // if self.is_login { //暂时没看到有订阅的频道需要登陆 所以暂时不做
|
|
|
-// // }
|
|
|
-//
|
|
|
-// 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;
|
|
|
-// let _ = 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 {
|
|
|
-// let _ = 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<String>)
|
|
|
-// {
|
|
|
-// info!("走代理-链接成功!开始数据读取");
|
|
|
-// let label = self.label.clone();
|
|
|
-// /*****消息溜***/
|
|
|
-// let mut ping_interval = chrono::Utc::now().timestamp_millis();
|
|
|
-// 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();
|
|
|
-// // trace!("--心跳-{}-{}-{}-{}-{}", get_time, ping_interval
|
|
|
-// // , (get_time - ping_interval), self.ws_param.ws_ping_interval, (get_time - ping_interval) >= self.ws_param.ws_ping_interval);
|
|
|
-// if (get_time - ping_interval) >= self.ws_param.ws_ping_interval {
|
|
|
-// web_socket.write_message(Message::Ping(Vec::from("ping")))
|
|
|
-// .unwrap();
|
|
|
-// trace!("--发送心跳-ping");
|
|
|
-// ping_interval = get_time;
|
|
|
-// ping_timeout = get_time;
|
|
|
-// } else if (get_time - ping_timeout) > (self.ws_param.ws_ping_timeout + self.ws_param.ws_ping_interval) {
|
|
|
-// //心跳超时-发送心跳之后 一定时间没有响应
|
|
|
-// trace!("--心跳相应超时-重连");
|
|
|
-// break;
|
|
|
-// }
|
|
|
-//
|
|
|
-// trace!("获取推送:{}",text.clone());
|
|
|
-// // trace!(stdout, "Text-响应--{:?}", text.clone()).unwrap();
|
|
|
-// let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
-// res_data.time = get_time_microsecond();
|
|
|
-// // trace!("获取推送:{:?}", res_data);
|
|
|
-// if res_data.code == "-200" {//表示链接成功
|
|
|
-// for sub in &subscription {
|
|
|
-// trace!("--发起订阅:{:?}", sub);
|
|
|
-// web_socket.write_message(Message::Text(sub.parse().unwrap()))
|
|
|
-// .unwrap();
|
|
|
-// }
|
|
|
-// } 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()));
|
|
|
-// ping_timeout = chrono::Utc::now().timestamp_millis();
|
|
|
-// }
|
|
|
-// Ok(Message::Close(_)) => {
|
|
|
-// // trace!("socket 关闭: ");
|
|
|
-// trace!( "Close-响应");
|
|
|
-// }
|
|
|
-// 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<String>)
|
|
|
-// {
|
|
|
-// info!("链接成功!开始数据读取");
|
|
|
-// let label = self.label.clone();
|
|
|
-// /*****消息溜***/
|
|
|
-// let mut ping_interval = chrono::Utc::now().timestamp_millis();
|
|
|
-// 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();
|
|
|
-// // trace!("--心跳-{}-{}-{}-{}-{}", get_time, ping_interval
|
|
|
-// // , (get_time - ping_interval), self.ws_param.ws_ping_interval, (get_time - ping_interval) >= self.ws_param.ws_ping_interval);
|
|
|
-// if (get_time - ping_interval) >= self.ws_param.ws_ping_interval {
|
|
|
-// web_socket.write_message(Message::Ping(Vec::from("ping")))
|
|
|
-// .unwrap();
|
|
|
-// trace!("--发送心跳-ping");
|
|
|
-// ping_interval = get_time;
|
|
|
-// ping_timeout = get_time;
|
|
|
-// } else if (get_time - ping_timeout) > (self.ws_param.ws_ping_timeout + self.ws_param.ws_ping_interval) {
|
|
|
-// //心跳超时-发送心跳之后 一定时间没有响应
|
|
|
-// trace!("--心跳相应超时-重连");
|
|
|
-// break;
|
|
|
-// }
|
|
|
-//
|
|
|
-// trace!("获取推送:{}",text.clone());
|
|
|
-// // trace!(stdout, "Text-响应--{:?}", text.clone()).unwrap();
|
|
|
-// let mut res_data = Self::ok_text(label.to_string(), text);
|
|
|
-// res_data.time = get_time_microsecond();
|
|
|
-// // trace!("获取推送:{:?}", res_data);
|
|
|
-// if res_data.code == "-200" {//表示链接成功
|
|
|
-// for sub in &subscription {
|
|
|
-// trace!("--发起订阅:{:?}", sub);
|
|
|
-// web_socket.write_message(Message::Text(sub.parse().unwrap()))
|
|
|
-// .unwrap();
|
|
|
-// }
|
|
|
-// } 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()));
|
|
|
-// ping_timeout = chrono::Utc::now().timestamp_millis();
|
|
|
-// }
|
|
|
-// Ok(Message::Close(_)) => {
|
|
|
-// // trace!("socket 关闭: ");
|
|
|
-// trace!( "Close-响应");
|
|
|
-// }
|
|
|
-// 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
|
|
|
-// {
|
|
|
-// let mut res_data = ResponseData::new(lable, "200".to_string(), "success".to_string(), "".to_string());
|
|
|
-// let json_value: serde_json::Value = serde_json::from_str(&text).unwrap();
|
|
|
-// //订阅 相应
|
|
|
-// if json_value["type"].as_str() == Option::from("welcome") {
|
|
|
-// //链接成功
|
|
|
-// res_data.code = "-200".to_string();
|
|
|
-// res_data.message = "链接成功,主动发起订阅".to_string();
|
|
|
-// trace!("链接成功,主动发起订阅:");
|
|
|
-// } else if json_value["type"].as_str() == Option::from("ack") {
|
|
|
-// res_data.code = "-201".to_string();
|
|
|
-// res_data.message = "订阅成功".to_string();
|
|
|
-// } else if json_value["type"].as_str() == Option::from("error") {
|
|
|
-// res_data.code = format!("{}", json_value["code"]);
|
|
|
-// res_data.message = format!("{}", json_value["data"].as_str().unwrap());
|
|
|
-// } else if json_value.get("topic").is_some() {
|
|
|
-// res_data.channel = format!("{}", json_value["subject"].as_str().unwrap());
|
|
|
-//
|
|
|
-// if json_value["topic"].as_str() == Option::from("/contractAccount/wallet") {
|
|
|
-// res_data.code = "-202".to_string();
|
|
|
-// if json_value["subject"].as_str() == Option::from("availableBalance.change") {
|
|
|
-// res_data.code = "200".to_string();
|
|
|
-// res_data.data = json_value["data"].to_string();
|
|
|
-// } else {}
|
|
|
-// } else {
|
|
|
-// res_data.data = json_value["data"].to_string();
|
|
|
-// }
|
|
|
-// } else {
|
|
|
-// res_data.code = "-1".to_string();
|
|
|
-// res_data.message = "未知解析".to_string();
|
|
|
-// }
|
|
|
-// res_data
|
|
|
-// }
|
|
|
-// }
|
|
|
+use std::collections::BTreeMap;
|
|
|
+use std::sync::Arc;
|
|
|
+use std::sync::atomic::AtomicBool;
|
|
|
+
|
|
|
+use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
|
|
|
+use tokio::sync::Mutex;
|
|
|
+use tokio_tungstenite::tungstenite::{Error, Message};
|
|
|
+use tracing::{error, info, trace};
|
|
|
+
|
|
|
+use crate::kucoin_swap_rest::KucoinSwapRest;
|
|
|
+use crate::response_base::ResponseData;
|
|
|
+use crate::socket_tool::{AbstractWsMode, HeartbeatType};
|
|
|
+use crate::utils::get_time_microsecond;
|
|
|
+
|
|
|
+//类型
|
|
|
+pub enum KucoinSwapWsType {
|
|
|
+ Public,
|
|
|
+ Private,
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+#[derive(Debug)]
|
|
|
+#[derive(Clone)]
|
|
|
+pub struct KucoinSwapWsParam {
|
|
|
+ pub token: String,
|
|
|
+ pub ws_url: String,
|
|
|
+ pub ws_ping_interval: i64,
|
|
|
+ pub ws_ping_timeout: i64,
|
|
|
+ pub is_ok_subscribe: bool,
|
|
|
+}
|
|
|
+
|
|
|
+//订阅频道
|
|
|
+#[derive(Clone)]
|
|
|
+pub enum KucoinSwapSubscribeType {
|
|
|
+ PuContractMarketLevel2Depth50,
|
|
|
+ PuContractMarketExecution,
|
|
|
+ PuContractMarkettickerV2,
|
|
|
+
|
|
|
+ PrContractAccountWallet,
|
|
|
+ PrContractPosition,
|
|
|
+ PrContractMarketTradeOrdersSys,
|
|
|
+ PrContractMarketTradeOrders,
|
|
|
+}
|
|
|
+
|
|
|
+//账号信息
|
|
|
+#[derive(Clone, Debug)]
|
|
|
+pub struct KucoinSwapLogin {
|
|
|
+ pub access_key: String,
|
|
|
+ pub secret_key: String,
|
|
|
+ pub pass_key: String,
|
|
|
+}
|
|
|
+
|
|
|
+#[derive(Clone)]
|
|
|
+#[allow(dead_code)]
|
|
|
+pub struct KucoinSwapWs {
|
|
|
+ //类型
|
|
|
+ label: String,
|
|
|
+ //地址
|
|
|
+ address_url: String,
|
|
|
+ //账号
|
|
|
+ login_param: Option<KucoinSwapLogin>,
|
|
|
+ //登陆数据
|
|
|
+ ws_param: KucoinSwapWsParam,
|
|
|
+ //币对
|
|
|
+ symbol_s: Vec<String>,
|
|
|
+ //订阅
|
|
|
+ subscribe_types: Vec<KucoinSwapSubscribeType>,
|
|
|
+ //心跳间隔
|
|
|
+ heartbeat_time: u64,
|
|
|
+}
|
|
|
+
|
|
|
+impl KucoinSwapWs {
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************获取一个对象****************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ pub async fn new(is_colo: bool, login_param: Option<KucoinSwapLogin>, ws_type: KucoinSwapWsType) -> KucoinSwapWs {
|
|
|
+ return Self::new_label("default-KucoinSwapWs".to_string(), is_colo, login_param, ws_type).await;
|
|
|
+ }
|
|
|
+ pub async fn new_label(label: String, is_colo: bool, login_param: Option<KucoinSwapLogin>, ws_type: KucoinSwapWsType) -> KucoinSwapWs {
|
|
|
+ /*******公共频道-私有频道数据组装*/
|
|
|
+ let mut ws_param = KucoinSwapWsParam {
|
|
|
+ token: "".to_string(),
|
|
|
+ ws_url: "".to_string(),
|
|
|
+ ws_ping_interval: 0,
|
|
|
+ ws_ping_timeout: 0,
|
|
|
+ is_ok_subscribe: false,
|
|
|
+ };
|
|
|
+
|
|
|
+ /*******公共频道-私有频道数据组装*/
|
|
|
+ let res_data = Self::get_rul_token(ws_type, login_param.clone()).await;
|
|
|
+ let address_url = match res_data {
|
|
|
+ Ok(param) => {
|
|
|
+ ws_param = param;
|
|
|
+ format!("{}?token={}", ws_param.ws_url, ws_param.token)
|
|
|
+ }
|
|
|
+ Err(error) => {
|
|
|
+ error!("-链接地址等参数错误:{:?}", error);
|
|
|
+ "".to_string()
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ if is_colo {
|
|
|
+ info!("开启高速(未配置,走普通:{})通道",address_url);
|
|
|
+ } else {
|
|
|
+ info!("走普通通道:{}",address_url);
|
|
|
+ }
|
|
|
+
|
|
|
+ KucoinSwapWs {
|
|
|
+ label,
|
|
|
+ address_url,
|
|
|
+ login_param,
|
|
|
+ ws_param,
|
|
|
+ symbol_s: vec![],
|
|
|
+ subscribe_types: vec![],
|
|
|
+ heartbeat_time: 1000 * 18,
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************订阅函数********************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ //根据当前类型获取对应的频道 地址 与 token
|
|
|
+ async fn get_rul_token(ws_type: KucoinSwapWsType, login_param: Option<KucoinSwapLogin>) -> Result<KucoinSwapWsParam, reqwest::Error> {
|
|
|
+ let mut kucoin_exc = KucoinSwapRest::new(false, match login_param {
|
|
|
+ None => {
|
|
|
+ let btree_map: BTreeMap<String, String> = BTreeMap::new();
|
|
|
+ btree_map
|
|
|
+ }
|
|
|
+ Some(d) => {
|
|
|
+ let mut btree_map: BTreeMap<String, String> = BTreeMap::new();
|
|
|
+ btree_map.insert("access_key".to_string(), d.access_key);
|
|
|
+ btree_map.insert("secret_key".to_string(), d.secret_key);
|
|
|
+ btree_map.insert("pass_key".to_string(), d.pass_key);
|
|
|
+ btree_map
|
|
|
+ }
|
|
|
+ });
|
|
|
+
|
|
|
+
|
|
|
+ let res_data = match ws_type {
|
|
|
+ KucoinSwapWsType::Public => {
|
|
|
+ kucoin_exc.get_public_token().await
|
|
|
+ }
|
|
|
+ KucoinSwapWsType::Private => {
|
|
|
+ kucoin_exc.get_private_token().await
|
|
|
+ }
|
|
|
+ };
|
|
|
+
|
|
|
+ trace!("kucoin-swap-rest 获取ws连接地址:{:?}",res_data);
|
|
|
+
|
|
|
+ if res_data.code == "200" {
|
|
|
+ let mut ws_url = "".to_string();
|
|
|
+ let mut ws_token = "".to_string();
|
|
|
+ let mut ws_ping_interval: i64 = 0;
|
|
|
+ let mut ws_ping_timeout: i64 = 0;
|
|
|
+
|
|
|
+
|
|
|
+ //数据解析
|
|
|
+ let parsed_json: serde_json::Value = serde_json::from_str(res_data.data.as_str()).unwrap();
|
|
|
+ if let Some(value) = parsed_json.get("token") {
|
|
|
+ let formatted_value = match value {
|
|
|
+ serde_json::Value::String(s) => s.clone(),
|
|
|
+ _ => value.to_string()
|
|
|
+ };
|
|
|
+ ws_token = format!("{}", formatted_value);
|
|
|
+ }
|
|
|
+ if let Some(endpoint) = parsed_json["instanceServers"][0]["endpoint"].as_str() {
|
|
|
+ ws_url = format!("{}", endpoint);
|
|
|
+ }
|
|
|
+ if let Some(ping_interval) = parsed_json["instanceServers"][0]["pingInterval"].as_i64() {
|
|
|
+ ws_ping_interval = ping_interval;
|
|
|
+ }
|
|
|
+ if let Some(ping_timeout) = parsed_json["instanceServers"][0]["pingTimeout"].as_i64() {
|
|
|
+ ws_ping_timeout = ping_timeout;
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ Ok(KucoinSwapWsParam { ws_url, token: ws_token, ws_ping_interval, ws_ping_timeout, is_ok_subscribe: false })
|
|
|
+ } else {
|
|
|
+ error!("公共/私有-频道获取失败:{:?}", res_data);
|
|
|
+ panic!("公共/私有-频道获取失败:{:?}", res_data);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ //手动添加订阅信息
|
|
|
+ pub fn set_subscribe(&mut self, subscribe_types: Vec<KucoinSwapSubscribeType>) {
|
|
|
+ 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 {
|
|
|
+ KucoinSwapSubscribeType::PuContractMarketLevel2Depth50 => false,
|
|
|
+ KucoinSwapSubscribeType::PuContractMarketExecution => false,
|
|
|
+ KucoinSwapSubscribeType::PuContractMarkettickerV2 => false,
|
|
|
+
|
|
|
+ KucoinSwapSubscribeType::PrContractAccountWallet => true,
|
|
|
+ KucoinSwapSubscribeType::PrContractPosition => true,
|
|
|
+ KucoinSwapSubscribeType::PrContractMarketTradeOrdersSys => true,
|
|
|
+ KucoinSwapSubscribeType::PrContractMarketTradeOrders => true,
|
|
|
+ } {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ false
|
|
|
+ }
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************工具函数********************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ //订阅枚举解析
|
|
|
+ pub fn enum_to_string(symbol: String, subscribe_type: KucoinSwapSubscribeType) -> serde_json::Value {
|
|
|
+ match subscribe_type {
|
|
|
+ KucoinSwapSubscribeType::PuContractMarketLevel2Depth50 => {//level2
|
|
|
+ serde_json::json!({
|
|
|
+ "topic": format!("/contractMarket/level2Depth50:{}", symbol),
|
|
|
+ "type": "subscribe",
|
|
|
+ "response": true
|
|
|
+ })
|
|
|
+ }
|
|
|
+ KucoinSwapSubscribeType::PuContractMarketExecution => {//match
|
|
|
+ serde_json::json!({
|
|
|
+ "topic": format!("/contractMarket/execution:{}", symbol),
|
|
|
+ "type": "subscribe",
|
|
|
+ "response": true
|
|
|
+ })
|
|
|
+ }
|
|
|
+ KucoinSwapSubscribeType::PuContractMarkettickerV2 => {//tickerV2
|
|
|
+ serde_json::json!({
|
|
|
+ "topic": format!("/contractMarket/tickerV2:{}", symbol),
|
|
|
+ "type": "subscribe",
|
|
|
+ "response": true
|
|
|
+ })
|
|
|
+ }
|
|
|
+ KucoinSwapSubscribeType::PrContractAccountWallet => {//orderMargin.change
|
|
|
+ serde_json::json!({
|
|
|
+ "type": "subscribe",
|
|
|
+ "topic": "/contractAccount/wallet",
|
|
|
+ "privateChannel":true,
|
|
|
+ "response":true,
|
|
|
+ })
|
|
|
+ }
|
|
|
+ KucoinSwapSubscribeType::PrContractPosition => {//position.change
|
|
|
+ serde_json::json!({
|
|
|
+ "type": "subscribe",
|
|
|
+ "topic": format!("/contract/position:{}", symbol),
|
|
|
+ "privateChannel":true,
|
|
|
+ "response":true,
|
|
|
+ })
|
|
|
+ }
|
|
|
+ KucoinSwapSubscribeType::PrContractMarketTradeOrdersSys => {//orderChange
|
|
|
+ serde_json::json!({
|
|
|
+ "type": "subscribe",
|
|
|
+ "topic": format!("/contractMarket/tradeOrders"),
|
|
|
+ "privateChannel":true,
|
|
|
+ "response":true,
|
|
|
+ })
|
|
|
+ }
|
|
|
+ KucoinSwapSubscribeType::PrContractMarketTradeOrders => {//symbolOrderChange
|
|
|
+ serde_json::json!({
|
|
|
+ "type": "subscribe",
|
|
|
+ "topic": format!("/contractMarket/tradeOrders:{}", symbol),
|
|
|
+ "privateChannel":true,
|
|
|
+ "response":true,
|
|
|
+ })
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ //订阅信息生成
|
|
|
+ pub fn get_subscription(&self) -> Vec<String> {
|
|
|
+ let mut array = 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());
|
|
|
+ array.push(ty_str.to_string());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ array
|
|
|
+ }
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************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 = write_tx_am.clone();
|
|
|
+ tokio::spawn(async move {
|
|
|
+ trace!("线程-异步心跳-开始");
|
|
|
+ AbstractWsMode::ping_or_pong(write_tx_clone1, HeartbeatType::Ping, heartbeat_time).await;
|
|
|
+ trace!("线程-异步心跳-结束");
|
|
|
+ });
|
|
|
+
|
|
|
+
|
|
|
+ //设置订阅
|
|
|
+ let subscribe_array = subscription.clone();
|
|
|
+ if login_is {
|
|
|
+ //登录相关
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ //1 链接
|
|
|
+
|
|
|
+ let t2 = tokio::spawn(async move {
|
|
|
+ trace!("线程-异步链接-开始");
|
|
|
+ AbstractWsMode::ws_connect_async(bool_v1, address_url.clone(),
|
|
|
+ label.clone(), subscribe_array,
|
|
|
+ write_rx, read_tx,
|
|
|
+ KucoinSwapWs::message_text,
|
|
|
+ KucoinSwapWs::message_ping,
|
|
|
+ KucoinSwapWs::message_pong,
|
|
|
+ ).await.expect("kucoin-期货");
|
|
|
+ trace!("线程-异步链接-结束");
|
|
|
+ });
|
|
|
+ tokio::try_join!(t2).unwrap();
|
|
|
+ trace!("线程-心跳与链接-结束");
|
|
|
+
|
|
|
+
|
|
|
+ Ok(())
|
|
|
+ }
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ /*****************************************数据解析*****************************************************/
|
|
|
+ /*******************************************************************************************************/
|
|
|
+ //数据解析-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
|
|
|
+ }
|
|
|
+ "-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);
|
|
|
+ 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["type"].as_str() == Option::from("welcome") {
|
|
|
+ //链接成功
|
|
|
+ res_data.code = "-200".to_string();
|
|
|
+ res_data.message = "链接成功,主动发起订阅".to_string();
|
|
|
+ } else if json_value["type"].as_str() == Option::from("ack") {
|
|
|
+ res_data.code = "-201".to_string();
|
|
|
+ res_data.message = "订阅成功".to_string();
|
|
|
+ } else if json_value["type"].as_str() == Option::from("error") {
|
|
|
+ res_data.code = format!("{}", json_value["code"]);
|
|
|
+ res_data.message = format!("{}", json_value["data"].as_str().unwrap());
|
|
|
+ } else if json_value.get("topic").is_some() {
|
|
|
+ res_data.channel = format!("{}", json_value["subject"].as_str().unwrap());
|
|
|
+
|
|
|
+ if json_value["topic"].as_str() == Option::from("/contractAccount/wallet") {
|
|
|
+ res_data.code = "-202".to_string();
|
|
|
+ if json_value["subject"].as_str() == Option::from("availableBalance.change") {
|
|
|
+ res_data.code = "200".to_string();
|
|
|
+ res_data.data = json_value["data"].to_string();
|
|
|
+ } else {}
|
|
|
+ } else {
|
|
|
+ res_data.data = json_value["data"].to_string();
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ res_data.code = "-202".to_string();
|
|
|
+ res_data.message = "未知解析".to_string();
|
|
|
+ }
|
|
|
+ res_data
|
|
|
+ }
|
|
|
+}
|