hexedtech · alemidev · Feb 19, 2025 · Feb 19, 2025
diff --git a/src/api/controller.rs b/src/api/controller.rs
@@ -39,7 +39,7 @@ where
 /// Details about the receiving end are left to the implementor.
 pub trait AsyncSender<T: Sized + Send + Sync>: Sized + Send + Sync {
 	/// Enqueue a new value to be sent to all other users without blocking
-	fn send(&self, x: T) -> ControllerResult<()>;
+	fn send(&self, x: T) -> ControllerResult<impl std::future::Future<Output = bool>>;
 }
 
 /// Asynchronous and thread-safe handle to receive data from a stream.

diff --git a/src/buffer/controller.rs b/src/buffer/controller.rs
@@ -58,7 +58,7 @@ pub(crate) struct BufferControllerInner {
 	pub(crate) path: String,
 	pub(crate) latest_version: watch::Receiver<diamond_types::LocalVersion>,
 	pub(crate) local_version: watch::Receiver<diamond_types::LocalVersion>,
-	pub(crate) ops_in: mpsc::UnboundedSender<TextChange>,
+	pub(crate) ops_in: mpsc::UnboundedSender<(TextChange, oneshot::Sender<bool>)>,
 	pub(crate) poller: mpsc::UnboundedSender<oneshot::Sender<()>>,
 	pub(crate) content_request: mpsc::Sender<oneshot::Sender<String>>,
 	pub(crate) delta_request: mpsc::Sender<oneshot::Sender<Option<BufferUpdate>>>,
@@ -71,9 +71,10 @@ pub(crate) struct BufferControllerInner {
 impl Controller<TextChange, BufferUpdate> for BufferController {}
 
 impl AsyncSender<TextChange> for BufferController {
-	fn send(&self, op: TextChange) -> ControllerResult<()> {
-		self.0.ops_in.send(op)?;
-		Ok(())
+	fn send(&self, op: TextChange) -> ControllerResult<impl std::future::Future<Output = bool>> {
+		let (tx, rx) = oneshot::channel();
+		self.0.ops_in.send((op, tx))?;
+		Ok(async move { rx.await.unwrap_or(false) })
 	}
 }
 

diff --git a/src/buffer/worker.rs b/src/buffer/worker.rs
@@ -23,7 +23,7 @@ struct BufferWorker {
 	latest_version: watch::Sender<diamond_types::LocalVersion>,
 	local_version: watch::Sender<diamond_types::LocalVersion>,
 	ack_rx: mpsc::UnboundedReceiver<LocalVersion>,
-	ops_in: mpsc::UnboundedReceiver<TextChange>,
+	ops_in: mpsc::UnboundedReceiver<(TextChange, oneshot::Sender<bool>)>,
 	poller: mpsc::UnboundedReceiver<oneshot::Sender<()>>,
 	pollers: Vec<oneshot::Sender<()>>,
 	content_checkout: mpsc::Receiver<oneshot::Sender<String>>,
@@ -133,7 +133,7 @@ impl BufferController {
 				// received a text change from editor
 				res = worker.ops_in.recv() => match res {
 					None => break tracing::debug!("stopping: editor closed channel"),
-					Some(change) => worker.handle_editor_change(change, &tx).await,
+					Some((change, sent)) => worker.handle_editor_change(change, sent, &tx).await,
 				},
 
 				// received a message from server: add to oplog and update latest version (+unlock pollers)
@@ -170,7 +170,7 @@ impl BufferController {
 
 impl BufferWorker {
 	#[tracing::instrument(skip(self, tx))]
-	async fn handle_editor_change(&mut self, change: TextChange, tx: &mpsc::Sender<Operation>) {
+	async fn handle_editor_change(&mut self, change: TextChange, sent: oneshot::Sender<bool>, tx: &mpsc::Sender<Operation>) {
 		let last_ver = self.oplog.local_version();
 		// clip to buffer extents
    		let clip_start = change.start_idx as usize;
@@ -199,14 +199,17 @@ impl BufferWorker {
 			tx.send(Operation {
 				data: self.oplog.encode_from(ENCODE_PATCH, &last_ver),
 			})
-			.await
-			.unwrap_or_warn("failed to send change!");
+				.await
+				.unwrap_or_warn("failed to send change!");
 			self.latest_version
 				.send(self.oplog.local_version())
 				.unwrap_or_warn("failed to update latest version!");
 			self.local_version
 				.send(self.branch.local_version())
 				.unwrap_or_warn("failed to update local version!");
+			let _ = sent.send(true);
+		} else {
+			let _ = sent.send(false);
 		}
 	}
 

diff --git a/src/cursor/controller.rs b/src/cursor/controller.rs
@@ -45,15 +45,15 @@ impl Controller<Selection, Cursor> for CursorController {}
 
 #[cfg_attr(feature = "async-trait", async_trait::async_trait)]
 impl AsyncSender<Selection> for CursorController {
-	fn send(&self, mut cursor: Selection) -> ControllerResult<()> {
+	fn send(&self, mut cursor: Selection) -> ControllerResult<impl std::future::Future<Output = bool>> {
 		if cursor.start_row > cursor.end_row
 			|| (cursor.start_row == cursor.end_row && cursor.start_col > cursor.end_col)
 		{
 			std::mem::swap(&mut cursor.start_row, &mut cursor.end_row);
 			std::mem::swap(&mut cursor.start_col, &mut cursor.end_col);
 		}
 
-		Ok(self.0.op.send(CursorPosition {
+		self.0.op.send(CursorPosition {
 			buffer: BufferNode {
 				path: cursor.buffer,
 			},
@@ -65,7 +65,9 @@ impl AsyncSender<Selection> for CursorController {
 				row: cursor.end_row,
 				col: cursor.end_col,
 			},
-		})?)
+		})?;
+
+		Ok(std::future::ready(true))
 	}
 }