use async_std::sync::Mutex;
use async_std::task;
use crypto_box::SecretKey;
use log::LevelFilter;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use vented::event::Event;
use vented::server::data::{Node, ServerTimeouts};
use vented::server::server_events::NODE_LIST_REQUEST_EVENT;
use vented::server::VentedServer;
fn setup() {
let _ = simple_logger::SimpleLogger::new()
.with_module_level("async_std", LevelFilter::Warn)
.with_module_level("async_io", LevelFilter::Warn)
.with_module_level("polling", LevelFilter::Warn)
.init();
}
#[test]
fn test_server_communication() {
setup();
let ping_count = Arc::new(AtomicUsize::new(0));
let pong_count = Arc::new(AtomicUsize::new(0));
let c_pinged = Arc::new(AtomicBool::new(false));
let mut rng = rand::thread_rng();
let global_secret_a = SecretKey::generate(&mut rng);
let global_secret_b = SecretKey::generate(&mut rng);
let global_secret_c = SecretKey::generate(&mut rng);
let nodes = vec![
Node {
id: "A".to_string(),
addresses: vec!["localhost:22222".to_string()],
public_key: global_secret_a.public_key(),
trusted: true,
},
Node {
id: "B".to_string(),
addresses: vec![],
public_key: global_secret_b.public_key(),
trusted: false,
},
Node {
id: "C".to_string(),
addresses: vec![],
public_key: global_secret_c.public_key(),
trusted: false,
},
];
let mut nodes_a = nodes.clone();
for i in 0..10 {
nodes_a.push(Node {
id: format!("Node-{}", i),
addresses: vec!["192.168.178.1".to_string()],
public_key: global_secret_c.public_key(),
trusted: false,
})
}
task::block_on(async {
let mut server_a = VentedServer::new(
"A".to_string(),
global_secret_a,
nodes_a,
ServerTimeouts::default(),
);
let mut server_b = VentedServer::new(
"B".to_string(),
global_secret_b,
nodes.clone(),
ServerTimeouts::default(),
);
let mut server_c = VentedServer::new(
"C".to_string(),
global_secret_c,
nodes,
ServerTimeouts::default(),
);
server_a.listen("localhost:22222".to_string());
server_a.on("ping", {
let ping_count = Arc::clone(&ping_count);
move |_| {
let ping_count = Arc::clone(&ping_count);
Box::pin(async move {
ping_count.fetch_add(1, Ordering::Relaxed);
Some(Event::new("pong".to_string()))
})
}
});
server_b.on("pong", {
let pong_count = Arc::clone(&pong_count);
move |_| {
let pong_count = Arc::clone(&pong_count);
Box::pin(async move {
pong_count.fetch_add(1, Ordering::Relaxed);
None
})
}
});
server_c.on("ping", {
let c_pinged = Arc::clone(&c_pinged);
move |_| {
let c_pinged = Arc::clone(&c_pinged);
Box::pin(async move {
c_pinged.store(true, Ordering::Relaxed);
None
})
}
});
for i in 0..10 {
assert!(server_a
.emit(format!("Nodes-{}", i), Event::new("ping"))
.await
.is_err());
}
server_b
.emit("A", Event::new(NODE_LIST_REQUEST_EVENT))
.await
.unwrap();
server_c
.emit("A", Event::new("ping".to_string()))
.await
.unwrap();
server_b
.emit("C", Event::new("ping".to_string()))
.await
.unwrap();
for _ in 0..9 {
server_b
.emit("A", Event::new("ping".to_string()))
.await
.unwrap();
}
server_a
.emit("B", Event::new("pong".to_string()))
.await
.unwrap();
task::sleep(Duration::from_secs(2)).await;
});
assert_eq!(ping_count.load(Ordering::SeqCst), 10);
assert_eq!(pong_count.load(Ordering::SeqCst), 10);
assert!(c_pinged.load(Ordering::SeqCst));
}
const COUNT: usize = 20000;
#[test]
fn test_high_traffic() {
setup();
let ping_count = Arc::new(AtomicUsize::new(0));
let pong_count = Arc::new(AtomicUsize::new(0));
let last_pong = Arc::new(Mutex::new(Instant::now()));
let running = Arc::new(AtomicBool::new(true));
let mut rng = rand::thread_rng();
let global_secret_a = SecretKey::generate(&mut rng);
let global_secret_b = SecretKey::generate(&mut rng);
let nodes = vec![
Node {
id: "A".to_string(),
addresses: vec!["localhost:22223".to_string()],
public_key: global_secret_a.public_key(),
trusted: true,
},
Node {
id: "B".to_string(),
addresses: vec![],
public_key: global_secret_b.public_key(),
trusted: false,
},
];
task::block_on(async {
let mut server_a = VentedServer::new(
"A".to_string(),
global_secret_a,
nodes.clone(),
ServerTimeouts::default(),
);
let mut server_b = VentedServer::new(
"B".to_string(),
global_secret_b,
nodes,
ServerTimeouts::default(),
);
server_a.listen("localhost:22223".to_string());
task::sleep(Duration::from_millis(10)).await;
server_a.on("ping", {
let ping_count = Arc::clone(&ping_count);
move |_| {
let ping_count = Arc::clone(&ping_count);
Box::pin(async move {
ping_count.fetch_add(1, Ordering::Relaxed);
Some(Event::new("pong".to_string()))
})
}
});
server_b.on("pong", {
let pong_count = Arc::clone(&pong_count);
let running = Arc::clone(&running);
let last_pong = Arc::clone(&last_pong);
move |_| {
let pong_count = Arc::clone(&pong_count);
let running = Arc::clone(&running);
let last_pong = Arc::clone(&last_pong);
Box::pin(async move {
*last_pong.lock().await = Instant::now();
let num = pong_count.fetch_add(1, Ordering::Relaxed);
log::info!("Received pong nr. {}", num);
if num >= COUNT - (COUNT / 100) {
running.store(true, Ordering::Relaxed);
None
} else {
Some(Event::new("ping".to_string()))
}
})
}
});
let mut promises = Vec::new();
for _ in 0..COUNT / 100 {
promises.push(server_b.emit("A", Event::new("ping")));
}
futures::future::join_all(promises).await;
while running.load(Ordering::Relaxed)
&& last_pong.lock().await.elapsed() < Duration::from_secs(2)
{
task::sleep(Duration::from_micros(10)).await;
}
});
assert_eq!(ping_count.load(Ordering::SeqCst), COUNT);
assert_eq!(pong_count.load(Ordering::SeqCst), COUNT);
}