Pause threads when main application is paused

When running the external editor, the main application is put into a
paused state. The view refresh and input read threads were not properly
paused and were interfering with using an external editor in some cases.

5 files changed, 72 insertions, 3 deletions

diff --git a/src/core/src/process/mod.rs b/src/core/src/process/mod.rs
index 364c83f..7bb72ce 100644
--- a/src/core/src/process/mod.rs
+++ b/src/core/src/process/mod.rs
@@ -180,6 +180,7 @@ impl Process {
 
 	fn run_command(&mut self, external_command: &(String, Vec<String>)) -> Result<MetaEvent> {
 		self.view_sender.stop()?;
+		self.event_sender.pause();
 
 		let mut cmd = Command::new(external_command.0.clone());
 		let _ = cmd.args(external_command.1.clone());
@@ -196,6 +197,7 @@ impl Process {
 			})
 			.map_err(|err| anyhow!(err));
 
+		self.event_sender.resume();
 		self.view_sender.start()?;
 
 		result
diff --git a/src/input/src/sender.rs b/src/input/src/sender.rs
index f1c4712..2bed127 100644
--- a/src/input/src/sender.rs
+++ b/src/input/src/sender.rs
@@ -24,6 +24,7 @@ const EVENT_POLL_TIMEOUT: Duration = Duration::from_secs(1);
 pub struct Sender<CustomEvent: crate::CustomEvent> {
 	event_queue: Arc<Mutex<VecDeque<Event<CustomEvent>>>>,
 	poisoned: Arc<AtomicBool>,
+	paused: Arc<AtomicBool>,
 	receiver: crossbeam_channel::Receiver<()>,
 	sender: crossbeam_channel::Sender<EventAction<CustomEvent>>,
 }
@@ -39,6 +40,7 @@ impl<CustomEvent: crate::CustomEvent> Sender<CustomEvent> {
 		Self {
 			event_queue: Arc::new(Mutex::new(VecDeque::new())),
 			poisoned: Arc::new(AtomicBool::new(false)),
+			paused: Arc::new(AtomicBool::new(false)),
 			receiver,
 			sender,
 		}
@@ -58,6 +60,13 @@ impl<CustomEvent: crate::CustomEvent> Sender<CustomEvent> {
 		self.poisoned.load(Ordering::Relaxed)
 	}
 
+	/// Is the sender paused from reading events.
+	#[inline]
+	#[must_use]
+	pub fn is_paused(&self) -> bool {
+		self.paused.load(Ordering::Relaxed)
+	}
+
 	#[inline]
 	pub(crate) fn clone_event_queue(&self) -> Arc<Mutex<VecDeque<Event<CustomEvent>>>> {
 		Arc::clone(&self.event_queue)
@@ -111,6 +120,18 @@ impl<CustomEvent: crate::CustomEvent> Sender<CustomEvent> {
 	pub fn push_event(&self, event: Event<CustomEvent>) -> Result<()> {
 		self.sender.send(EventAction::PushEvent(event)).map_err(map_send_err)
 	}
+
+	/// Pause the event read thread.
+	#[inline]
+	pub fn pause(&self) {
+		self.paused.store(true, Ordering::Relaxed);
+	}
+
+	/// Resume the event read thread.
+	#[inline]
+	pub fn resume(&self) {
+		self.paused.store(false, Ordering::Relaxed);
+	}
 }
 
 #[cfg(test)]
diff --git a/src/input/src/thread.rs b/src/input/src/thread.rs
index e2e8b42..6f32a26 100644
--- a/src/input/src/thread.rs
+++ b/src/input/src/thread.rs
@@ -1,7 +1,8 @@
 use std::{
 	mem,
 	sync::atomic::Ordering,
-	thread::{spawn, JoinHandle},
+	thread::{sleep, spawn, JoinHandle},
+	time::{Duration, Instant},
 };
 
 use anyhow::Result;
@@ -10,6 +11,7 @@ use crossbeam_channel::{bounded, unbounded};
 use crate::{event::Event, event_action::EventAction, sender::Sender};
 
 const MAXIMUM_EVENTS: usize = 100;
+const MINIMUM_PAUSE_RATE: Duration = Duration::from_millis(100);
 
 /// Spawn a thead for handling events.
 ///
@@ -27,6 +29,7 @@ where F: Fn() -> Result<Option<crossterm::event::Event>> {
 	let event_queue = event_sender.clone_event_queue();
 	let push_thread_event_sender = event_sender.clone();
 	let poisoned = event_sender.clone_poisoned();
+
 	let thread = spawn(move || {
 		for msg in receiver {
 			match msg {
@@ -61,7 +64,12 @@ where F: Fn() -> Result<Option<crossterm::event::Event>> {
 	});
 
 	let _push_events_thread = spawn(move || {
+		let mut time = Instant::now();
 		while !push_thread_event_sender.is_poisoned() {
+			while push_thread_event_sender.is_paused() {
+				sleep(time.saturating_duration_since(Instant::now()));
+				time += MINIMUM_PAUSE_RATE;
+			}
 			if let Ok(Some(event)) = (event_provider)() {
 				let _result = push_thread_event_sender.enqueue_event(Event::from(event));
 			}
@@ -232,4 +240,23 @@ mod tests {
 		assert_eq!(events_received.first().unwrap(), &Event::from('b'));
 		assert_eq!(events_received.last().unwrap(), &Event::from('a'));
 	}
+
+	#[test]
+	fn thread_pause_resume() {
+		// setup event provider to continuously provide a key event
+		let (mut sender, _thread) = spawn_event_thread(|| {
+			Ok(Some(crossterm::event::Event::Key(crossterm::event::KeyEvent::new(
+				crossterm::event::KeyCode::Char('a'),
+				crossterm::event::KeyModifiers::empty(),
+			))))
+		});
+
+		sender.pause();
+		sender.clone_event_queue().lock().clear(); // remove any events that were already enqueued
+		assert_eq!(sender.read_event(), Event::None); // sadly this will pause for a second
+		sender.resume();
+		assert_eq!(sender.read_event(), Event::from('a'));
+		sender.end().unwrap();
+		while !sender.is_poisoned() {}
+	}
 }
diff --git a/src/view/src/sender.rs b/src/view/src/sender.rs
index d69872c..cd42938 100644
--- a/src/view/src/sender.rs
+++ b/src/view/src/sender.rs
@@ -20,6 +20,7 @@ fn map_send_err(_: channel::SendError<ViewAction>) -> Error {
 #[derive(Clone, Debug)]
 pub struct Sender {
 	poisoned: Arc<AtomicBool>,
+	paused: Arc<AtomicBool>,
 	sender: channel::Sender<ViewAction>,
 	render_slice: Arc<Mutex<RenderSlice>>,
 }
@@ -32,6 +33,7 @@ impl Sender {
 		Self {
 			poisoned: Arc::new(AtomicBool::new(false)),
 			sender,
+			paused: Arc::new(AtomicBool::new(false)),
 			render_slice: Arc::new(Mutex::new(RenderSlice::new())),
 		}
 	}
@@ -50,6 +52,13 @@ impl Sender {
 		self.poisoned.load(Ordering::Acquire)
 	}
 
+	/// Is the sender paused from refreshing the view.
+	#[inline]
+	#[must_use]
+	pub fn is_paused(&self) -> bool {
+		self.paused.load(Ordering::Relaxed)
+	}
+
 	/// Clone the render slice.
 	#[inline]
 	#[must_use]
@@ -63,6 +72,7 @@ impl Sender {
 	/// Results in an error if the sender has been closed.
 	#[inline]
 	pub fn start(&self) -> Result<()> {
+		self.paused.store(false, Ordering::Relaxed);
 		self.sender.send(ViewAction::Start).map_err(map_send_err)
 	}
 
@@ -72,6 +82,7 @@ impl Sender {
 	/// Results in an error if the sender has been closed.
 	#[inline]
 	pub fn stop(&self) -> Result<()> {
+		self.paused.store(true, Ordering::Relaxed);
 		self.sender.send(ViewAction::Stop).map_err(map_send_err)
 	}
 
@@ -176,6 +187,7 @@ mod tests {
 		with_view_sender(|context| {
 			context.sender.start().unwrap();
 			context.assert_sent_messages(vec!["Start"]);
+			assert!(!context.sender.is_paused());
 		});
 	}
 
@@ -192,6 +204,7 @@ mod tests {
 		with_view_sender(|context| {
 			context.sender.stop().unwrap();
 			context.assert_sent_messages(vec!["Stop"]);
+			assert!(context.sender.is_paused());
 		});
 	}
 
diff --git a/src/view/src/thread.rs b/src/view/src/thread.rs
index 16923ad..8126512 100644
--- a/src/view/src/thread.rs
+++ b/src/view/src/thread.rs
@@ -20,6 +20,7 @@ const MINIMUM_TICK_RATE: Duration = Duration::from_millis(20); // ~50 Hz update
 pub fn spawn_view_thread<T: Tui + Send + 'static>(mut view: View<T>) -> (Sender, JoinHandle<()>) {
 	let (sender, receiver) = unbounded();
 	let view_sender = Sender::new(sender.clone());
+	let refresh_thread_view_sender = view_sender.clone();
 	let view_render_slice = view_sender.clone_render_slice();
 	let crashed = view_sender.clone_poisoned();
 
@@ -61,8 +62,13 @@ pub fn spawn_view_thread<T: Tui + Send + 'static>(mut view: View<T>) -> (Sender,
 		let sleep_time = MINIMUM_TICK_RATE / 2;
 		let mut time = Instant::now();
 		while sender.send(ViewAction::Refresh).is_ok() {
-			sleep(time.saturating_duration_since(Instant::now()));
-			time += sleep_time;
+			loop {
+				sleep(time.saturating_duration_since(Instant::now()));
+				time += sleep_time;
+				if !refresh_thread_view_sender.is_paused() {
+					break;
+				}
+			}
 		}
 	});