feat: standardized Controller and ControllerWorker

src/buffer/controller.rs

@@ -1,11 +1,10 @@
-use operational_transform::OperationSeq;
+use tonic::{transport::Channel, Status};
-use tonic::{transport::Channel, Status, Streaming, async_trait};
 use uuid::Uuid;
 
 use crate::{
 	ControllerWorker,
-	buffer::handle::{BufferHandle, OperationControllerEditor, OperationControllerWorker},
+	buffer::handle::{BufferHandle, OperationControllerWorker},
-	proto::{buffer_client::BufferClient, BufferPayload, RawOp, OperationRequest},
+	proto::{buffer_client::BufferClient, BufferPayload},
 };
 
 #[derive(Clone)]
@@ -75,45 +74,16 @@ impl BufferController {
 
 		let stream = self.client.attach(req).await?.into_inner();
 
-		let controller = OperationControllerWorker::new((self.clone(), stream), &content, path);
+		let controller = OperationControllerWorker::new(self.id().to_string(), &content, path);
 		let factory = controller.subscribe();
+		let client = self.client.clone();
 
 		tokio::spawn(async move {
 			tracing::debug!("buffer worker started");
-			controller.work().await;
+			controller.work(client, stream).await;
 			tracing::debug!("buffer worker stopped");
 		});
 
 		Ok(factory)
 	}
 }
-
-#[async_trait]
-impl OperationControllerEditor for (BufferController, Streaming<RawOp>) {
-	async fn edit(&mut self, path: String, op: OperationSeq) -> bool {
-		let req = OperationRequest {
-			hash: "".into(),
-			opseq: serde_json::to_string(&op).unwrap(),
-			path,
-			user: self.0.id().to_string(),
-		};
-		match self.0.client.edit(req).await {
-			Ok(_) => true,
-			Err(e) => {
-				tracing::error!("error sending edit: {}", e);
-				false
-			}
-		}
-	}
-
-	async fn recv(&mut self) -> Option<OperationSeq> {
-		match self.1.message().await {
-			Ok(Some(op)) => Some(serde_json::from_str(&op.opseq).unwrap()),
-			Ok(None) => None,
-			Err(e) => {
-				tracing::error!("could not receive edit from server: {}", e);
-				None
-			}
-		}
-	}
-}

src/buffer/handle.rs

@@ -2,10 +2,12 @@ use std::{sync::Arc, collections::VecDeque, ops::Range};
 
 use operational_transform::OperationSeq;
 use tokio::sync::{watch, mpsc, broadcast, Mutex};
-use tonic::async_trait;
+use tonic::transport::Channel;
+use tonic::{async_trait, Streaming};
 
-use crate::ControllerWorker;
+use crate::proto::{OperationRequest, RawOp};
-use crate::errors::IgnorableError;
+use crate::proto::buffer_client::BufferClient;
+use crate::{ControllerWorker, Controller, CodempError};
 use crate::buffer::factory::{leading_noop, tailing_noop, OperationFactory};
 
 pub struct TextChange {
@@ -26,53 +28,63 @@ impl OperationFactory for BufferHandle {
 	}
 }
 
-impl BufferHandle {
+#[async_trait]
-	async fn poll(&self) -> Option<TextChange> {
+impl Controller<TextChange> for BufferHandle {
-		let op = self.stream.lock().await.recv().await.ok()?;
+	type Input = OperationSeq;
+
+	async fn recv(&self) -> Result<TextChange, CodempError> {
+		let op = self.stream.lock().await.recv().await?;
 		let after = self.content.borrow().clone();
 		let skip = leading_noop(op.ops()) as usize; 
 		let before_len = op.base_len();
 		let tail = tailing_noop(op.ops()) as usize;
 		let span = skip..before_len-tail;
 		let content = after[skip..after.len()-tail].to_string();
-		Some(TextChange { span, content })
+		Ok(TextChange { span, content })
 	}
 
-	async fn apply(&self, op: OperationSeq) {
+	async fn send(&self, op: OperationSeq) -> Result<(), CodempError> {
-		self.operations.send(op).await
+		Ok(self.operations.send(op).await?)
-			.unwrap_or_warn("could not apply+send operation")
+	}
 }
 
-	// fn subscribe(&self) -> Self {
+pub(crate) struct OperationControllerWorker {
-	// 	OperationControllerHandle {
+	uid: String,
-	// 		content: self.content.clone(),
-	// 		operations: self.operations.clone(),
-	// 		original: self.original.clone(),
-	// 		stream: Arc::new(Mutex::new(self.original.subscribe())),
-	// 	}
-	// }
-}
-
-#[async_trait]
-pub(crate) trait OperationControllerEditor {
-	async fn edit(&mut self, path: String, op: OperationSeq) -> bool;
-	async fn recv(&mut self) -> Option<OperationSeq>;
-}
-
-pub(crate) struct OperationControllerWorker<C : OperationControllerEditor> {
 	pub(crate) content: watch::Sender<String>,
 	pub(crate) operations: mpsc::Receiver<OperationSeq>,
 	pub(crate) stream: Arc<broadcast::Sender<OperationSeq>>,
 	pub(crate) queue: VecDeque<OperationSeq>,
 	receiver: watch::Receiver<String>,
 	sender: mpsc::Sender<OperationSeq>,
-	client: C,
 	buffer: String,
 	path: String,
 }
 
+impl OperationControllerWorker {
+	pub fn new(uid: String, buffer: &str, path: &str) -> Self {
+		let (txt_tx, txt_rx) = watch::channel(buffer.to_string());
+		let (op_tx, op_rx) = mpsc::channel(64);
+		let (s_tx, _s_rx) = broadcast::channel(64);
+		OperationControllerWorker {
+			uid,
+			content: txt_tx,
+			operations: op_rx,
+			stream: Arc::new(s_tx),
+			receiver: txt_rx,
+			sender: op_tx,
+			queue: VecDeque::new(),
+			buffer: buffer.to_string(),
+			path: path.to_string(),
+		}
+	}
+}
+
 #[async_trait]
-impl<C : OperationControllerEditor + Send> ControllerWorker<BufferHandle> for OperationControllerWorker<C> {
+impl ControllerWorker<TextChange> for OperationControllerWorker {
+	type Controller = BufferHandle;
+	type Tx = BufferClient<Channel>;
+	type Rx = Streaming<RawOp>;
+
 	fn subscribe(&self) -> BufferHandle {
 		BufferHandle {
 			content: self.receiver.clone(),
@@ -81,10 +93,10 @@ impl<C : OperationControllerEditor + Send> ControllerWorker<BufferHandle> for Op
 		}
 	}
 
-	async fn work(mut self) {
+	async fn work(mut self, mut tx: Self::Tx, mut rx: Self::Rx) {
 		loop {
 			let op = tokio::select! {
-				Some(operation) = self.client.recv() => {
+				Some(operation) = recv_opseq(&mut rx) => {
 					let mut out = operation;
 					for op in self.queue.iter_mut() {
 						(*op, out) = op.transform(&out).unwrap();
@@ -102,29 +114,36 @@ impl<C : OperationControllerEditor + Send> ControllerWorker<BufferHandle> for Op
 			self.content.send(self.buffer.clone()).unwrap();
 
 			while let Some(op) = self.queue.get(0) {
-				if !self.client.edit(self.path.clone(), op.clone()).await { break }
+				if !send_opseq(&mut tx, self.uid.clone(), self.path.clone(), op.clone()).await { break }
 				self.queue.pop_front();
 			}
 		}
 	}
-
 }
 
-impl<C : OperationControllerEditor> OperationControllerWorker<C> {
+async fn send_opseq(tx: &mut BufferClient<Channel>, uid: String, path: String, op: OperationSeq) -> bool {
-	pub fn new(client: C, buffer: &str, path: &str) -> Self {
+	let req = OperationRequest {
-		let (txt_tx, txt_rx) = watch::channel(buffer.to_string());
+		hash: "".into(),
-		let (op_tx, op_rx) = mpsc::channel(64);
+		opseq: serde_json::to_string(&op).unwrap(),
-		let (s_tx, _s_rx) = broadcast::channel(64);
+		path,
-		OperationControllerWorker {
+		user: uid,
-			content: txt_tx,
+	};
-			operations: op_rx,
+	match tx.edit(req).await {
-			stream: Arc::new(s_tx),
+		Ok(_) => true,
-			receiver: txt_rx,
+		Err(e) => {
-			sender: op_tx,
+			tracing::error!("error sending edit: {}", e);
-			queue: VecDeque::new(),
+			false
-			buffer: buffer.to_string(),
+		}
-			path: path.to_string(),
+	}
-			client,
+}
+
+async fn recv_opseq(rx: &mut Streaming<RawOp>) -> Option<OperationSeq> {
+	match rx.message().await {
+		Ok(Some(op)) => Some(serde_json::from_str(&op.opseq).unwrap()),
+		Ok(None) => None,
+		Err(e) => {
+			tracing::error!("could not receive edit from server: {}", e);
+			None
 		}
 	}
 }

src/cursor/client.rs

@@ -1 +0,0 @@
-// TODO separate cursor movement from buffer operations in protocol!

src/cursor/tracker.rs

@@ -1,9 +1,9 @@
 use std::sync::Arc;
 
 use tokio::sync::{mpsc, broadcast::{self, error::RecvError}, Mutex};
-use tonic::{Streaming, transport::Channel};
+use tonic::{Streaming, transport::Channel, async_trait};
 
-use crate::{proto::{RowColumn, CursorPosition, buffer_client::BufferClient}, errors::IgnorableError, CodempError};
+use crate::{proto::{CursorPosition, cursor_client::CursorClient, RowColumn}, errors::IgnorableError, CodempError, Controller, ControllerWorker};
 
 pub struct CursorTracker {
 	uid: String,
@@ -11,19 +11,22 @@ pub struct CursorTracker {
 	stream: Mutex<broadcast::Receiver<CursorPosition>>,
 }
 
-impl CursorTracker {
+#[async_trait]
-	pub async fn moved(&self, path: &str, start: RowColumn, end: RowColumn) -> Result<(), CodempError> {
+impl Controller<CursorPosition> for CursorTracker {
+	type Input = (String, RowColumn, RowColumn);
+
+	async fn send(&self, (buffer, start, end): Self::Input) -> Result<(), CodempError> {
 		Ok(self.op.send(CursorPosition {
 			user: self.uid.clone(),
-			buffer: path.to_string(),
+			start: Some(start),
-			start: start.into(),
+			end: Some(end),
-			end: end.into(),
+			buffer,
 		}).await?)
 	}
 
 	// TODO is this cancelable? so it can be used in tokio::select!
 	// TODO is the result type overkill? should be an option?
-	pub async fn recv(&self) -> Result<CursorPosition, CodempError> {
+	async fn recv(&self) -> Result<CursorPosition, CodempError> {
 		let mut stream = self.stream.lock().await;
 		match stream.recv().await {
 			Ok(x) => Ok(x),
@@ -51,14 +54,14 @@ impl CursorTracker {
 	// }
 }
 
-pub(crate) struct CursorPositionTrackerWorker {
+pub(crate) struct CursorTrackerWorker {
 	uid: String,
 	producer: mpsc::Sender<CursorPosition>,
 	op: mpsc::Receiver<CursorPosition>,
 	channel: Arc<broadcast::Sender<CursorPosition>>,
 }
 
-impl CursorPositionTrackerWorker {
+impl CursorTrackerWorker {
 	pub(crate) fn new(uid: String) -> Self {
 		let (op_tx, op_rx) = mpsc::channel(64);
 		let (cur_tx, _cur_rx) = broadcast::channel(64);
@@ -69,8 +72,15 @@ impl CursorPositionTrackerWorker {
 			channel: Arc::new(cur_tx),
 		}
 	}
+}
 
-	pub(crate) fn subscribe(&self) -> CursorTracker {
+#[async_trait]
+impl ControllerWorker<CursorPosition> for CursorTrackerWorker {
+	type Controller = CursorTracker;
+	type Tx = CursorClient<Channel>;
+	type Rx = Streaming<CursorPosition>;
+
+	fn subscribe(&self) -> CursorTracker {
 		CursorTracker {
 			uid: self.uid.clone(),
 			op: self.producer.clone(),
@@ -78,12 +88,11 @@ impl CursorPositionTrackerWorker {
 		}
 	}
 
-	// TODO is it possible to avoid passing directly tonic Streaming and proto BufferClient ?
+	async fn work(mut self, mut tx: Self::Tx, mut rx: Self::Rx) {
-	pub(crate) async fn work(mut self, mut rx: Streaming<CursorPosition>, mut tx: BufferClient<Channel>) {
 		loop {
 			tokio::select!{
 				Ok(Some(cur)) = rx.message() => self.channel.send(cur).unwrap_or_warn("could not broadcast event"),
-				Some(op) = self.op.recv() => { todo!() } // tx.moved(op).await.unwrap_or_warn("could not update cursor"); },
+				Some(op) = self.op.recv() => { tx.moved(op).await.unwrap_or_warn("could not update cursor"); },
 				else => break,
 			}
 		}

src/errors.rs

@@ -1,6 +1,6 @@
 use std::{error::Error, fmt::Display};
 
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, broadcast};
 use tonic::{Status, Code};
 use tracing::warn;
 
@@ -66,3 +66,9 @@ impl<T> From<mpsc::error::SendError<T>> for CodempError {
 		CodempError::Channel { send: true }
 	}
 }
+
+impl From<broadcast::error::RecvError> for CodempError {
+	fn from(_value: broadcast::error::RecvError) -> Self {
+		CodempError::Channel { send: false }
+	}
+}

src/lib.rs

@@ -3,6 +3,7 @@ pub mod errors;
 pub mod buffer;
 
 pub mod state;
+pub mod client;
 
 pub use tonic;
 pub use tokio;
@@ -19,12 +20,18 @@ pub use errors::CodempError;
 
 #[tonic::async_trait] // TODO move this somewhere?
 pub(crate) trait ControllerWorker<T> {
-	fn subscribe(&self) -> T;
+	type Controller : Controller<T>;
-	async fn work(self);
+	type Tx;
+	type Rx;
+
+	fn subscribe(&self) -> Self::Controller;
+	async fn work(self, tx: Self::Tx, rx: Self::Rx);
 }
 
 #[tonic::async_trait]
 pub trait Controller<T> {
+	type Input;
+
+	async fn send(&self, x: Self::Input) -> Result<(), CodempError>;
 	async fn recv(&self) -> Result<T, CodempError>;
-	async fn send(&self, x: T) -> Result<(), CodempError>;
 }