|
|
@@ -104,38 +104,43 @@ impl AbstractWsMode {
|
|
|
let read = r.lock().await;
|
|
|
|
|
|
let mut data_c = data.clone();
|
|
|
- data_c.label = lable.clone();
|
|
|
- data_c.time = chrono::Utc::now().timestamp_micros();
|
|
|
|
|
|
+ let mut is_send = false;
|
|
|
if data_c.label.contains("gate_usdt_swap") {
|
|
|
if data_c.channel == "futures.order_book" {
|
|
|
if read.len() == 0 {
|
|
|
- read.unbounded_send(data_c).unwrap();
|
|
|
+ is_send = true;
|
|
|
}
|
|
|
} else {
|
|
|
- read.unbounded_send(data_c).unwrap();
|
|
|
+ is_send = true;
|
|
|
}
|
|
|
} else if data_c.label.contains("binance_usdt_swap") {
|
|
|
if data_c.channel == "bookTicker" {
|
|
|
if read.len() == 0 {
|
|
|
- read.unbounded_send(data_c).unwrap();
|
|
|
+ is_send = true;
|
|
|
}
|
|
|
} else {
|
|
|
- read.unbounded_send(data_c).unwrap();
|
|
|
+ is_send = true;
|
|
|
}
|
|
|
} else if data_c.label.contains("bybit_usdt_swap") {
|
|
|
if data_c.channel == "orderbook" {
|
|
|
if read.len() == 0 {
|
|
|
- read.unbounded_send(data_c).unwrap();
|
|
|
+ is_send = true;
|
|
|
}
|
|
|
} else {
|
|
|
- read.unbounded_send(data_c).unwrap();
|
|
|
+ is_send = true;
|
|
|
}
|
|
|
} else {
|
|
|
if read.len() == 0 {
|
|
|
- read.unbounded_send(data_c).unwrap();
|
|
|
+ is_send = true;
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
+ if is_send {
|
|
|
+ data_c.label = lable.clone();
|
|
|
+ data_c.time = chrono::Utc::now().timestamp_micros();
|
|
|
+ read.unbounded_send(data_c).unwrap();
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
let code = data.code.clone();
|
|
|
@@ -189,23 +194,23 @@ impl AbstractWsMode {
|
|
|
Ok::<(), Error>(())
|
|
|
};
|
|
|
|
|
|
- // 防止 cpu 休眠。
|
|
|
- let read2 = read_arc.clone();
|
|
|
- spawn(async move {
|
|
|
- let t_str = "t".to_string();
|
|
|
- let response = ResponseData::new(t_str.clone(),
|
|
|
- t_str.clone(),
|
|
|
- t_str.clone(),
|
|
|
- t_str.clone());
|
|
|
- loop {
|
|
|
- tokio::time::sleep(Duration::from_micros(1)).await;
|
|
|
-
|
|
|
- let t = response.clone();
|
|
|
- let r = read2.clone();
|
|
|
- let read = r.lock().await;
|
|
|
- read.unbounded_send(t.clone()).unwrap();
|
|
|
- }
|
|
|
- });
|
|
|
+ // // 防止 cpu 休眠。
|
|
|
+ // let read2 = read_arc.clone();
|
|
|
+ // spawn(async move {
|
|
|
+ // let t_str = "t".to_string();
|
|
|
+ // let response = ResponseData::new(t_str.clone(),
|
|
|
+ // t_str.clone(),
|
|
|
+ // t_str.clone(),
|
|
|
+ // t_str.clone());
|
|
|
+ // loop {
|
|
|
+ // tokio::time::sleep(Duration::from_micros(1)).await;
|
|
|
+ //
|
|
|
+ // let t = response.clone();
|
|
|
+ // let r = read2.clone();
|
|
|
+ // let read = r.lock().await;
|
|
|
+ // read.unbounded_send(t.clone()).unwrap();
|
|
|
+ // }
|
|
|
+ // });
|
|
|
|
|
|
//必须操作。,因为不同于其他的高级语言,有自动内存管理,所以为了防范地址改变,所以需要做此处理
|
|
|
pin_mut!(stdin_to_ws, ws_to_stdout,);
|