2023-08-20 00:46:55 +02:00
|
|
|
//! ### controller
|
|
|
|
//!
|
|
|
|
//! a controller implementation for buffer actions
|
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
use operational_transform::OperationSeq;
|
2023-08-21 02:35:56 +02:00
|
|
|
use tokio::sync::broadcast::error::TryRecvError;
|
2023-08-16 23:09:47 +02:00
|
|
|
use tokio::sync::{watch, mpsc, broadcast, Mutex};
|
|
|
|
use tonic::async_trait;
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2023-08-19 04:02:21 +02:00
|
|
|
use crate::errors::IgnorableError;
|
2023-09-03 23:04:08 +02:00
|
|
|
use crate::{api::Controller, Error};
|
2023-08-16 23:09:47 +02:00
|
|
|
use crate::buffer::factory::{leading_noop, tailing_noop, OperationFactory};
|
|
|
|
|
|
|
|
use super::TextChange;
|
2023-08-11 15:33:40 +02:00
|
|
|
|
2023-08-20 00:46:55 +02:00
|
|
|
/// the buffer controller implementation
|
|
|
|
///
|
|
|
|
/// this contains
|
|
|
|
/// * a watch channel which always contains an updated view of the buffer content
|
|
|
|
/// * a sink to send buffer operations into
|
|
|
|
/// * a mutexed broadcast receiver for buffer operations
|
|
|
|
/// * a channel to stop the associated worker
|
|
|
|
///
|
|
|
|
/// for each controller a worker exists, managing outgoing and inbound
|
|
|
|
/// queues, transforming outbound delayed ops and applying remote changes
|
|
|
|
/// to the local buffer
|
|
|
|
///
|
|
|
|
/// this controller implements [crate::buffer::OperationFactory], allowing to produce
|
|
|
|
/// Operation Sequences easily
|
|
|
|
///
|
|
|
|
/// upon dropping this handle will stop the associated worker
|
2023-08-21 02:47:52 +02:00
|
|
|
#[derive(Debug)]
|
2023-08-11 15:33:40 +02:00
|
|
|
pub struct BufferController {
|
2023-08-16 23:09:47 +02:00
|
|
|
content: watch::Receiver<String>,
|
2023-08-19 21:44:27 +02:00
|
|
|
operations: mpsc::UnboundedSender<OperationSeq>,
|
2023-08-21 02:35:56 +02:00
|
|
|
last_op: Mutex<watch::Receiver<String>>,
|
2023-08-16 23:09:47 +02:00
|
|
|
stream: Mutex<broadcast::Receiver<OperationSeq>>,
|
2023-08-19 04:02:21 +02:00
|
|
|
stop: mpsc::UnboundedSender<()>,
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
|
|
|
|
2023-08-17 00:04:37 +02:00
|
|
|
impl BufferController {
|
|
|
|
pub(crate) fn new(
|
|
|
|
content: watch::Receiver<String>,
|
2023-08-19 21:44:27 +02:00
|
|
|
operations: mpsc::UnboundedSender<OperationSeq>,
|
2023-08-17 00:04:37 +02:00
|
|
|
stream: Mutex<broadcast::Receiver<OperationSeq>>,
|
2023-08-19 04:02:21 +02:00
|
|
|
stop: mpsc::UnboundedSender<()>,
|
2023-08-17 00:04:37 +02:00
|
|
|
) -> Self {
|
2023-08-21 02:35:56 +02:00
|
|
|
BufferController {
|
|
|
|
last_op: Mutex::new(content.clone()),
|
|
|
|
content, operations, stream, stop,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
fn op_to_change(&self, op: OperationSeq) -> TextChange {
|
|
|
|
let after = self.content.borrow().clone();
|
|
|
|
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();
|
|
|
|
TextChange { span, content }
|
2023-08-19 04:02:21 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
impl Drop for BufferController {
|
|
|
|
fn drop(&mut self) {
|
|
|
|
self.stop.send(()).unwrap_or_warn("could not send stop message to worker");
|
2023-08-17 00:04:37 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
#[async_trait]
|
|
|
|
impl OperationFactory for BufferController {
|
|
|
|
fn content(&self) -> String {
|
|
|
|
self.content.borrow().clone()
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
|
|
|
|
2023-08-16 23:09:47 +02:00
|
|
|
#[async_trait]
|
|
|
|
impl Controller<TextChange> for BufferController {
|
|
|
|
type Input = OperationSeq;
|
|
|
|
|
2023-08-21 02:35:56 +02:00
|
|
|
async fn poll(&self) -> Result<(), Error> {
|
|
|
|
Ok(self.last_op.lock().await.changed().await?)
|
|
|
|
}
|
|
|
|
|
|
|
|
fn try_recv(&self) -> Result<Option<TextChange>, Error> {
|
|
|
|
match self.stream.blocking_lock().try_recv() {
|
|
|
|
Ok(op) => Ok(Some(self.op_to_change(op))),
|
|
|
|
Err(TryRecvError::Empty) => Ok(None),
|
|
|
|
Err(TryRecvError::Closed) => Err(Error::Channel { send: false }),
|
|
|
|
Err(TryRecvError::Lagged(n)) => {
|
|
|
|
tracing::warn!("buffer channel lagged, skipping {} events", n);
|
|
|
|
Ok(self.try_recv()?)
|
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-08-20 00:46:55 +02:00
|
|
|
/// receive an operation seq and transform it into a TextChange from buffer content
|
2023-08-17 02:58:55 +02:00
|
|
|
async fn recv(&self) -> Result<TextChange, Error> {
|
2023-08-16 23:09:47 +02:00
|
|
|
let op = self.stream.lock().await.recv().await?;
|
2023-08-21 02:35:56 +02:00
|
|
|
Ok(self.op_to_change(op))
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
|
|
|
|
2023-08-20 00:46:55 +02:00
|
|
|
/// enqueue an opseq for processing
|
2023-08-19 21:44:27 +02:00
|
|
|
fn send(&self, op: OperationSeq) -> Result<(), Error> {
|
|
|
|
Ok(self.operations.send(op)?)
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|