feat: process indexing (#559)

Co-authored-by: David Yamnitsky <[email protected]>

Author: isabella

packages/cli/src/progress.rs

@@ -152,7 +152,6 @@ impl<T> State<T> {
 			.unwrap()
 			.as_millis();
 		for indicator in self.indicators.values() {
-			const LENGTH: u64 = 20;
 			const SPINNER: [char; 10] = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];
 			let mut line = String::new();
 			let position = (now / (1000 / 10)) % 10;
@@ -166,19 +165,19 @@ impl<T> State<T> {
 				title_length = title_length.unwrap(),
 			)
 			.unwrap();
+			const LENGTH: u64 = 20;
 			if let (Some(current), Some(total)) = (indicator.current, indicator.total) {
-				let current = current.min(total);
 				write!(line, " [").unwrap();
-				let last = current * LENGTH / total;
-				for _ in 0..last {
+				let n = if total > 0 { current / total } else { 1 } * LENGTH;
+				for _ in 0..n {
 					write!(line, "=").unwrap();
 				}
 				if current < total {
 					write!(line, ">").unwrap();
 				} else {
 					write!(line, "=").unwrap();
 				}
-				for _ in last..LENGTH {
+				for _ in n..LENGTH {
 					write!(line, " ").unwrap();
 				}
 				write!(line, "]").unwrap();

packages/messenger/src/memory.rs

@@ -105,7 +105,7 @@ impl Messenger {
 		payload: Bytes,
 	) -> Result<impl Future<Output = Result<StreamPublishInfo, Error>>, Error> {
 		let stream = self.streams.get(&name).ok_or(Error::NotFound)?.clone();
-		let future = async move { stream.publish(payload).await };
+		let future = stream.publish(payload).boxed();
 		Ok(future)
 	}