fix: one factory per buffer, create on attach

src/client/nvim/main.rs

@@ -104,40 +104,19 @@ impl Handler for NeovimHandler {
 				}
 			},
 
-			"sync" => {
-				if args.len() < 1 {
-					return Err(Value::from("no path given"));
-				}
-				let path = default_empty_str(&args, 0);
-
-				let mut c = self.client.clone();
-				match c.sync(path).await {
-					Err(e) => Err(Value::from(format!("could not sync: {}", e))),
-					Ok(content) => match nvim.get_current_buf().await {
-						Err(e) => return Err(Value::from(format!("could not get current buffer: {}", e))),
-						Ok(b) => {
-							let lines : Vec<String> = content.split("\n").map(|x| x.to_string()).collect();
-							match b.set_lines(0, -1, false, lines).await {
-								Err(e) => Err(Value::from(format!("failed sync: {}", e))),
-								Ok(()) => Ok(Value::from("synched")),
-							}
-						},
-					},
-				}
-			}
-
 			"attach" => {
 				if args.len() < 1 {
 					return Err(Value::from("no path given"));
 				}
 				let path = default_empty_str(&args, 0);
-				let buf = match nvim.get_current_buf().await {
+				let buffer = match nvim.get_current_buf().await {
 					Ok(b) => b,
 					Err(e) => return Err(Value::from(format!("could not get current buffer: {}", e))),
 				};
 
 				let mut c = self.client.clone();
 
+				let buf = buffer.clone();
 				match c.attach(path, move |x| {
 					let lines : Vec<String> = x.split("\n").map(|x| x.to_string()).collect();
 					let b = buf.clone();
@@ -147,8 +126,14 @@ impl Handler for NeovimHandler {
 						}
 					});
 				}).await {
-					Ok(()) => Ok(Value::from("spawned worker")),
 					Err(e) => Err(Value::from(format!("could not attach to stream: {}", e))),
+					Ok(content) => {
+						let lines : Vec<String> = content.split("\n").map(|x| x.to_string()).collect();
+						if let Err(e) = buffer.set_lines(0, -1, false, lines).await {
+							error!("could not update buffer: {}", e);
+						}
+						Ok(Value::from("spawned worker"))
+					},
 				}
 			},
 

src/lib/client.rs

@@ -1,6 +1,6 @@
 /// TODO better name for this file
 
-use std::sync::Arc;
+use std::{sync::{Arc, RwLock}, collections::BTreeMap};
 use tracing::{error, warn};
 use uuid::Uuid;
 
@@ -10,12 +10,14 @@ use crate::{
 	tonic::{transport::Channel, Status, Streaming},
 };
 
+pub type FactoryStore = Arc<RwLock<BTreeMap<String, Arc<AsyncFactory>>>>;
+
 impl From::<BufferClient<Channel>> for CodempClient {
 	fn from(x: BufferClient<Channel>) -> CodempClient {
 		CodempClient {
 			id: Uuid::new_v4(),
 			client:x,
-			factory: Arc::new(AsyncFactory::new(None)),
+			factories: Arc::new(RwLock::new(BTreeMap::new())),
 		}
 	}
 }
@@ -24,85 +26,85 @@ impl From::<BufferClient<Channel>> for CodempClient {
 pub struct CodempClient {
 	id: Uuid,
 	client: BufferClient<Channel>,
-	factory: Arc<AsyncFactory>,
+	factories: FactoryStore,
 }
 
 impl CodempClient {
+	fn get_factory(&self, path: &String) -> Result<Arc<AsyncFactory>, Status> {
+		match self.factories.read().unwrap().get(path) {
+			Some(f) => Ok(f.clone()),
+			None => Err(Status::not_found("no active buffer for given path")),
+		}
+	}
+
+	pub fn add_factory(&self, path: String, factory:Arc<AsyncFactory>) {
+		self.factories.write().unwrap().insert(path, factory);
+	}
+
 	pub async fn create(&mut self, path: String, content: Option<String>) -> Result<bool, Status> {
-		Ok(
+		let req = BufferPayload {
-			self.client.create(
+			path: path.clone(),
-				BufferPayload {
+			content: content.clone(),
-					path,
+			user: self.id.to_string(),
-					content,
+		};
-					user: self.id.to_string(),
+
-				}
+		let res = self.client.create(req).await?.into_inner();
-			)
+
-				.await?
+		Ok(res.accepted)
-				.into_inner()
-				.accepted
-		)
 	}
 
 	pub async fn insert(&mut self, path: String, txt: String, pos: u64) -> Result<bool, Status> {
-		match self.factory.insert(txt, pos).await {
+		let factory = self.get_factory(&path)?;
-			Ok(op) => {
+		match factory.insert(txt, pos).await {
-				Ok(
-					self.client.edit(
-						OperationRequest {
-							path,
-							hash: "".into(),
-							opseq: serde_json::to_string(&op).unwrap(),
-							user: self.id.to_string(),
-						}
-					)
-						.await?
-						.into_inner()
-						.accepted
-				)
-			},
 			Err(e) => Err(Status::internal(format!("invalid operation: {}", e))),
+			Ok(op) => {
+				let req = OperationRequest {
+					path,
+					hash: "".into(),
+					user: self.id.to_string(),
+					opseq: serde_json::to_string(&op)
+						.map_err(|_| Status::invalid_argument("could not serialize opseq"))?,
+				};
+				let res = self.client.edit(req).await?.into_inner();
+				Ok(res.accepted)
+			},
 		}
 	}
 
 	pub async fn delete(&mut self, path: String, pos: u64, count: u64) -> Result<bool, Status> {
-		match self.factory.delete(pos, count).await {
+		let factory = self.get_factory(&path)?;
-			Ok(op) => {
+		match factory.delete(pos, count).await {
-				Ok(
-					self.client.edit(
-						OperationRequest {
-							path,
-							hash: "".into(),
-							opseq: serde_json::to_string(&op).unwrap(),
-							user: self.id.to_string(),
-						}
-					)
-						.await?
-						.into_inner()
-						.accepted
-				)
-			},
 			Err(e) => Err(Status::internal(format!("invalid operation: {}", e))),
+			Ok(op) => {
+				let req = OperationRequest {
+					path,
+					hash: "".into(),
+					user: self.id.to_string(),
+					opseq: serde_json::to_string(&op)
+						.map_err(|_| Status::invalid_argument("could not serialize opseq"))?,
+				};
+				let res = self.client.edit(req).await?.into_inner();
+				Ok(res.accepted)
+			},
 		}
 	}
 
-	pub async fn attach<F : Fn(String) -> () + Send + 'static>(&mut self, path: String, callback: F) -> Result<(), Status> {
+	pub async fn attach<F>(&mut self, path: String, callback: F) -> Result<String, Status>
-		let stream = self.client.attach(
+	where F : Fn(String) -> () + Send + 'static {
-				BufferPayload {
+		let content = self.sync(path.clone()).await?;
-					path,
+		let factory = Arc::new(AsyncFactory::new(Some(content.clone())));
-					content: None,
+		self.add_factory(path.clone(), factory.clone());
-					user: self.id.to_string(),
+		let req = BufferPayload {
-				}
+			path,
-			)
+			content: None,
-				.await?
+			user: self.id.to_string(),
-				.into_inner();
+		};
-
+		let stream = self.client.attach(req).await?.into_inner();
-		let factory = self.factory.clone();
 		tokio::spawn(async move { Self::worker(stream, factory, callback).await } );
-
+		Ok(content)
-		Ok(())
 	}
 
-	pub async fn sync(&mut self, path: String) -> Result<String, Status> {
+	async fn sync(&mut self, path: String) -> Result<String, Status> {
 		let res = self.client.sync(
 			BufferPayload {
 				path, content: None, user: self.id.to_string(),
@@ -111,20 +113,19 @@ impl CodempClient {
 		Ok(res.into_inner().content.unwrap_or("".into()))
 	}
 
-	async fn worker<F : Fn(String) -> ()>(mut stream: Streaming<RawOp>, factory: Arc<AsyncFactory>, callback: F) {
+	async fn worker<F>(mut stream: Streaming<RawOp>, factory: Arc<AsyncFactory>, callback: F)
+	where F : Fn(String) -> () {
 		loop {
 			match stream.message().await {
 				Err(e) => break error!("error receiving change: {}", e),
 				Ok(v) => match v {
 					None => break warn!("stream closed"),
-					Some(operation) => {
+					Some(operation) => match serde_json::from_str(&operation.opseq) {
-						match serde_json::from_str(&operation.opseq) {
+						Err(e) => break error!("could not deserialize opseq: {}", e),
-							Err(e) => break error!("could not deserialize opseq: {}", e),
+						Ok(op) => match factory.process(op).await {
-							Ok(op) => match factory.process(op).await {
+							Err(e) => break error!("desynched: {}", e),
-								Err(e) => break error!("desynched: {}", e),
+							Ok(x) => callback(x),
-								Ok(x) => callback(x),
+						},
-							},
-						}
 					}
 				},
 			}