|
|
@@ -0,0 +1,104 @@
|
|
|
+use std::collections::{BTreeMap, HashMap};
|
|
|
+use std::sync::{Arc};
|
|
|
+use std::sync::atomic::AtomicBool;
|
|
|
+use lazy_static::lazy_static;
|
|
|
+use tokio::sync::{Mutex};
|
|
|
+use tracing::info;
|
|
|
+use exchanges::binance_swap_rest::BinanceSwapRest;
|
|
|
+use exchanges::binance_swap_ws::{BinanceSwapSubscribeType, BinanceSwapWs, BinanceSwapWsType};
|
|
|
+use exchanges::response_base::ResponseData;
|
|
|
+use standard::{Record, SpecialTrade};
|
|
|
+use standard::exchange::ExchangeEnum;
|
|
|
+use standard::exchange_struct_handler::ExchangeStructHandler;
|
|
|
+use crate::listener_tools::{update_record, update_trade};
|
|
|
+
|
|
|
+// type DepthMap = HashMap<String, Vec<SpecialDepth>>;
|
|
|
+pub type TradeMap = HashMap<String, Vec<SpecialTrade>>;
|
|
|
+pub type RecordMap = HashMap<String, Record>;
|
|
|
+const EXCHANGE_NAME: &str = "binance_usdt_swap";
|
|
|
+
|
|
|
+lazy_static! {
|
|
|
+ // static ref DEPTH_MAP: Mutex<DepthMap> = Mutex::new(HashMap::new());
|
|
|
+ static ref TRADES_MAP: Mutex<TradeMap> = Mutex::new(HashMap::new());
|
|
|
+ static ref RECORD_MAP: Mutex<RecordMap> = Mutex::new(HashMap::new());
|
|
|
+}
|
|
|
+
|
|
|
+pub async fn run_listener(is_shutdown_arc: Arc<AtomicBool>) {
|
|
|
+ let name = "binance_usdt_swap_listener";
|
|
|
+ // 订阅所有币种
|
|
|
+ let login = BTreeMap::new();
|
|
|
+ let mut binance_rest = BinanceSwapRest::new(false, login);
|
|
|
+ let response = binance_rest.get_exchange_info().await;
|
|
|
+ let mut symbols = vec![];
|
|
|
+ if response.code == 200 {
|
|
|
+ let data = response.data["symbols"].as_array().unwrap();
|
|
|
+ for info in data {
|
|
|
+ let s = info["symbol"].as_str().unwrap().to_string();
|
|
|
+ if !s.ends_with("USDT") {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ let symbol = s.replace("USDT", "_USDT");
|
|
|
+ symbols.push(symbol)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ info!(?symbols);
|
|
|
+
|
|
|
+ for chunk in symbols.chunks(20) {
|
|
|
+ let ws_name = name.to_string();
|
|
|
+ let (write_tx, write_rx) = futures_channel::mpsc::unbounded();
|
|
|
+ let write_tx_am = Arc::new(Mutex::new(write_tx));
|
|
|
+ let symbols_chunk = chunk.iter().cloned().collect::<Vec<String>>();
|
|
|
+ let is_shutdown_clone = Arc::clone(&is_shutdown_arc);
|
|
|
+
|
|
|
+ tokio::spawn(async move {
|
|
|
+ let mut ws = BinanceSwapWs::new_with_tag(ws_name, false, None, BinanceSwapWsType::PublicAndPrivate);
|
|
|
+ ws.set_subscribe(vec![
|
|
|
+ BinanceSwapSubscribeType::PuAggTrade,
|
|
|
+ BinanceSwapSubscribeType::PuKline,
|
|
|
+ ]);
|
|
|
+
|
|
|
+ // 建立链接
|
|
|
+ ws.set_symbols(symbols_chunk);
|
|
|
+ ws.ws_connect_async(is_shutdown_clone, data_listener, &write_tx_am, write_rx).await.expect("链接失败(内部一个心跳线程应该已经关闭了)");
|
|
|
+ });
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+// 读取数据
|
|
|
+pub async fn data_listener(response: ResponseData) {
|
|
|
+ if response.code != 200 {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+
|
|
|
+ match response.channel.as_str() {
|
|
|
+ // 深度数据
|
|
|
+ "深度" => {
|
|
|
+ // let depth = ExchangeStructHandler::order_book_handle(ExchangeEnum::GateSwap, &response);
|
|
|
+ //
|
|
|
+ // update_depth(&depth).await;
|
|
|
+ },
|
|
|
+ // 订单流数据
|
|
|
+ "aggTrade" => {
|
|
|
+ let trades = ExchangeStructHandler::trades_handle(ExchangeEnum::BinanceSwap, &response);
|
|
|
+
|
|
|
+ for trade in trades.iter() {
|
|
|
+ let trades_map = TRADES_MAP.lock().await;
|
|
|
+
|
|
|
+ update_trade(trade, trades_map, EXCHANGE_NAME).await;
|
|
|
+ }
|
|
|
+ },
|
|
|
+ // k线数据
|
|
|
+ "kline" => {
|
|
|
+ let records = ExchangeStructHandler::records_handle(ExchangeEnum::BinanceSwap, &response);
|
|
|
+
|
|
|
+ for record in records.iter() {
|
|
|
+ let record_map= RECORD_MAP.lock().await;
|
|
|
+
|
|
|
+ update_record(record, record_map, EXCHANGE_NAME).await;
|
|
|
+ }
|
|
|
+ },
|
|
|
+ _ => {
|
|
|
+ info!("48 未知的数据类型: {:?}", response)
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
