saas_zmq/
lib.rs

1
2pub mod common;
3pub mod config;
4pub mod subscribe;
5pub mod request;
6
7pub use common::PEERS as PEERS;
8pub use common::COMMON_WORKING_ID as COMMON_WORKING_ID;
9pub use request::request as request;
10pub use subscribe::do_subscribe as do_subscribe;
11pub use subscribe::get_subscribe as get_subscribe;
12
13#[cfg(test)]
14mod tests {
15    
16    use std::collections::HashMap;
17    use std::str::from_utf8;
18    use std::time::Duration;
19    use std::{fs, thread};
20
21    use chrono::Local;
22    use futures::future::join_all;
23    use log::{debug, error, info, trace, warn};
24    use request::{ZmqRequest, ZmqRequestInfo, ZmqRequestManager, ZmqRequester};
25    use subscribe::do_subscribe;
26
27    use super::*;
28    use super::common::*;
29    use lazy_static::lazy_static;
30
31    lazy_static! {
32        static ref SYMBOLS:Vec<String>=test_symbols();
33    }
34            
35    #[actix::test]
36    async fn test_sub(){
37        let _=logger_init("DEBUG".to_string(), "server.log".to_string());
38        let (markprice_msg_tx,markprice_msg_rx)=async_channel::bounded(1024);
39        let (markprice_sub_tx,markprice_sub_rx)=async_channel::bounded(256);
40        let markprice_topics:Vec<String>=SYMBOLS.clone().into_iter().map(|s| format!("B/m/l/{}",s)).collect();
41        // println!("{:?}",markprice_topics);
42        actix::spawn(do_subscribe(markprice_topics,markprice_msg_tx,markprice_sub_rx));
43        for i in 0..20 {
44            let mpmsg=markprice_msg_rx.recv().await.unwrap();
45            for it in mpmsg {
46                info!("markprice_msg_rx message: {:?}", from_utf8(&it));
47            }
48        }
49    }
50
51    #[actix::test]
52    async fn test_req(){
53        let _=logger_init("DEBUG".to_string(), "server.log".to_string());
54        let r=get_request();
55        let response=request(r).await;
56        info!("response >>> {:?}",response);
57    }
58
59    #[actix::test]
60    async fn test_req_autochange(){
61        let _=logger_init("DEBUG".to_string(), "server.log".to_string());
62        for i in 0..20 {
63            let r=get_request();
64            let response=request(r).await;
65            info!("response >>> {:?}",response);
66        }
67    }
68
69    #[actix::test]
70    async fn test_req_mt(){
71        let _=logger_init("DEBUG".to_string(), "server.log".to_string());
72        let mut jhs=vec![];
73        
74        for i in 0..20 {
75            let jh=actix::spawn(async move{
76                let r=get_request();
77                let response=request(r).await;
78                info!("{} response >>> {:?}",i,response);
79            });
80            jhs.push(jh);
81        }
82        join_all(jhs).await;
83    }
84
85    #[actix::test]
86    async fn test_requester_init(){
87        let _=logger_init("DEBUG".to_string(), "server.log".to_string());
88        let requester: ZmqRequester=ZmqRequester::new();
89        let mut pi=tokio::time::interval(Duration::from_secs(1));
90        for i in 0..20 {
91            pi.tick().await;
92            println!(">>> {:?}",requester.get_working_id());
93        }
94        // drop(req);
95        // std::future::pending::<()>().await;
96    }
97
98    #[actix::test]
99    async fn test_requester_req(){
100        let _=logger_init("DEBUG".to_string(), "server.log".to_string());
101        let requester: ZmqRequester=ZmqRequester::new();
102        let request=get_request();
103        let response=requester.req(request).await;
104        info!("response >>> {:?}",response);
105    }
106
107
108
109    #[actix::test]
110    async fn test_req_pool(){
111        _ = logger_init("DEBUG".to_string(), "server.log".to_string());
112        let pool=ZmqRequestManager::init_request_pool();
113        
114        for i in 0..20 {
115            let p=pool.clone();
116            let request=get_request();
117
118            // tokio::spawn(async move{
119                let requester=p.get().await.unwrap();
120                // let requester=mz.lock().unwrap();
121                let response=requester.req(request).await;
122            // });
123            
124            info!("Got response >>> {:?}",i);
125        }
126        
127    }
128
129
130    fn test_symbols() -> Vec<String>{
131        vec!["btcusdt".to_string(),"ethusdt".to_string(),"ltcusdt".to_string(),"eosusdt".to_string(),"bchusdt".to_string(),"trxusdt".to_string(),"filusdt".to_string(),"linkusdt".to_string(),"dotusdt".to_string(),"xrpusdt".to_string(),"dogeusdt".to_string(),"shibusdt".to_string(),"adausdt".to_string(),"maticusdt".to_string(),"solusdt".to_string(),"bnbusdt".to_string(),"avaxusdt".to_string(),"manausdt".to_string(),"axsusdt".to_string(),"uniusdt".to_string(),"opusdt".to_string(),"blurusdt".to_string(),"wldusdt".to_string(),"compusdt".to_string(),"pepeusdt".to_string(),"blzusdt".to_string(),"cyberusdt".to_string(),"bigtimeusdt".to_string(),"trbusdt".to_string(),"arbusdt".to_string(),"ordiusdt".to_string(),"memeusdt".to_string(),"atomusdt".to_string(),"bsvusdt".to_string(),"seiusdt".to_string()]
132    }
133
134
135    fn get_request() -> ZmqRequest {
136        let mut param=HashMap::new();
137        param.insert("symbole".to_string(), "btcusdt".to_string());
138        let request_info: ZmqRequestInfo=ZmqRequestInfo::new("get/kline/".to_string(), param, None);
139        let request=ZmqRequest::new(ZmqServer::KlineCache, Duration::from_millis(333), request_info);
140        return request;
141    }
142
143        fn logger_init(level: String,log_file: String) -> Result<(), fern::InitError> {
144        // let log_level = env::var("LOG_LEVEL").unwrap_or("INFO".into());
145        let log_level=level;
146        let log_level = log_level
147            .parse::<log::LevelFilter>()
148            .unwrap_or(log::LevelFilter::Info);
149
150        let mut builder = fern::Dispatch::new()
151            .format(|out, message, record| {
152                out.finish(format_args!(
153                    "[{}][{}:{}][{}][{:?}] {}",
154                    // local times is fine for toy project,
155                    // in prod we'd use UTC and add dates though
156                    // chrono::Local::now().format("%Y-%m-%d %H:%M:%S.%z"),
157                    Local::now(),
158                    record.target(),
159                    match record.line(){
160                        Some(n)=>n,
161                        None=>0,
162                    },
163                    record.level(),
164                    thread::current().id(),
165                    message
166                ))
167            })
168            .level(log_level)
169            // log to stderr
170            .chain(std::io::stderr());
171
172        // also log to file if one is provided via env
173        // let log_file = env::var("LOG_FILE").ok();
174        // if let Some(log_file) = log_file {
175            let log_file = fs::File::create(log_file)?;
176            builder = builder.chain(log_file);
177        // }
178
179        // globally apply logger
180        builder.apply()?;
181
182        trace!("TRACE output enabled");
183        debug!("DEBUG output enabled");
184        info!("INFO output enabled");
185        warn!("WARN output enabled");
186        error!("ERROR output enabled");
187
188        Ok(())
189    }
190}
saas_zmq/lib.rs

saas_zmq/
lib.rs
