mirror of
https://github.com/hexedtech/codemp.git
synced 2024-11-22 15:24:48 +01:00
feat: send granular changes, added deadlocked error
This commit is contained in:
parent
39f2bd6ac2
commit
9df4594408
2 changed files with 27 additions and 32 deletions
|
@ -9,7 +9,7 @@ use tokio::sync::{watch, mpsc, RwLock};
|
||||||
use tonic::async_trait;
|
use tonic::async_trait;
|
||||||
|
|
||||||
use crate::errors::IgnorableError;
|
use crate::errors::IgnorableError;
|
||||||
use crate::{api::Controller, Error};
|
use crate::api::Controller;
|
||||||
|
|
||||||
use crate::api::TextChange;
|
use crate::api::TextChange;
|
||||||
|
|
||||||
|
@ -61,7 +61,7 @@ impl Drop for StopOnDrop {
|
||||||
impl Controller<TextChange> for BufferController {
|
impl Controller<TextChange> for BufferController {
|
||||||
type Input = TextChange;
|
type Input = TextChange;
|
||||||
|
|
||||||
async fn poll(&self) -> Result<(), Error> {
|
async fn poll(&self) -> crate::Result<()> {
|
||||||
let mut poller = self.content.clone();
|
let mut poller = self.content.clone();
|
||||||
loop {
|
loop {
|
||||||
poller.changed().await?;
|
poller.changed().await?;
|
||||||
|
@ -73,45 +73,37 @@ impl Controller<TextChange> for BufferController {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn try_recv(&self) -> Result<Option<TextChange>, Error> {
|
fn try_recv(&self) -> crate::Result<Option<TextChange>> {
|
||||||
let cur = match self.seen.try_read() {
|
match self.seen.try_read() {
|
||||||
Err(e) => {
|
Err(_) => Err(crate::Error::Deadlocked),
|
||||||
tracing::error!("try_recv invoked while being mutated: {}", e);
|
Ok(x) => {
|
||||||
return Ok(None);
|
if *self.content.borrow() != *x {
|
||||||
},
|
match self.seen.try_write() {
|
||||||
Ok(x) => x.clone(),
|
Err(_) => Err(crate::Error::Deadlocked),
|
||||||
};
|
Ok(mut w) => {
|
||||||
if *self.content.borrow() != cur {
|
let change = TextChange::from_diff(&w, &self.content.borrow());
|
||||||
match self.seen.try_write() {
|
*w = self.content.borrow().clone();
|
||||||
Err(e) => {
|
Ok(Some(change))
|
||||||
tracing::error!("try_recv mutating while being mutated: {}", e);
|
}
|
||||||
return Ok(None);
|
}
|
||||||
},
|
} else {
|
||||||
Ok(mut w) => {
|
Ok(None)
|
||||||
*w = self.content.borrow().clone();
|
|
||||||
// TODO it's not the whole buffer that changed
|
|
||||||
return Ok(Some(TextChange {
|
|
||||||
span: 0..cur.len(),
|
|
||||||
content: self.content.borrow().clone(),
|
|
||||||
after: "".to_string(),
|
|
||||||
}));
|
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return Ok(None);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn recv(&self) -> Result<TextChange, Error> {
|
async fn recv(&self) -> crate::Result<TextChange> {
|
||||||
self.poll().await?;
|
self.poll().await?;
|
||||||
match self.try_recv()? {
|
let cur = self.seen.read().await.clone();
|
||||||
Some(x) => Ok(x),
|
let change = TextChange::from_diff(&cur, &self.content.borrow());
|
||||||
None => Err(crate::Error::Filler { message: "wtfff".into() }),
|
let mut seen = self.seen.write().await;
|
||||||
}
|
*seen = self.content.borrow().clone();
|
||||||
|
Ok(change)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// enqueue an opseq for processing
|
/// enqueue an opseq for processing
|
||||||
fn send(&self, op: TextChange) -> Result<(), Error> {
|
fn send(&self, op: TextChange) -> crate::Result<()> {
|
||||||
Ok(self.operations.send(op)?)
|
Ok(self.operations.send(op)?)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -61,6 +61,9 @@ pub enum Error {
|
||||||
msg: String,
|
msg: String,
|
||||||
},
|
},
|
||||||
|
|
||||||
|
/// errors caused by wrong interlocking, safe to retry
|
||||||
|
Deadlocked,
|
||||||
|
|
||||||
/// if you see these errors someone is being lazy (:
|
/// if you see these errors someone is being lazy (:
|
||||||
Filler { // TODO filler error, remove later
|
Filler { // TODO filler error, remove later
|
||||||
message: String,
|
message: String,
|
||||||
|
|
Loading…
Reference in a new issue