fix: ignore errors better, unified op processing

This commit is contained in:
əlemi 2023-07-01 13:54:34 +02:00
parent 107c4673ef
commit 2c695a41f6
3 changed files with 42 additions and 30 deletions

15
src/errors.rs Normal file
View file

@ -0,0 +1,15 @@
use tracing::warn;
pub trait IgnorableError {
fn unwrap_or_log(self, msg: &str);
}
impl<T, E> IgnorableError for Result<T, E>
where E : std::fmt::Display {
fn unwrap_or_log(self, msg: &str) {
match self {
Ok(_) => {},
Err(e) => warn!("{}: {}", msg, e),
}
}
}

View file

@ -2,6 +2,8 @@ pub mod proto;
pub mod client; pub mod client;
pub mod operation; pub mod operation;
pub mod cursor; pub mod cursor;
pub mod errors;
pub use tonic; pub use tonic;
pub use tokio; pub use tokio;

View file

@ -2,9 +2,10 @@ use std::{sync::Mutex, collections::VecDeque, ops::Range};
use operational_transform::{OperationSeq, OTError}; use operational_transform::{OperationSeq, OTError};
use tokio::sync::watch; use tokio::sync::watch;
use tracing::{warn, error}; use tracing::error;
use super::{OperationFactory, OperationProcessor, op_effective_range}; use super::{OperationFactory, OperationProcessor, op_effective_range};
use crate::errors::IgnorableError;
pub struct OperationController { pub struct OperationController {
@ -37,8 +38,8 @@ impl OperationController {
pub async fn wait(&self) -> Range<u64> { pub async fn wait(&self) -> Range<u64> {
let mut blocker = self.changed.lock().unwrap().clone(); let mut blocker = self.changed.lock().unwrap().clone();
// TODO less jank way // TODO less jank way
ignore_and_log(blocker.changed().await, "waiting for changed content #1"); blocker.changed().await.unwrap_or_log("waiting for changed content #1");
ignore_and_log(blocker.changed().await, "waiting for changed content #2"); blocker.changed().await.unwrap_or_log("waiting for changed content #2");
let span = blocker.borrow().clone(); let span = blocker.borrow().clone();
span span
} }
@ -48,8 +49,8 @@ impl OperationController {
if len <= 0 { if len <= 0 {
let mut recv = self.last.lock().unwrap().clone(); let mut recv = self.last.lock().unwrap().clone();
// TODO less jank way // TODO less jank way
ignore_and_log(recv.changed().await, "wairing for op changes #1"); // acknowledge current state recv.changed().await.unwrap_or_log("wairing for op changes #1"); // acknowledge current state
ignore_and_log(recv.changed().await, "wairing for op changes #2"); // wait for a change in state recv.changed().await.unwrap_or_log("wairing for op changes #2"); // wait for a change in state
} }
Some(self.queue.lock().unwrap().get(0)?.clone()) Some(self.queue.lock().unwrap().get(0)?.clone())
} }
@ -61,8 +62,8 @@ impl OperationController {
pub fn stop(&self) -> bool { pub fn stop(&self) -> bool {
match self.stop.send(false) { match self.stop.send(false) {
Ok(()) => { Ok(()) => {
ignore_and_log(self.changed_notifier.send(0..0), "unlocking downstream for stop"); self.changed_notifier.send(0..0).unwrap_or_log("unlocking downstream for stop");
ignore_and_log(self.notifier.send(OperationSeq::default()), "unlocking upstream for stop"); self.notifier.send(OperationSeq::default()).unwrap_or_log("unlocking upstream for stop");
true true
}, },
Err(e) => { Err(e) => {
@ -75,6 +76,13 @@ impl OperationController {
pub fn run(&self) -> bool { pub fn run(&self) -> bool {
*self.run.borrow() *self.run.borrow()
} }
async fn operation(&self, op: &OperationSeq) -> Result<Range<u64>, OTError> {
let txt = self.content();
let res = op.apply(&txt)?;
*self.text.lock().unwrap() = res.clone();
Ok(op_effective_range(op))
}
} }
impl OperationFactory for OperationController { impl OperationFactory for OperationController {
@ -83,38 +91,25 @@ impl OperationFactory for OperationController {
} }
} }
/// TODO properly handle errors rather than sinking them all in here!
fn ignore_and_log<T, E : std::fmt::Display>(x: Result<T, E>, msg: &str) {
match x {
Ok(_) => {},
Err(e) => {
warn!("ignored error {}: {}", msg, e);
}
}
}
#[tonic::async_trait] #[tonic::async_trait]
impl OperationProcessor for OperationController { impl OperationProcessor for OperationController {
async fn apply(&self, op: OperationSeq) -> Result<Range<u64>, OTError> { async fn apply(&self, op: OperationSeq) -> Result<Range<u64>, OTError> {
let txt = self.content(); let span = self.operation(&op).await?;
let res = op.apply(&txt)?;
*self.text.lock().unwrap() = res.clone();
self.queue.lock().unwrap().push_back(op.clone()); self.queue.lock().unwrap().push_back(op.clone());
ignore_and_log(self.notifier.send(op.clone()), "notifying of applied change"); self.notifier.send(op.clone()).unwrap_or_log("notifying of applied change");
Ok(op_effective_range(&op)) Ok(span)
} }
async fn process(&self, mut op: OperationSeq) -> Result<Range<u64>, OTError> { async fn process(&self, mut op: OperationSeq) -> Result<Range<u64>, OTError> {
let mut queue = self.queue.lock().unwrap(); {
for el in queue.iter_mut() { let mut queue = self.queue.lock().unwrap();
(op, *el) = op.transform(el)?; for el in queue.iter_mut() {
(op, *el) = op.transform(el)?;
}
} }
let txt = self.content(); let span = self.operation(&op).await?;
let res = op.apply(&txt)?; self.changed_notifier.send(span.clone()).unwrap_or_log("notifying of changed content");
let span = op_effective_range(&op);
*self.text.lock().unwrap() = res.clone();
ignore_and_log(self.changed_notifier.send(span.clone()), "notifying of changed content");
Ok(span) Ok(span)
} }
} }