cnsprcy 0.3.2

use crate::message::{
	Payload,
	Packet,
	PingOrPong,
};

use crate::peer::Clock;

use super::*;


async fn assert_reply_ok(reply_rx: ReplyReceiver<bool>, msg: &str) {
	let reply = timeout_unwrap(reply_rx).await;
	assert!(reply, "{}", msg);
}

async fn assert_reply_false(reply_rx: ReplyReceiver<bool>, msg: &str) {
	let reply = timeout_unwrap(reply_rx).await;
	assert!(!reply, "{}", msg);
}

async fn print_ifaces(r_q_s: &Sender<RouterReq>) {
	let if_addrs = r_q_s
		.get(RouterReq::GetInterfaces)
		.timeout_ms(100)
		.await
		.expect("timed out trying to get interface list")
		.expect("failed to get interface list");
	println!("interfaces:");
	if_addrs.iter().for_each(|iface| println!("\t{iface}"));
}


#[test_log::test(tokio::test(start_paused = true))]
async fn chacha_setup_and_loopback() {

	let span = tracing::info_span!("cc20_lo_test");

	let RouterSetup {
		router,
		mut inbound_packet_receiver,
		router_monitor,
		outbound_packet_sender,
		router_query_sender
	} = Router::setup(KeyArg::random(), &span);

	let router_handle = tokio::task::spawn(router.run());

	assert!(
		!*router_monitor.borrow(),
		"router should be dormant"
	);

	let msg = "bind interface 1";
	let reply = router_query_sender.request(
		RouterReq::BindInterface,
		loopback_iface_address(1)
	);
	assert_reply_ok(reply, msg).await;

	assert!(
		*router_monitor.borrow(),
		"router should be active"
	);

	let example_id = NodeIDArg::random().get();

	let pkt = Packet {
		dst: example_id,
		src: example_id,
		clk: Clock::null_clock(),
		hsh: 0,
		vec: Vec::new(),
		png: PingOrPong::Ping,
		pyl: Payload::None
	};

	// send a packet to interface 2 (currently unbound)
	outbound_packet_sender.send((
		loopback_address(2),
		pkt.clone()
	)).unwrap();

	// yield to make sure the interface 2 really doesn't get it
	task::yield_now().await;

	// packet shouldn't be received
	assert_eq!(
		inbound_packet_receiver.try_recv().unwrap_err(),
		tokio::sync::mpsc::error::TryRecvError::Empty
	);

	let msg = "bind interface 2";
	let reply = router_query_sender.request(
		RouterReq::BindInterface,
		loopback_iface_address(2)
	);
	assert_reply_ok(reply, msg).await;

	// send the packet again
	outbound_packet_sender.send((
		loopback_address(2),
		pkt.clone()
	)).unwrap();

	// this time it should be received
	let (_addr, pkt_received) = inbound_packet_receiver.recv().await.unwrap();
	assert_eq!(pkt, pkt_received);

	// send the packet back
	outbound_packet_sender.send((
		loopback_address(1),
		pkt.clone()
	)).unwrap();

	let (_addr, pkt_received) = inbound_packet_receiver.recv().await.unwrap();
	assert_eq!(pkt, pkt_received);

	print_ifaces(&router_query_sender).await;

	let msg = "remove interface 1";
	let reply = router_query_sender.request(
		RouterReq::RemoveInterface,
		dbg!(loopback_address(1))
	);
	assert_reply_ok(reply, msg).await;

	print_ifaces(&router_query_sender).await;

	let msg = "remove interface 1 again, should error";
	let reply = router_query_sender.request(
		RouterReq::RemoveInterface,
		loopback_address(1)
	);
	assert_reply_false(reply, msg).await;

	// packet queue should still be empty (i guess?)
	assert_eq!(
		inbound_packet_receiver.try_recv().unwrap_err(),
		tokio::sync::mpsc::error::TryRecvError::Empty
	);

	let msg = "remove interface 2";
	let reply = router_query_sender.request(
		RouterReq::RemoveInterface,
		loopback_address(2)
	);
	assert_reply_ok(reply, msg).await;

	let msg = "remove interface 2 again, should error";
	let reply = router_query_sender.request(
		RouterReq::RemoveInterface,
		loopback_address(2)
	);
	assert_reply_false(reply, msg).await;

	let msg = "remove an interface that was never bound, should error";
	let reply = router_query_sender.request(
		RouterReq::RemoveInterface,
		loopback_address(3)
	);
	assert_reply_false(reply, msg).await;

	assert!(
		!*router_monitor.borrow(),
		"router should be dormant"
	);

	// this shuts down the router
	drop(router_query_sender);
	assert!(inbound_packet_receiver.recv().await.is_none());

	timeout_100ms(router_handle).await.unwrap();
}

#[test_log::test(tokio::test(start_paused = true))]
async fn dummy_servers_loopback() {

	let span = tracing::info_span!("dummy_servers_loopback");

	let RouterSetup {
		router,
		mut inbound_packet_receiver,
		router_monitor: _,
		outbound_packet_sender,
		router_query_sender
	} = Router::setup(KeyArg::random(), &span);


	let (dummies, observer_handle) = DummyServer::create_n(10);

	let launchers: Vec<InterfaceLauncher> = dummies
		.into_iter()
		.map(InterfaceLauncher::Dummy)
		.collect();

	let addr1 = launchers[0].get_address().get_address();
	let addr2 = launchers[1].get_address().get_address();

	let router_handle = task::spawn(router.run());

	for launcher in launchers {
		let reply = router_query_sender.request(
			RouterReq::LaunchInterface,
			launcher
		);
		assert!(reply.await.unwrap());
	}

	// give the tasks time to launch
	task::yield_now().await;

	let example_id = NodeIDArg::random().get();
	let pkt = Packet {
		dst: example_id,
		src: example_id,
		clk: Clock::null_clock(),
		hsh: 0,
		vec: Vec::new(),
		png: PingOrPong::Ping,
		pyl: Payload::None
	};

	// Sending packet from ??? to 2
	outbound_packet_sender.send((
		addr2,
		pkt.clone()
	)).unwrap();

	let (_sending_addr, pkt_received) = inbound_packet_receiver
		.recv()
		.timeout_ms(100)
		.await
		.unwrap()
		.unwrap();

	assert_eq!(pkt, pkt_received);

	// Sending packet from ??? to 1
	outbound_packet_sender.send((
		addr1,
		pkt.clone()
	)).unwrap();

	let (_sending_addr, pkt_received) = inbound_packet_receiver
		.recv()
		.timeout_ms(100)
		.await
		.unwrap()
		.unwrap();

	assert_eq!(pkt, pkt_received);

	outbound_packet_sender.send((
		dummy_address(99),
		pkt.clone()
	)).unwrap();

	drop(router_query_sender);

	assert!(
		timeout_100ms(inbound_packet_receiver.recv()).await.is_none()
	);


	timeout_unwrap(router_handle).await;
	timeout_unwrap(observer_handle).await;
}
/*
use futures::stream::{
	FuturesUnordered,
	StreamExt
};

#[test_log::test(tokio::test)]
async fn tokio_select_test() {

	let mut handles = FuturesUnordered::new();

	let (tx1, mut rx1) = new_channel::<u8>();
	let (tx2, mut rx2) = new_channel::<u8>();

	handles.push(tokio::spawn(async move {
		let num = rx1.recv().await.unwrap();
		println!("task: {}", num);
		let _ = tx2.send(num);
	}));

	let _ = tx1.send(123);

	let in_2s = tokio::time::Instant::now() + Duration::from_secs(2);

	let mut count = 0;

	loop { tokio::select! {
		_ = tokio::time::sleep_until(in_2s) => {
			println!("loop timer finished");
			break;
		},
		h = handles.next() => { match h {
			Some(x) => {
				println!("task finished {:?}", x);
			},
			None => {
				println!("handles done {}", count);
				count += 1;
				tokio::time::sleep(Duration::from_millis(500)).await;
				assert!(count <= 100);
			}
		}},
		r = rx2.recv() => { match r {
			Some(num) => {
				println!("loop: {}", num);
			}
			None => {
				println!("tx2 dropped");
			}
		}}
	}}


	panic!("done");

}

*/