|
@@ -1,519 +1,516 @@
|
|
|
-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 tungstenite::client::{AutoStream, connect_with_proxy, ProxyAutoStream};
|
|
|
|
|
-use tungstenite::{connect, Message, WebSocket};
|
|
|
|
|
-use tungstenite::protocol::WebSocketConfig;
|
|
|
|
|
-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(lable: 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, 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(lable: 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
|
|
|
|
|
+// }
|
|
|
|
|
+// }
|