1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

//! Utilities for writing tests that interact with the runtime.
use std::{
	borrow::BorrowMut,
	mem,
	sync::{
		atomic::{AtomicBool, Ordering},
		Arc,
	},
	thread::{sleep, spawn},
	time::Duration,
};

use crossbeam_channel::{bounded, Receiver, Sender};
use parking_lot::Mutex;

use crate::{Installer, Status, ThreadStatuses};

const WAIT_TIME: Duration = Duration::from_millis(100);

/// A mocked version of the `Notifier`, that will interact directly with a `ThreadStatuses` without the use of a thread
/// or the `Runtime`.
#[derive(Debug)]
pub struct MockNotifier<'notifier> {
	threadable_statuses: &'notifier ThreadStatuses,
}

impl<'notifier> MockNotifier<'notifier> {
	/// Create a new instance of a `MockNotifier`.
	#[inline]
	#[must_use]
	pub const fn new(threadable_statuses: &'notifier ThreadStatuses) -> Self {
		Self { threadable_statuses }
	}

	/// Register a thread by name and status. This does not create a thread.
	#[inline]
	pub fn register_thread(&mut self, thread_name: &str, status: Status) {
		self.threadable_statuses.register_thread(thread_name, status);
	}
}

/// A tester utility for `Threadable`.
#[derive(Clone, Debug)]
pub struct ThreadableTester {
	receiver: Receiver<(String, Status)>,
	sender: Sender<(String, Status)>,
	statuses: Arc<Mutex<Vec<Status>>>,
	ended: Arc<AtomicBool>,
}

impl ThreadableTester {
	/// Create a new instance of the test utility.
	#[inline]
	#[must_use]
	pub fn new() -> Self {
		let (sender, receiver) = bounded(0);

		Self {
			receiver,
			sender,
			statuses: Arc::new(Mutex::new(vec![Status::New])),
			ended: Arc::new(AtomicBool::new(true)),
		}
	}

	/// Take the current `Status` changes.
	#[inline]
	#[must_use]
	pub fn take_statuses(&self) -> Vec<Status> {
		mem::take(self.statuses.lock().borrow_mut())
	}

	/// Start a `Threadable` running the thread specified by the name, to completion in a separate thread.
	#[inline]
	pub fn start_threadable<Threadable: crate::Threadable>(&self, theadable: &Threadable, thread_name: &str) {
		self.ended.store(false, Ordering::Release);
		let installer = Installer::new(self.sender.clone());
		theadable.install(&installer);
		let mut ops = installer.into_ops();
		let op = ops.remove(thread_name).expect("Expected to find thead");

		let statuses = Arc::clone(&self.statuses);
		let receiver = self.receiver.clone();

		let _status_thread_id = spawn(move || {
			// capture!(receiver, statuses);
			for (_, status) in &receiver {
				let mut statuses_lock = statuses.lock();
				let last_status = statuses_lock.last().expect("Last status always exists");
				if !matches!(*last_status, Status::Error(_)) && last_status != &status {
					statuses_lock.push(status);
				}
			}
		});
		let _op_id = spawn(op);
		self.ended.store(true, Ordering::Release);
	}

	/// Wait for a particular status to be reached.
	///
	/// # Panics
	///
	/// Will panic if the wait takes too long and times out.
	#[inline]
	pub fn wait_for_status(&self, status: &Status) {
		let mut attempt = 0;

		loop {
			let statuses_lock = self.statuses.lock();
			let current_status = statuses_lock.last().expect("Expected to lock");

			if current_status == status {
				break;
			}
			assert!(
				attempt <= 100,
				"Timeout waited for status change to '{:?}' on thread.\n Status is: {:?}",
				status,
				current_status,
			);

			sleep(WAIT_TIME);
			attempt += 1;
		}
	}

	/// Wait for the thread started in `start_threadable` to finish.
	///
	/// # Panics
	///
	/// Will panic if the wait takes too long and times out.
	#[inline]
	pub fn wait_for_finished(&self) {
		let mut attempt = 0;

		loop {
			if self.ended.load(Ordering::Acquire) {
				break;
			}

			sleep(WAIT_TIME);
			attempt += 1;
			assert!(attempt <= 100, "Timeout waited for thread to finish");
		}
	}
}