feat: major restructure, workspace rework, tweaks

all controllers use internal mutability so that they can all be put
behind Arcs

client/vscode/src/lib.rs

@@ -3,9 +3,13 @@ use std::sync::Arc;
 use neon::prelude::*;
 use once_cell::sync::OnceCell;
 use codemp::{
-	controller::{cursor::{CursorSubscriber, CursorControllerHandle}, buffer::{OperationControllerHandle, OperationControllerSubscriber}},
+	cursor::controller::{CursorSubscriber, CursorControllerHandle},
+	buffer::{
+		controller::{OperationControllerHandle, OperationControllerSubscriber},
 		client::CodempClient,
-	proto::buffer_client::BufferClient, factory::OperationFactory,
+		factory::OperationFactory,
+	},
+	proto::buffer_client::BufferClient, 
 };
 use codemp::tokio::{runtime::Runtime, sync::Mutex};
 
@@ -75,7 +79,7 @@ fn create_client(mut cx: FunctionContext) -> JsResult<JsPromise> {
 	let channel = cx.channel();
 
 	runtime(&mut cx)?.spawn(async move {
-		match rc.lock().await.create(path, content).await {
+		match rc.lock().await.create(&path, content.as_deref()).await {
 			Ok(accepted) => deferred.settle_with(&channel, move |mut cx| Ok(cx.boolean(accepted))),
 			Err(e) => deferred.settle_with(&channel, move |mut cx| cx.throw_error::<String, neon::handle::Handle<JsString>>(e.to_string())),
 		}
@@ -123,7 +127,7 @@ fn attach_client(mut cx: FunctionContext) -> JsResult<JsPromise> {
 	let channel = cx.channel();
 
 	runtime(&mut cx)?.spawn(async move {
-		match rc.lock().await.attach(path).await {
+		match rc.lock().await.attach(&path).await {
 			Ok(controller) => {
 				deferred.settle_with(&channel, move |mut cx| {
 					let obj = cx.empty_object();
@@ -183,7 +187,7 @@ fn callback_operation(mut cx: FunctionContext) -> JsResult<JsUndefined> {
 	let boxed : Handle<JsBox<OperationControllerJs>> = this.get(&mut cx, "boxed")?;
 	let callback = Arc::new(cx.argument::<JsFunction>(0)?.root(&mut cx));
 
-	let mut rc = boxed.0.clone();
+	let rc = boxed.0.clone();
 	let channel = cx.channel();
 
 	// TODO when garbage collecting OperationController stop this worker
@@ -213,7 +217,7 @@ fn callback_cursor(mut cx: FunctionContext) -> JsResult<JsUndefined> {
 	let boxed : Handle<JsBox<CursorEventsHandle>> = this.get(&mut cx, "boxed")?;
 	let callback = Arc::new(cx.argument::<JsFunction>(0)?.root(&mut cx));
 
-	let mut rc = boxed.0.clone();
+	let rc = boxed.0.clone();
 	let channel = cx.channel();
 
 	// TODO when garbage collecting OperationController stop this worker

src/buffer/client.rs

@@ -3,10 +3,9 @@ use tonic::{transport::Channel, Status, Streaming, async_trait};
 use uuid::Uuid;
 
 use crate::{
-	controller::{ControllerWorker,
+	ControllerWorker,
-		cursor::{CursorControllerHandle, CursorControllerWorker, CursorEditor},
+	cursor::controller::{CursorControllerHandle, CursorControllerWorker, CursorEditor},
-		buffer::{OperationControllerHandle, OperationControllerEditor, OperationControllerWorker}
+	buffer::controller::{OperationControllerHandle, OperationControllerEditor, OperationControllerWorker},
-	},
 	proto::{buffer_client::BufferClient, BufferPayload, RawOp, OperationRequest, Cursor},
 };
 
@@ -29,9 +28,10 @@ impl CodempClient {
 
 	pub fn id(&self) -> &str { &self.id }
 
-	pub async fn create(&mut self, path: String, content: Option<String>) -> Result<bool, Status> {
+	pub async fn create(&mut self, path: &str, content: Option<&str>) -> Result<bool, Status> {
 		let req = BufferPayload {
-			path, content,
+			path: path.to_string(),
+			content: content.map(|x| x.to_string()),
 			user: self.id.clone(),
 		};
 
@@ -61,9 +61,9 @@ impl CodempClient {
 		Ok(handle)
 	}
 
-	pub async fn attach(&mut self, path: String) -> Result<OperationControllerHandle, Status> {
+	pub async fn attach(&mut self, path: &str) -> Result<OperationControllerHandle, Status> {
 		let req = BufferPayload {
-			path: path.clone(),
+			path: path.to_string(),
 			content: None,
 			user: self.id.clone(),
 		};
@@ -76,7 +76,7 @@ impl CodempClient {
 
 		let stream = self.client.attach(req).await?.into_inner();
 
-		let controller = OperationControllerWorker::new((self.clone(), stream), content, path);
+		let controller = OperationControllerWorker::new((self.clone(), stream), &content, path);
 		let factory = controller.subscribe();
 
 		tokio::spawn(async move {

src/buffer/controller.rs

@@ -1,12 +1,12 @@
 use std::{sync::Arc, collections::VecDeque, ops::Range};
 
 use operational_transform::OperationSeq;
-use tokio::sync::{watch, mpsc, broadcast};
+use tokio::sync::{watch, mpsc, broadcast, Mutex};
 use tonic::async_trait;
 
-use super::{leading_noop, tailing_noop, ControllerWorker};
+use crate::ControllerWorker;
 use crate::errors::IgnorableError;
-use crate::factory::OperationFactory;
+use crate::buffer::factory::{leading_noop, tailing_noop, OperationFactory};
 
 pub struct TextChange {
 	pub span: Range<usize>,
@@ -15,26 +15,16 @@ pub struct TextChange {
 
 #[async_trait]
 pub trait OperationControllerSubscriber {
-	async fn poll(&mut self) -> Option<TextChange>;
+	async fn poll(&self) -> Option<TextChange>;
 	async fn apply(&self, op: OperationSeq);
 }
 
+#[derive(Clone)]
 pub struct OperationControllerHandle {
 	content: watch::Receiver<String>,
 	operations: mpsc::Sender<OperationSeq>,
 	original: Arc<broadcast::Sender<OperationSeq>>,
-	stream: broadcast::Receiver<OperationSeq>,
+	stream: Arc<Mutex<broadcast::Receiver<OperationSeq>>>,
-}
-
-impl Clone for OperationControllerHandle {
-	fn clone(&self) -> Self {
-		OperationControllerHandle {
-			content: self.content.clone(),
-			operations: self.operations.clone(),
-			original: self.original.clone(),
-			stream: self.original.subscribe(),
-		}
-	}
 }
 
 #[async_trait]
@@ -46,8 +36,8 @@ impl OperationFactory for OperationControllerHandle {
 
 #[async_trait]
 impl OperationControllerSubscriber for OperationControllerHandle {
-	async fn poll(&mut self) -> Option<TextChange> {
+	async fn poll(&self) -> Option<TextChange> {
-		let op = self.stream.recv().await.ok()?;
+		let op = self.stream.lock().await.recv().await.ok()?;
 		let after = self.content.borrow().clone();
 		let skip = leading_noop(op.ops()) as usize; 
 		let before_len = op.base_len();
@@ -61,6 +51,15 @@ impl OperationControllerSubscriber for OperationControllerHandle {
 		self.operations.send(op).await
 			.unwrap_or_warn("could not apply+send operation")
 	}
+
+	// fn subscribe(&self) -> Self {
+	// 	OperationControllerHandle {
+	// 		content: self.content.clone(),
+	// 		operations: self.operations.clone(),
+	// 		original: self.original.clone(),
+	// 		stream: Arc::new(Mutex::new(self.original.subscribe())),
+	// 	}
+	// }
 }
 
 #[async_trait]
@@ -88,7 +87,7 @@ impl<C : OperationControllerEditor + Send> ControllerWorker<OperationControllerH
 			content: self.receiver.clone(),
 			operations: self.sender.clone(),
 			original: self.stream.clone(),
-			stream: self.stream.subscribe(),
+			stream: Arc::new(Mutex::new(self.stream.subscribe())),
 		}
 	}
 
@@ -122,8 +121,8 @@ impl<C : OperationControllerEditor + Send> ControllerWorker<OperationControllerH
 }
 
 impl<C : OperationControllerEditor> OperationControllerWorker<C> {
-	pub fn new(client: C, buffer: String, path: String) -> Self {
+	pub fn new(client: C, buffer: &str, path: &str) -> Self {
-		let (txt_tx, txt_rx) = watch::channel(buffer.clone());
+		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 {
@@ -133,7 +132,9 @@ impl<C : OperationControllerEditor> OperationControllerWorker<C> {
 			receiver: txt_rx,
 			sender: op_tx,
 			queue: VecDeque::new(),
-			client, buffer, path
+			buffer: buffer.to_string(),
+			path: path.to_string(),
+			client,
 		}
 	}
 }

src/buffer/factory.rs

@@ -1,6 +1,25 @@
-use operational_transform::OperationSeq;
+use std::ops::Range;
+
+use operational_transform::{OperationSeq, Operation};
 use similar::{TextDiff, ChangeTag};
 
+pub const fn leading_noop(seq: &[Operation]) -> u64 { count_noop(seq.first()) }
+pub const fn tailing_noop(seq: &[Operation]) -> u64 { count_noop(seq.last())  }
+
+const fn count_noop(op: Option<&Operation>) -> u64 {
+	match op {
+		None => 0,
+		Some(Operation::Retain(n)) => *n,
+		Some(_) => 0,
+	}
+}
+
+pub fn op_effective_range(op: &OperationSeq) -> Range<u64> {
+	let first = leading_noop(op.ops());
+	let last = op.base_len() as u64 - tailing_noop(op.ops());
+	first..last
+}
+
 pub trait OperationFactory {
 	fn content(&self) -> String;
 

src/buffer/mod.rs

@@ -0,0 +1,3 @@
+pub mod factory;
+pub mod client;
+pub mod controller;

src/controller/mod.rs

@@ -1,30 +0,0 @@
-pub mod buffer;
-pub mod cursor;
-
-use std::ops::Range;
-
-use operational_transform::{Operation, OperationSeq};
-use tonic::async_trait;
-
-#[async_trait]
-pub trait ControllerWorker<T> {
-	fn subscribe(&self) -> T;
-	async fn work(self);
-}
-
-pub const fn leading_noop(seq: &[Operation]) -> u64 { count_noop(seq.first()) }
-pub const fn tailing_noop(seq: &[Operation]) -> u64 { count_noop(seq.last())  }
-
-const fn count_noop(op: Option<&Operation>) -> u64 {
-	match op {
-		None => 0,
-		Some(Operation::Retain(n)) => *n,
-		Some(_) => 0,
-	}
-}
-
-pub fn op_effective_range(op: &OperationSeq) -> Range<u64> {
-	let first = leading_noop(op.ops());
-	let last = op.base_len() as u64 - tailing_noop(op.ops());
-	first..last
-}

src/cursor/controller.rs

@@ -1,30 +1,31 @@
 use std::sync::Arc;
 
-use tokio::sync::{mpsc, broadcast};
+use tokio::sync::{mpsc, broadcast::{self, error::{TryRecvError, RecvError}}, Mutex};
 use tonic::async_trait;
 
-use crate::{proto::{Position, Cursor}, errors::IgnorableError, controller::ControllerWorker};
+use crate::{proto::{Position, Cursor}, errors::IgnorableError, ControllerWorker};
 
 #[async_trait]
 pub trait CursorSubscriber {
 	async fn send(&self, path: &str, start: Position, end: Position);
-	async fn poll(&mut self) -> Option<Cursor>;
+	async fn poll(&self) -> Option<Cursor>;
+	fn try_poll(&self) -> Option<Option<Cursor>>; // TODO fuck this fuck neovim
 }
 
 pub struct CursorControllerHandle {
 	uid: String,
 	op: mpsc::Sender<Cursor>,
-	stream: broadcast::Receiver<Cursor>,
 	original: Arc<broadcast::Sender<Cursor>>,
+	stream: Mutex<broadcast::Receiver<Cursor>>,
 }
 
 impl Clone for CursorControllerHandle {
 	fn clone(&self) -> Self {
-		Self {
+		CursorControllerHandle {
 			uid: self.uid.clone(),
 			op: self.op.clone(),
-			stream: self.original.subscribe(),
+			original: self.original.clone(),
-			original: self.original.clone()
+			stream: Mutex::new(self.original.subscribe()),
 		}
 	}
 }
@@ -40,12 +41,30 @@ impl CursorSubscriber for CursorControllerHandle {
 		}).await.unwrap_or_warn("could not send cursor op")
 	}
 
-	async fn poll(&mut self) -> Option<Cursor> {
+	// TODO is this cancelable? so it can be used in tokio::select!
-		match self.stream.recv().await {
+	async fn poll(&self) -> Option<Cursor> {
+		let mut stream = self.stream.lock().await;
+		match stream.recv().await {
 			Ok(x) => Some(x),
-			Err(e) => {
+			Err(RecvError::Closed) => None,
-				tracing::warn!("could not poll for cursor: {}", e);
+			Err(RecvError::Lagged(n)) => {
-				None
+				tracing::error!("cursor channel lagged behind, skipping {} events", n);
+				Some(stream.recv().await.expect("could not receive after lagging"))
+			}
+		}
+	}
+
+	fn try_poll(&self) -> Option<Option<Cursor>> {
+		match self.stream.try_lock() {
+			Err(_) => None,
+			Ok(mut x) => match x.try_recv() {
+				Ok(x) => Some(Some(x)),
+				Err(TryRecvError::Empty) => None,
+				Err(TryRecvError::Closed) => Some(None),
+				Err(TryRecvError::Lagged(n)) => {
+					tracing::error!("cursor channel lagged behind, skipping {} events", n);
+					Some(Some(x.try_recv().expect("could not receive after lagging")))
+				}
 			}
 		}
 	}
@@ -84,8 +103,8 @@ impl<C : CursorEditor + Send> ControllerWorker<CursorControllerHandle> for Curso
 		CursorControllerHandle {
 			uid: self.uid.clone(),
 			op: self.producer.clone(),
-			stream: self.channel.subscribe(),
 			original: self.channel.clone(),
+			stream: Mutex::new(self.channel.subscribe()),
 		}
 	}
 
@@ -98,7 +117,6 @@ impl<C : CursorEditor + Send> ControllerWorker<CursorControllerHandle> for Curso
 			}
 		}
 	}
-
 }
 
 

src/cursor/mod.rs

@@ -1,3 +1,5 @@
+pub mod controller;
+
 use crate::proto::{Position, Cursor};
 
 impl From::<Position> for (i32, i32) {

src/errors.rs

@@ -1,3 +1,6 @@
+use std::{error::Error, fmt::Display};
+
+use tonic::{Status, Code};
 use tracing::warn;
 
 pub trait IgnorableError {
@@ -13,3 +16,44 @@ where E : std::fmt::Display {
 		}
 	}
 }
+
+// TODO split this into specific errors for various parts of the library
+#[derive(Debug)]
+pub enum CodempError {
+	Transport {
+		status: Code,
+		message: String,
+	},
+	Channel { },
+
+	// TODO filler error, remove later
+	Filler {
+		message: String,
+	},
+}
+
+impl Error for CodempError {}
+
+impl Display for CodempError {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		match self {
+			Self::Transport { status, message } => write!(f, "Transport error: ({}) {}", status, message),
+			Self::Channel { } => write!(f, "Channel error"),
+			_ => write!(f, "Unknown error"),
+		}
+	}
+}
+
+impl From<Status> for CodempError {
+	fn from(status: Status) -> Self {
+		CodempError::Transport { status: status.code(), message: status.message().to_string() }
+	}
+}
+
+impl From<tonic::transport::Error> for CodempError {
+	fn from(err: tonic::transport::Error) -> Self {
+		CodempError::Transport {
+			status: Code::Unknown, message: format!("underlying transport error: {:?}", err)
+		}
+	}
+}

src/lib.rs

@@ -1,12 +1,20 @@
-pub mod proto;
-pub mod client;
 pub mod workspace;
-pub mod controller;
 pub mod cursor;
 pub mod errors;
-pub mod factory;
+pub mod buffer;
 
 pub use tonic;
 pub use tokio;
 pub use operational_transform as ot;
 
+use tonic::async_trait;
+
+#[async_trait] // TODO move this somewhere?
+pub trait ControllerWorker<T> {
+	fn subscribe(&self) -> T;
+	async fn work(self);
+}
+
+pub mod proto {
+	tonic::include_proto!("buffer");
+}

src/proto.rs

@@ -1 +0,0 @@
-tonic::include_proto!("buffer");

src/workspace.rs

@@ -1,57 +0,0 @@
-use std::{collections::BTreeMap, ops::Range};
-
-use tokio::sync::RwLock;
-use tonic::Status;
-
-use crate::{
-	client::CodempClient, factory::OperationFactory,
-	controller::{buffer::{OperationControllerHandle, OperationControllerSubscriber},
-	cursor::{CursorControllerHandle, CursorSubscriber}}, proto::Cursor,
-};
-
-pub struct Workspace {
-	client: CodempClient,
-	buffers: RwLock<BTreeMap<String, OperationControllerHandle>>,
-	cursor: RwLock<CursorControllerHandle>,
-}
-
-impl Workspace {
-	pub async fn new(mut client: CodempClient) -> Result<Self, Status> {
-		Ok(
-			Workspace {
-				cursor: RwLock::new(client.listen().await?),
-				client,
-				buffers: RwLock::new(BTreeMap::new()),
-			}
-		)
-	}
-
-	pub async fn create(&self, path: &str, content: Option<String>) -> Result<bool, Status> {
-		self.client.clone().create(path.into(), content).await
-	}
-
-	pub async fn attach(&self, path: String) -> Result<(), Status> {
-		self.buffers.write().await.insert(
-			path.clone(),
-			self.client.clone().attach(path).await?
-		);
-		Ok(())
-	}
-
-	pub async fn diff(&self, path: &str, span: Range<usize>, text: &str) {
-		if let Some(controller) = self.buffers.read().await.get(path) {
-			if let Some(op) = controller.delta(span.start, text, span.end) {
-				controller.apply(op).await
-			}
-		}
-	}
-
-	pub async fn send(&self, path: &str, start: (i32, i32), end: (i32, i32)) {
-		self.cursor.read().await.send(path, start.into(), end.into()).await
-	}
-
-	pub async fn recv(&self) -> Option<Cursor> {
-		self.cursor.write().await.poll().await
-
-	}
-}

src/workspace/controller.rs

@@ -0,0 +1,79 @@
+use std::{collections::BTreeMap, sync::Arc};
+
+use tokio::sync::RwLock;
+use tonic::async_trait;
+
+use crate::{
+	buffer::{client::CodempClient, controller::OperationControllerSubscriber},
+	cursor::controller::CursorSubscriber, errors::CodempError,
+};
+
+pub struct Workspace {
+	client: CodempClient,
+	buffers: RwLock<BTreeMap<Box<str>, BufferController>>,
+	cursor: CursorController,
+}
+
+pub type CursorController = Arc<dyn CursorSubscriber + Send + Sync>;
+pub type BufferController = Arc<dyn OperationControllerSubscriber + Send + Sync>;
+
+#[async_trait]
+pub trait WorkspaceHandle {
+	fn cursor(&self) -> CursorController;
+	async fn buffer(&self, path: &str) -> Option<BufferController>;
+	async fn attach(&self, path: &str) -> Result<(), CodempError>;
+	async fn create(&self, path: &str, content: Option<&str>) -> Result<bool, CodempError>;
+}
+
+impl Workspace {
+	pub async fn new(dest: &str) -> Result<Self, CodempError> {
+		let mut client = CodempClient::new(dest).await?;
+		let cursor = Arc::new(client.listen().await?);
+		Ok(
+			Workspace {
+				buffers: RwLock::new(BTreeMap::new()),
+				cursor,
+				client,
+			}
+		)
+	}
+}
+
+#[async_trait]
+impl WorkspaceHandle for Workspace {
+	// Cursor
+	fn cursor(&self) -> CursorController {
+		self.cursor.clone()
+	}
+
+	// Buffer
+	async fn buffer(&self, path: &str) -> Option<BufferController> {
+		self.buffers.read().await.get(path).cloned()
+	}
+
+	async fn create(&self, path: &str, content: Option<&str>) -> Result<bool, CodempError> {
+		Ok(self.client.clone().create(path, content).await?)
+	}
+
+	async fn attach(&self, path: &str) -> Result<(), CodempError> {
+		let controller = self.client.clone().attach(path).await?;
+		self.buffers.write().await.insert(path.into(), Arc::new(controller));
+		Ok(())
+	}
+
+	// pub async fn diff(&self, path: &str, span: Range<usize>, text: &str) {
+	// 	if let Some(controller) = self.inner.read().await.buffers.get(path) {
+	// 		if let Some(op) = controller.delta(span.start, text, span.end) {
+	// 			controller.apply(op).await
+	// 		}
+	// 	}
+	// }
+
+	// async fn send(&self, path: &str, start: (i32, i32), end: (i32, i32)) {
+	// 	self.inner.read().await.cursor.send(path, start.into(), end.into()).await
+	// }
+
+	// pub async fn recv(&self) -> Option<Cursor> {
+	// 	self.inner.write().await.cursor.poll().await
+	// }
+}

src/workspace/mod.rs

@@ -0,0 +1 @@
+pub mod controller;