2023-08-16 23:09:47 +02:00
|
|
|
use std::{sync::Arc, collections::VecDeque};
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2023-09-04 03:09:32 +02:00
|
|
|
use operational_transform::{OperationSeq, OTError};
|
2023-08-11 15:33:40 +02:00
|
|
|
use tokio::sync::{watch, mpsc, broadcast, Mutex};
|
2023-08-16 18:58:42 +02:00
|
|
|
use tonic::transport::Channel;
|
|
|
|
use tonic::{async_trait, Streaming};
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2023-09-04 03:09:32 +02:00
|
|
|
use crate::errors::{IgnorableError, IgnorableDefaultableError};
|
2023-08-16 18:58:42 +02:00
|
|
|
use crate::proto::{OperationRequest, RawOp};
|
|
|
|
use crate::proto::buffer_client::BufferClient;
|
2023-09-03 23:04:08 +02:00
|
|
|
use crate::api::ControllerWorker;
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
use super::TextChange;
|
|
|
|
use super::controller::BufferController;
|
2023-09-04 03:09:32 +02:00
|
|
|
use super::factory::{leading_noop, tailing_noop};
|
2023-08-11 15:33:40 +02:00
|
|
|
|
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
pub(crate) struct BufferControllerWorker {
|
2023-08-16 18:58:42 +02:00
|
|
|
uid: String,
|
2023-08-11 15:33:40 +02:00
|
|
|
pub(crate) content: watch::Sender<String>,
|
2023-08-19 21:44:27 +02:00
|
|
|
pub(crate) operations: mpsc::UnboundedReceiver<OperationSeq>,
|
2023-09-04 03:09:32 +02:00
|
|
|
pub(crate) stream: Arc<broadcast::Sender<TextChange>>,
|
2023-08-11 15:33:40 +02:00
|
|
|
pub(crate) queue: VecDeque<OperationSeq>,
|
|
|
|
receiver: watch::Receiver<String>,
|
2023-08-19 21:44:27 +02:00
|
|
|
sender: mpsc::UnboundedSender<OperationSeq>,
|
2023-08-11 15:33:40 +02:00
|
|
|
buffer: String,
|
|
|
|
path: String,
|
2023-08-19 04:02:21 +02:00
|
|
|
stop: mpsc::UnboundedReceiver<()>,
|
|
|
|
stop_control: mpsc::UnboundedSender<()>,
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
impl BufferControllerWorker {
|
2023-08-16 18:58:42 +02:00
|
|
|
pub fn new(uid: String, buffer: &str, path: &str) -> Self {
|
|
|
|
let (txt_tx, txt_rx) = watch::channel(buffer.to_string());
|
2023-08-19 21:44:27 +02:00
|
|
|
let (op_tx, op_rx) = mpsc::unbounded_channel();
|
2023-08-16 18:58:42 +02:00
|
|
|
let (s_tx, _s_rx) = broadcast::channel(64);
|
2023-08-19 04:02:21 +02:00
|
|
|
let (end_tx, end_rx) = mpsc::unbounded_channel();
|
2023-08-16 23:09:47 +02:00
|
|
|
BufferControllerWorker {
|
2023-08-16 18:58:42 +02:00
|
|
|
uid,
|
|
|
|
content: txt_tx,
|
|
|
|
operations: op_rx,
|
|
|
|
stream: Arc::new(s_tx),
|
|
|
|
receiver: txt_rx,
|
|
|
|
sender: op_tx,
|
|
|
|
queue: VecDeque::new(),
|
|
|
|
buffer: buffer.to_string(),
|
|
|
|
path: path.to_string(),
|
2023-08-19 04:02:21 +02:00
|
|
|
stop: end_rx,
|
|
|
|
stop_control: end_tx,
|
2023-08-16 18:58:42 +02:00
|
|
|
}
|
|
|
|
}
|
2023-09-04 03:09:32 +02:00
|
|
|
|
|
|
|
fn update(&mut self, op: OperationSeq) -> Result<TextChange, OTError> {
|
|
|
|
let before = Arc::new(self.buffer.clone());
|
|
|
|
let res = op.apply(&before)?;
|
|
|
|
self.content.send(res.clone())
|
|
|
|
.unwrap_or_warn("error showing updated buffer");
|
|
|
|
let after = Arc::new(res.clone());
|
|
|
|
self.buffer = res;
|
|
|
|
let skip = leading_noop(op.ops()) as usize;
|
|
|
|
let before_len = op.base_len();
|
|
|
|
let tail = tailing_noop(op.ops()) as usize;
|
|
|
|
let span = skip..before_len-tail;
|
|
|
|
let content = after[skip..after.len()-tail].to_string();
|
|
|
|
Ok(TextChange { span, content, before, after })
|
|
|
|
}
|
2023-08-16 18:58:42 +02:00
|
|
|
}
|
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
#[async_trait]
|
2023-08-16 23:09:47 +02:00
|
|
|
impl ControllerWorker<TextChange> for BufferControllerWorker {
|
|
|
|
type Controller = BufferController;
|
2023-08-16 18:58:42 +02:00
|
|
|
type Tx = BufferClient<Channel>;
|
|
|
|
type Rx = Streaming<RawOp>;
|
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
fn subscribe(&self) -> BufferController {
|
2023-08-17 00:04:37 +02:00
|
|
|
BufferController::new(
|
|
|
|
self.receiver.clone(),
|
|
|
|
self.sender.clone(),
|
|
|
|
Mutex::new(self.stream.subscribe()),
|
2023-08-19 04:02:21 +02:00
|
|
|
self.stop_control.clone(),
|
2023-08-17 00:04:37 +02:00
|
|
|
)
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
|
|
|
|
2023-08-16 18:58:42 +02:00
|
|
|
async fn work(mut self, mut tx: Self::Tx, mut rx: Self::Rx) {
|
2023-08-11 15:33:40 +02:00
|
|
|
loop {
|
2023-09-04 03:09:32 +02:00
|
|
|
tokio::select! {
|
|
|
|
|
2023-08-16 18:58:42 +02:00
|
|
|
Some(operation) = recv_opseq(&mut rx) => {
|
2023-08-11 15:33:40 +02:00
|
|
|
let mut out = operation;
|
|
|
|
for op in self.queue.iter_mut() {
|
2023-08-19 04:02:21 +02:00
|
|
|
(*op, out) = match op.transform(&out) {
|
|
|
|
Ok((x, y)) => (x, y),
|
|
|
|
Err(e) => {
|
|
|
|
tracing::warn!("could not transform enqueued operation: {}", e);
|
|
|
|
break
|
|
|
|
},
|
|
|
|
}
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
2023-09-04 03:09:32 +02:00
|
|
|
let change = self.update(out)
|
|
|
|
.unwrap_or_warn_default("coult not update with (transformed) remote operation");
|
|
|
|
self.stream.send(change)
|
|
|
|
.unwrap_or_warn("could not send operation to server");
|
2023-08-11 15:33:40 +02:00
|
|
|
},
|
2023-09-04 03:09:32 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
Some(op) = self.operations.recv() => {
|
|
|
|
self.queue.push_back(op.clone());
|
2023-09-04 03:09:32 +02:00
|
|
|
self.update(op)
|
|
|
|
.unwrap_or_warn("could not apply enqueued operation to current buffer");
|
|
|
|
while let Some(op) = self.queue.get(0) {
|
|
|
|
if !send_opseq(&mut tx, self.uid.clone(), self.path.clone(), op.clone()).await { break }
|
|
|
|
self.queue.pop_front();
|
|
|
|
}
|
2023-08-11 15:33:40 +02:00
|
|
|
},
|
2023-09-04 03:09:32 +02:00
|
|
|
|
2023-08-19 04:02:21 +02:00
|
|
|
Some(()) = self.stop.recv() => {
|
|
|
|
break;
|
|
|
|
}
|
2023-09-04 03:09:32 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
else => break
|
2023-09-04 03:09:32 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2023-08-16 18:58:42 +02:00
|
|
|
}
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2023-08-16 18:58:42 +02:00
|
|
|
async fn send_opseq(tx: &mut BufferClient<Channel>, uid: String, path: String, op: OperationSeq) -> bool {
|
2023-08-19 04:02:21 +02:00
|
|
|
let opseq = match serde_json::to_string(&op) {
|
|
|
|
Ok(x) => x,
|
|
|
|
Err(e) => {
|
|
|
|
tracing::warn!("could not serialize opseq: {}", e);
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
};
|
2023-08-16 18:58:42 +02:00
|
|
|
let req = OperationRequest {
|
2023-08-20 07:32:08 +02:00
|
|
|
path, hash: "".into(),
|
|
|
|
op: Some(RawOp {
|
|
|
|
opseq, user: uid,
|
|
|
|
}),
|
2023-08-16 18:58:42 +02:00
|
|
|
};
|
|
|
|
match tx.edit(req).await {
|
|
|
|
Ok(_) => true,
|
|
|
|
Err(e) => {
|
|
|
|
tracing::error!("error sending edit: {}", e);
|
|
|
|
false
|
|
|
|
}
|
|
|
|
}
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
|
|
|
|
2023-08-16 18:58:42 +02:00
|
|
|
async fn recv_opseq(rx: &mut Streaming<RawOp>) -> Option<OperationSeq> {
|
|
|
|
match rx.message().await {
|
2023-08-19 04:02:21 +02:00
|
|
|
Ok(Some(op)) => match serde_json::from_str(&op.opseq) {
|
|
|
|
Ok(x) => Some(x),
|
|
|
|
Err(e) => {
|
|
|
|
tracing::warn!("could not deserialize opseq: {}", e);
|
|
|
|
None
|
|
|
|
}
|
|
|
|
},
|
2023-08-16 18:58:42 +02:00
|
|
|
Ok(None) => None,
|
|
|
|
Err(e) => {
|
|
|
|
tracing::error!("could not receive edit from server: {}", e);
|
|
|
|
None
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|