progress command builder

src/main.rs

@@ -22,6 +22,7 @@ mod fixed_stack;
 mod process;
 mod work;
 
+#[cfg(feature="progress")] mod maybe_single;
 #[cfg(feature="progress")] mod task_list;
 #[cfg(feature="progress")] mod progress;
 

src/maybe_single.rs

@@ -0,0 +1,120 @@
+//! A one-or-more iterator type
+
+use std::{
+    iter::{self, Once,FromIterator,Extend},
+};
+
+pub enum IntoIter<T>
+{
+    Many(std::vec::IntoIter<T>),
+    Single(Once<T>),
+}
+
+/// A type that might hold once or more values
+pub enum MaybeSingle<T> {
+    Single(Option<T>),
+    Many(Vec<T>),
+}
+
+impl<T> Extend<T> for MaybeSingle<T>
+{
+    fn extend<I: IntoIterator<Item = T>>(&mut self, iter: I)
+    {
+	match self {
+	    Self::Single(single) => {
+		*self = Self::Many(iter::once(single.take().unwrap()).chain(iter).collect());
+	    },
+	    Self::Many(ref mut many) => {
+		many.extend(iter);
+	    },
+	}
+    }
+}
+
+impl<T> MaybeSingle<T> {
+    pub fn is_single(&self) -> bool
+    {
+	if let Self::Single(_) = &self {
+	    true
+	} else {
+	    false
+	}
+    }
+    
+    pub fn push(&mut self, value: T)
+    {
+	match self {
+	    Self::Single(single) => {
+		let first = single.take().unwrap();
+		*self = Self::Many(vec![first, value]);
+	    },
+	    Self::Many(ref mut many) => {
+		many.push(value);
+	    }
+	}
+    }
+
+    pub const fn single(from: T) -> Self
+    {
+	Self::Single(Some(from))
+    }
+}
+
+impl<T> FromIterator<T> for MaybeSingle<T>
+{
+    fn from_iter<I: IntoIterator<Item=T>>(iter: I) -> Self
+    {
+	let iter = iter.into_iter();
+	let mut vec = match iter.size_hint() {
+	    (0, None) => Vec::new(),
+	    (_, Some(value)) | (value, None) => Vec::with_capacity(value),  
+	};
+	vec.extend(iter);	
+	Self::Many(vec)
+    }
+}
+
+impl<T> ExactSizeIterator for IntoIter<T>{}
+
+impl<T> From<Vec<T>> for MaybeSingle<T>
+{
+    fn from(from: Vec<T>) -> Self
+    {
+	Self::Many(from)
+    }
+}
+
+impl<T> IntoIterator for MaybeSingle<T>
+{
+    type Item = T;
+    type IntoIter = IntoIter<T>;
+
+    fn into_iter(self) -> Self::IntoIter
+    {
+	match self {
+	    Self::Single(Some(single)) => IntoIter::Single(iter::once(single)),
+	    Self::Many(many) => IntoIter::Many(many.into_iter()),
+	    _ => panic!("Invalid state"),
+	}
+    }
+}
+
+impl<T> Iterator for IntoIter<T>
+{
+    type Item = T;
+    fn next(&mut self) -> Option<Self::Item>
+    {
+	match self {
+	    Self::Many(ref mut many) => many.next(),
+	    Self::Single(ref mut once) => once.next(),
+	}
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>)
+    {
+	match self {
+	    Self::Many(many) => many.size_hint(),
+	    Self::Single(single) => (0, Some(1)),
+	}
+    }
+}

src/progress.rs

@@ -29,6 +29,7 @@ use tokio::{
 	self,
 	JoinHandle,
     },
+    stream::StreamExt as _,
 };
 use futures::{
     future::{
@@ -42,14 +43,14 @@ use termprogress::{
 };
 
 #[derive(Debug)]
-struct Task
+pub struct Task
 {
     name: String,
     idx: oneshot::Sender<usize>,
 }
 
 #[derive(Debug)]
-enum CommandKind
+pub enum CommandKind
 {
     BumpHigh(usize),
     BumpLow(usize),
@@ -142,6 +143,115 @@ impl Future for TaskWaiter
     }
 }
 
+pub struct CommandBuilder<'a>
+{
+    sender: &'a mut ProgressSender,
+    commands: Option<maybe_single::MaybeSingle<CommandKind>>,
+}
+
+impl<'a> CommandBuilder<'a>
+{
+    fn send_command(&mut self, comm: CommandKind)
+    {
+	if let Some(ref mut commands) = self.commands {
+	    commands.push(comm);
+	} else {
+	    self.commands = Some(maybe_single::MaybeSingle::single(comm));
+	}
+    }
+
+    /// Add another Builder to this one
+    #[inline] pub fn chain<I: IntoIterator<Item = CommandKind>>(mut self, other: I) -> Self
+    {
+	self.extend(other);
+	self
+    }
+    
+    // Commands
+    /// Print line on the worker's progress bar
+    pub fn println(mut self, line: impl Into<String>) -> Self
+    {
+	self.send_command(CommandKind::PrintLine(line.into()));
+	self
+    }
+
+    /// Print error line on the worker's progress bar
+    pub fn eprintln(mut self, line: impl Into<String>) -> Self
+    {
+	self.send_command(CommandKind::PrintLineErr(line.into()));
+	self
+    }
+
+    /// Increase the worker's max number
+    pub fn bump_max(mut self, by: usize) -> Self
+    {
+	self.send_command(CommandKind::BumpHigh(by));
+	self
+    }
+
+    /// Increase the worker's min number
+    pub fn bump_min(mut self, by: usize) -> Self
+    {
+	self.send_command(CommandKind::BumpLow(by));
+	self
+    }
+
+    /// Remove a task by ID.
+    pub fn remove_task(mut self, task_idx: usize) -> Self
+    {
+	self.send_command(CommandKind::RemoveTask(task_idx));
+	self
+    }
+
+    /// Signal a shutdown to the worker
+    pub fn shutdown(mut self) -> Self
+    {
+	self.send_command(CommandKind::Complete);
+	self
+    }
+
+    /// Send this as an atom,
+    pub async fn send(self) -> Result<CommandWaiter, Error>
+    {
+	self.sender.send_command(match self.commands {
+	    Some(maybe_single::MaybeSingle::Single(Some(single))) => {
+		single
+	    },
+	    Some(maybe_single::MaybeSingle::Many(many)) => {
+		CommandKind::Many(many)
+	    },
+	    _ => return Err(Error::NoCommands),
+	}).await
+    }
+}
+
+impl<'a> Extend<CommandKind> for CommandBuilder<'a>
+{
+    fn extend<I: IntoIterator<Item = CommandKind>>(&mut self, other: I)
+    {
+	match self.commands {
+	    Some(ref mut commands) => commands.extend(other),
+	    _ => self.commands = Some(other.into_iter().collect()),
+	};
+    }	  
+}
+
+impl<'a> IntoIterator for CommandBuilder<'a>
+{
+    type Item= CommandKind;
+    type IntoIter = maybe_single::IntoIter<CommandKind>;
+
+    fn into_iter(self) -> Self::IntoIter
+    {
+	match self.commands {
+	    Some(value) => value.into_iter(),
+	    None => maybe_single::MaybeSingle::from(vec![]).into_iter(),
+	}
+    }
+}
+
+
+
 impl ProgressSender
 {
     /// Get the status of this progress
@@ -221,6 +331,15 @@ impl ProgressSender
     {
 	self.send_command(CommandKind::Complete).await
     }
+
+    /// Create a new command builder
+    pub fn builder<'a>(&'a mut self) -> CommandBuilder<'a>
+    {
+	CommandBuilder{
+	    commands: None,
+	    sender: self,
+	}
+    }
 }
 
 /// Status of the worker
@@ -273,29 +392,47 @@ pub fn create_progress<P: ProgressBar + WithTitle + Send + 'static,
 			(comm, d)
 		    };
 
-		    enum MaybeSingle
+		    enum MaybeSingle<T>
+		    where T: IntoIterator<Item = CommandKind>
 		    {
 			Single(Once<CommandKind>),
-			Many(Vec<CommandKind>),
+			Many(<T as IntoIterator>::IntoIter),
 		    }
 
-		    impl From<CommandKind> for MaybeSingle
+		    impl<T> ExactSizeIterator for MaybeSingle<T>
+		    where T: IntoIterator<Item = CommandKind>,
+		    <T as IntoIterator>::IntoIter: ExactSizeIterator{}
+		    
+		    impl<T> Iterator for MaybeSingle<T>
+		    where T: IntoIterator<Item = CommandKind>,
+		    <T as IntoIterator>::IntoIter: ExactSizeIterator
 		    {
-			fn from(from: CommandKind) -> Self
+			type Item = <T as IntoIterator>::Item;
+			fn next(&mut self) -> Option<Self::Item>
 			{
-			    match from {
+			    match self {
-				CommandKind::Many(many) => Self::Many(many),
+				Self::Many(ref mut many) => many.next(),
-				x => Self::Single(iter::once(x)),
+				Self::Single(ref mut single) => single.next(),
-			    }
 			    }
 			}
 
-		    //TODO: IntoIterator for Maybesingle??? We'll need our own type, I think. Maybe generalise it, and put in ext or util?
+			fn size_hint(&self) -> (usize, Option<usize>)
-
+			{
-		    let command = MaybeSingle::from(command);
+			    match &self {
+				Self::Many(many) => (many.len(), Some(many.len())),
+				Self::Single(_) => (1, Some(1)),
+			    }
+			}
+		    }
 		    
-		    // Match the command
+		    let mut commands: stage::Stage<CommandKind> = {
-		    for command in command.into_iter() {
+			let command: MaybeSingle<Vec<CommandKind>> = match command {
+			    CommandKind::Many(many) => MaybeSingle::Many(many.into_iter()),
+			    other => MaybeSingle::Single(iter::once(other)),
+			};
+			stage::Stage::from_iter(command)
+		    };
+		    while let Some(command) = commands.next().await {
 			match command {
 			    CommandKind::BumpHigh(high) => {
 				let stat = stat.to_mut();
@@ -329,6 +466,9 @@ pub fn create_progress<P: ProgressBar + WithTitle + Send + 'static,
 			    CommandKind::Complete => {
 				break;
 			    },
+			    CommandKind::Many(many) => {
+				let _ = commands.sender().send_many(many).await;
+			    },
 			}
 		    }
 
@@ -358,6 +498,7 @@ pub fn create_progress<P: ProgressBar + WithTitle + Send + 'static,
 #[derive(Debug,Clone)]
 pub enum Error
 {
+    NoCommands,
     WorkerPanic,
     WorkerDropped,
     Unknown,
@@ -369,6 +510,7 @@ impl std::fmt::Display for Error
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result
     {
 	match self {
+	    Error::NoCommands => write!(f, "an attempt was made to send 0 commands"),
 	    Error::WorkerPanic => write!(f, "worker panicked"),
 	    Error::WorkerDropped => write!(f, "tried to communicate with dropped worker"),
 	    _ => write!(f, "unknown error"),

src/stage.rs

@@ -38,6 +38,22 @@ impl<T> StageSender<T>
 	    }
 	}
     }
+
+    /// Send many values at once
+    pub async fn send_many<I: IntoIterator<Item=T>>(&self, values: I) -> Result<(), I>
+    {
+	loop {
+	    if let Some(internal) = self.internal.upgrade() {
+		let mut write = internal.write().await;
+		for item in values.into_iter() {
+		    write.push_back(item);
+		}
+		break Ok(());
+	    } else {
+		break Err(values);
+	    }
+	}
+    }
 }
 
 impl<T> Stage<T>