2023-11-10 04:29:50 +01:00
|
|
|
use std::collections::hash_map::DefaultHasher;
|
|
|
|
use std::hash::{Hash, Hasher};
|
2024-08-08 00:28:15 +02:00
|
|
|
use std::sync::Arc;
|
2023-11-10 04:29:50 +01:00
|
|
|
|
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;
|
2024-08-05 19:15:30 +02:00
|
|
|
use woot::crdt::CRDT;
|
2023-11-09 05:21:24 +01:00
|
|
|
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-03-09 19:59:36 +01:00
|
|
|
use codemp_proto::buffer::{BufferEvent, Operation};
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2024-08-08 00:28:15 +02:00
|
|
|
use super::controller::{BufferController, BufferControllerInner};
|
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
|
|
|
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
|
|
|
stop: mpsc::UnboundedReceiver<()>,
|
2024-08-08 00:28:15 +02:00
|
|
|
controller: Arc<BufferControllerInner>,
|
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-08-08 00:28:15 +02:00
|
|
|
let controller = BufferControllerInner::new(
|
|
|
|
path.to_string(),
|
|
|
|
txt_rx,
|
|
|
|
op_tx,
|
|
|
|
poller_tx,
|
|
|
|
end_tx,
|
|
|
|
);
|
2024-01-25 02:13:45 +01:00
|
|
|
BufferWorker {
|
|
|
|
_user_id: user_id,
|
2023-11-27 14:55:46 +01:00
|
|
|
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(),
|
2023-08-19 04:02:21 +02:00
|
|
|
stop: end_rx,
|
2024-08-08 00:28:15 +02:00
|
|
|
controller: Arc::new(controller),
|
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 {
|
2024-08-08 00:28:15 +02:00
|
|
|
BufferController(self.controller.clone())
|
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 {
|
2024-02-09 00:35:08 +01:00
|
|
|
None => break tracing::debug!("stopping: editor closed channel"),
|
|
|
|
Some(change) => match change.transform(&self.buffer) {
|
|
|
|
Err(e) => break tracing::error!("could not apply operation from client: {}", e),
|
|
|
|
Ok(ops) => {
|
|
|
|
for op in ops {
|
2024-08-05 19:15:30 +02:00
|
|
|
self.buffer.merge(op.0.clone());
|
2024-02-09 00:35:08 +01:00
|
|
|
let operation = Operation {
|
2024-08-05 19:15:30 +02:00
|
|
|
data: postcard::to_extend(&op.0, Vec::new()).unwrap(),
|
2024-02-09 00:35:08 +01:00
|
|
|
};
|
|
|
|
if let Err(e) = tx.send(operation).await {
|
2024-08-05 19:15:30 +02:00
|
|
|
tracing::error!("server refused to broadcast {}: {}", op.0, e);
|
2024-02-09 00:35:08 +01:00
|
|
|
}
|
2023-11-30 03:04:26 +01:00
|
|
|
}
|
2024-02-09 00:35:08 +01:00
|
|
|
self.content.send(self.buffer.view())
|
|
|
|
.unwrap_or_warn("could not send buffer update");
|
|
|
|
},
|
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-08-05 19:15:30 +02:00
|
|
|
Ok(Some(change)) => match postcard::from_bytes::<woot::crdt::Op>(&change.op.data) {
|
2024-02-07 01:09:28 +01:00
|
|
|
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
|
|
|
}
|