2023-11-10 04:29:50 +01:00
|
|
|
use std::collections::hash_map::DefaultHasher;
|
|
|
|
use std::hash::{Hash, Hasher};
|
|
|
|
|
2023-11-09 05:21:24 +01:00
|
|
|
use similar::{TextDiff, ChangeTag};
|
2023-11-23 15:52:36 +01:00
|
|
|
use tokio::sync::{watch, mpsc, oneshot};
|
2023-08-16 18:58:42 +02:00
|
|
|
use tonic::{async_trait, Streaming};
|
2024-01-25 02:13:45 +01:00
|
|
|
use uuid::Uuid;
|
2023-11-09 05:21:24 +01:00
|
|
|
use woot::crdt::{Op, CRDT, TextEditor};
|
|
|
|
use woot::woot::Woot;
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2023-11-16 06:52:25 +01:00
|
|
|
use crate::errors::IgnorableError;
|
2023-09-10 03:00:47 +02:00
|
|
|
use crate::api::controller::ControllerWorker;
|
2023-11-17 05:45:31 +01:00
|
|
|
use crate::api::TextChange;
|
2024-02-07 01:09:28 +01:00
|
|
|
use crate::proto::buffer::{BufferEvent, Operation};
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
use super::controller::BufferController;
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2024-01-25 02:13:45 +01:00
|
|
|
pub(crate) struct BufferWorker {
|
|
|
|
_user_id: Uuid,
|
2023-11-27 14:55:46 +01:00
|
|
|
name: String,
|
|
|
|
buffer: Woot,
|
2023-09-05 20:13:09 +02:00
|
|
|
content: watch::Sender<String>,
|
2023-11-09 05:21:24 +01:00
|
|
|
operations: mpsc::UnboundedReceiver<TextChange>,
|
2023-11-30 03:03:09 +01:00
|
|
|
poller: mpsc::UnboundedReceiver<oneshot::Sender<()>>,
|
2023-11-23 15:52:36 +01:00
|
|
|
pollers: Vec<oneshot::Sender<()>>,
|
2023-11-27 14:55:46 +01:00
|
|
|
handles: ClonableHandlesForController,
|
|
|
|
stop: mpsc::UnboundedReceiver<()>,
|
|
|
|
}
|
|
|
|
|
|
|
|
struct ClonableHandlesForController {
|
|
|
|
operations: mpsc::UnboundedSender<TextChange>,
|
2023-11-30 03:03:09 +01:00
|
|
|
poller: mpsc::UnboundedSender<oneshot::Sender<()>>,
|
2023-11-27 14:55:46 +01:00
|
|
|
stop: mpsc::UnboundedSender<()>,
|
|
|
|
content: watch::Receiver<String>,
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
|
|
|
|
2024-01-25 02:13:45 +01:00
|
|
|
impl BufferWorker {
|
|
|
|
pub fn new(user_id: Uuid, path: &str) -> Self {
|
2023-11-16 06:52:25 +01:00
|
|
|
let (txt_tx, txt_rx) = watch::channel("".to_string());
|
2023-08-19 21:44:27 +02:00
|
|
|
let (op_tx, op_rx) = mpsc::unbounded_channel();
|
2023-08-19 04:02:21 +02:00
|
|
|
let (end_tx, end_rx) = mpsc::unbounded_channel();
|
2023-11-30 03:03:09 +01:00
|
|
|
let (poller_tx, poller_rx) = mpsc::unbounded_channel();
|
2023-11-10 04:29:50 +01:00
|
|
|
let mut hasher = DefaultHasher::new();
|
2024-01-25 02:13:45 +01:00
|
|
|
user_id.hash(&mut hasher);
|
2023-11-10 04:29:50 +01:00
|
|
|
let site_id = hasher.finish() as usize;
|
2024-01-25 02:13:45 +01:00
|
|
|
BufferWorker {
|
|
|
|
_user_id: user_id,
|
2023-11-27 14:55:46 +01:00
|
|
|
name: path.to_string(),
|
|
|
|
buffer: Woot::new(site_id % (2<<10), ""), // TODO remove the modulo, only for debugging!
|
2023-08-16 18:58:42 +02:00
|
|
|
content: txt_tx,
|
|
|
|
operations: op_rx,
|
2023-11-27 14:55:46 +01:00
|
|
|
poller: poller_rx,
|
|
|
|
pollers: Vec::new(),
|
|
|
|
handles: ClonableHandlesForController {
|
|
|
|
operations: op_tx,
|
|
|
|
poller: poller_tx,
|
|
|
|
stop: end_tx,
|
|
|
|
content: txt_rx,
|
|
|
|
},
|
2023-08-19 04:02:21 +02:00
|
|
|
stop: end_rx,
|
2023-08-16 18:58:42 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
#[async_trait]
|
2024-01-25 16:31:38 +01:00
|
|
|
impl ControllerWorker<TextChange> for BufferWorker {
|
2023-08-16 23:09:47 +02:00
|
|
|
type Controller = BufferController;
|
2024-01-25 02:13:45 +01:00
|
|
|
type Tx = mpsc::Sender<Operation>;
|
2024-02-07 01:09:28 +01:00
|
|
|
type Rx = Streaming<BufferEvent>;
|
2023-08-16 18:58:42 +02:00
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
fn subscribe(&self) -> BufferController {
|
2023-08-17 00:04:37 +02:00
|
|
|
BufferController::new(
|
2023-11-17 17:38:47 +01:00
|
|
|
self.name.clone(),
|
2023-11-27 14:55:46 +01:00
|
|
|
self.handles.content.clone(),
|
|
|
|
self.handles.operations.clone(),
|
|
|
|
self.handles.poller.clone(),
|
|
|
|
self.handles.stop.clone(),
|
2023-08-17 00:04:37 +02:00
|
|
|
)
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
|
|
|
|
2024-01-25 02:13:45 +01:00
|
|
|
async fn work(mut self, tx: Self::Tx, mut rx: Self::Rx) {
|
2023-08-11 15:33:40 +02:00
|
|
|
loop {
|
2023-09-05 20:13:09 +02:00
|
|
|
// block until one of these is ready
|
2023-09-04 03:09:32 +02:00
|
|
|
tokio::select! {
|
2023-09-10 03:01:37 +02:00
|
|
|
biased;
|
|
|
|
|
2023-11-09 05:21:24 +01:00
|
|
|
// received stop signal
|
|
|
|
_ = self.stop.recv() => break,
|
|
|
|
|
2023-11-23 15:52:36 +01:00
|
|
|
// received a new poller, add it to collection
|
2023-11-27 14:55:46 +01:00
|
|
|
res = self.poller.recv() => match res {
|
2023-11-23 15:52:36 +01:00
|
|
|
None => break tracing::error!("poller channel closed"),
|
|
|
|
Some(tx) => self.pollers.push(tx),
|
|
|
|
},
|
|
|
|
|
2023-11-09 05:21:24 +01:00
|
|
|
// received a text change from editor
|
|
|
|
res = self.operations.recv() => match res {
|
|
|
|
None => break,
|
|
|
|
Some(change) => {
|
2023-11-30 03:04:26 +01:00
|
|
|
if !change.is_empty() {
|
|
|
|
let view = self.buffer.view();
|
|
|
|
match view.get(change.span.clone()) {
|
|
|
|
None => tracing::error!("received illegal span from client: {:?} but buffer is of len {}", change.span, view.len()),
|
|
|
|
Some(span) => {
|
|
|
|
let diff = TextDiff::from_chars(span, &change.content);
|
|
|
|
|
|
|
|
let mut i = 0;
|
|
|
|
let mut ops = Vec::new();
|
|
|
|
for diff in diff.iter_all_changes() {
|
|
|
|
match diff.tag() {
|
|
|
|
ChangeTag::Equal => i += 1,
|
|
|
|
ChangeTag::Delete => match self.buffer.delete(change.span.start + i) {
|
|
|
|
Ok(op) => ops.push(op),
|
|
|
|
Err(e) => tracing::error!("could not apply deletion: {}", e),
|
|
|
|
},
|
|
|
|
ChangeTag::Insert => {
|
|
|
|
for c in diff.value().chars() {
|
|
|
|
match self.buffer.insert(change.span.start + i, c) {
|
|
|
|
Ok(op) => {
|
|
|
|
ops.push(op);
|
|
|
|
i += 1;
|
|
|
|
},
|
|
|
|
Err(e) => tracing::error!("could not apply insertion: {}", e),
|
|
|
|
}
|
2023-11-16 06:52:25 +01:00
|
|
|
}
|
2023-11-30 03:04:26 +01:00
|
|
|
},
|
|
|
|
}
|
2023-11-09 05:21:24 +01:00
|
|
|
}
|
2023-11-30 03:04:26 +01:00
|
|
|
|
|
|
|
for op in ops {
|
2024-01-25 02:13:45 +01:00
|
|
|
let operation = Operation {
|
|
|
|
data: postcard::to_extend(&op, Vec::new()).unwrap(),
|
|
|
|
};
|
|
|
|
|
|
|
|
match tx.send(operation).await {
|
2023-11-30 03:04:26 +01:00
|
|
|
Err(e) => tracing::error!("server refused to broadcast {}: {}", op, e),
|
|
|
|
Ok(()) => {
|
|
|
|
self.content.send(self.buffer.view()).unwrap_or_warn("could not send buffer update");
|
|
|
|
},
|
|
|
|
}
|
2023-11-16 06:52:25 +01:00
|
|
|
}
|
2023-11-30 03:04:26 +01:00
|
|
|
},
|
|
|
|
}
|
2023-09-05 20:13:09 +02:00
|
|
|
}
|
2023-09-04 03:09:32 +02:00
|
|
|
}
|
2023-08-11 15:33:40 +02:00
|
|
|
},
|
2023-09-10 03:01:37 +02:00
|
|
|
|
2023-11-23 15:52:36 +01:00
|
|
|
// received a message from server
|
2023-11-09 05:21:24 +01:00
|
|
|
res = rx.message() => match res {
|
|
|
|
Err(_e) => break,
|
|
|
|
Ok(None) => break,
|
2024-02-07 01:09:28 +01:00
|
|
|
Ok(Some(change)) => match postcard::from_bytes::<Op>(&change.op.data) {
|
|
|
|
Ok(op) => { // TODO here in change we receive info about the author, maybe propagate?
|
2023-11-16 06:52:25 +01:00
|
|
|
self.buffer.merge(op);
|
|
|
|
self.content.send(self.buffer.view()).unwrap_or_warn("could not send buffer update");
|
2023-11-30 03:04:26 +01:00
|
|
|
for tx in self.pollers.drain(..) {
|
2023-11-23 15:52:36 +01:00
|
|
|
tx.send(()).unwrap_or_warn("could not wake up poller");
|
|
|
|
}
|
2023-11-16 06:52:25 +01:00
|
|
|
},
|
|
|
|
Err(e) => tracing::error!("could not deserialize operation from server: {}", e),
|
2023-11-09 05:21:24 +01:00
|
|
|
},
|
2023-09-10 03:01:37 +02:00
|
|
|
},
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2023-08-16 18:58:42 +02:00
|
|
|
}
|