feat: refactored and improved operation controller

now polling for changes returns span and text so that it's possible to
edit just the changed region. greatly improved controller internal api
with crate-level traits keeping error handling localized
This commit is contained in:
əlemi 2023-07-09 03:44:27 +02:00
parent 75fdbfe48c
commit ca42874590
9 changed files with 337 additions and 315 deletions

View file

@ -1,14 +1,13 @@
use std::sync::Arc;
use operational_transform::OperationSeq;
use tonic::{transport::Channel, Status};
use tracing::{error, warn, debug};
use tonic::{transport::Channel, Status, Streaming, async_trait};
use uuid::Uuid;
use crate::{
cursor::{CursorControllerHandle, CursorControllerWorker, CursorProvider},
operation::{OperationProcessor, OperationController},
proto::{buffer_client::BufferClient, BufferPayload, OperationRequest}, errors::IgnorableError,
controller::{ControllerWorker,
cursor::{CursorControllerHandle, CursorControllerWorker, CursorEditor},
buffer::{OperationControllerHandle, OperationControllerEditor, OperationControllerWorker}
},
proto::{buffer_client::BufferClient, BufferPayload, RawOp, OperationRequest, Cursor},
};
#[derive(Clone)]
@ -24,6 +23,10 @@ impl From::<BufferClient<Channel>> for CodempClient {
}
impl CodempClient {
pub async fn new(dest: &str) -> Result<Self, tonic::transport::Error> {
Ok(BufferClient::connect(dest.to_string()).await?.into())
}
pub fn id(&self) -> &str { &self.id }
pub async fn create(&mut self, path: String, content: Option<String>) -> Result<bool, Status> {
@ -44,35 +47,21 @@ impl CodempClient {
user: self.id.clone(),
};
let mut stream = self.client.listen(req).await?.into_inner();
let stream = self.client.listen(req).await?.into_inner();
let mut controller = CursorControllerWorker::new(self.id().to_string());
let controller = CursorControllerWorker::new(self.id().to_string(), (self.clone(), stream));
let handle = controller.subscribe();
let mut _client = self.client.clone();
tokio::spawn(async move {
loop {
tokio::select!{
res = stream.message() => {
match res {
Err(e) => break error!("error receiving cursor: {}", e),
Ok(None) => break debug!("cursor worker clean exit"),
Ok(Some(x)) => { controller.broadcast(x); },
}
},
Some(op) = controller.wait() => {
_client.moved(op).await
.unwrap_or_warn("could not send cursor update")
}
}
}
tracing::debug!("cursor worker started");
controller.work().await;
tracing::debug!("cursor worker stopped");
});
Ok(handle)
}
pub async fn attach(&mut self, path: String) -> Result<Arc<OperationController>, Status> {
pub async fn attach(&mut self, path: String) -> Result<OperationControllerHandle, Status> {
let req = BufferPayload {
path: path.clone(),
content: None,
@ -82,59 +71,73 @@ impl CodempClient {
let content = self.client.sync(req.clone())
.await?
.into_inner()
.content;
.content
.unwrap_or("".into());
let mut stream = self.client.attach(req).await?.into_inner();
let stream = self.client.attach(req).await?.into_inner();
let factory = Arc::new(OperationController::new(content.unwrap_or("".into())));
let _factory = factory.clone();
let _path = path.clone();
let controller = OperationControllerWorker::new((self.clone(), stream), content, path);
let factory = controller.subscribe();
tokio::spawn(async move {
loop {
if !_factory.run() { break debug!("downstream worker clean exit") }
match stream.message().await {
Err(e) => break error!("error receiving update: {}", e),
Ok(None) => break warn!("stream closed for buffer {}", _path),
Ok(Some(x)) => match serde_json::from_str::<OperationSeq>(&x.opseq) {
Err(e) => error!("error deserializing opseq: {}", e),
Ok(v) => match _factory.process(v) {
Err(e) => break error!("could not apply operation from server: {}", e),
Ok(_range) => { } // range is obtained awaiting wait(), need to pass the OpSeq itself
}
},
}
}
});
let mut _client = self.client.clone();
let _uid = self.id.clone();
let _factory = factory.clone();
let _path = path.clone();
tokio::spawn(async move {
while let Some(op) = _factory.poll().await {
if !_factory.run() { break }
let req = OperationRequest {
hash: "".into(),
opseq: serde_json::to_string(&op).unwrap(),
path: _path.clone(),
user: _uid.clone(),
};
match _client.edit(req).await {
Ok(res) => match res.into_inner().accepted {
true => { _factory.ack().await; },
false => {
warn!("server rejected operation, retrying in 1s");
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
},
Err(e) => error!("could not send edit: {}", e),
}
}
debug!("upstream worker clean exit");
tracing::debug!("buffer worker started");
controller.work().await;
tracing::debug!("buffer worker stopped");
});
Ok(factory)
}
}
#[async_trait]
impl OperationControllerEditor for (CodempClient, Streaming<RawOp>) {
async fn edit(&mut self, path: String, op: OperationSeq) -> bool {
let req = OperationRequest {
hash: "".into(),
opseq: serde_json::to_string(&op).unwrap(),
path,
user: self.0.id().to_string(),
};
match self.0.client.edit(req).await {
Ok(res) => res.into_inner().accepted,
Err(e) => {
tracing::error!("error sending edit: {}", e);
false
}
}
}
async fn recv(&mut self) -> Option<OperationSeq> {
match self.1.message().await {
Ok(Some(op)) => Some(serde_json::from_str(&op.opseq).unwrap()),
Ok(None) => None,
Err(e) => {
tracing::error!("could not receive edit from server: {}", e);
None
}
}
}
}
#[async_trait]
impl CursorEditor for (CodempClient, Streaming<Cursor>) {
async fn moved(&mut self, cursor: Cursor) -> bool {
match self.0.client.moved(cursor).await {
Ok(res) => res.into_inner().accepted,
Err(e) => {
tracing::error!("could not send cursor movement: {}", e);
false
}
}
}
async fn recv(&mut self) -> Option<Cursor> {
match self.1.message().await {
Ok(cursor) => cursor,
Err(e) => {
tracing::error!("could not receive cursor update: {}", e);
None
}
}
}
}

139
src/controller/buffer.rs Normal file
View file

@ -0,0 +1,139 @@
use std::{sync::Arc, collections::VecDeque, ops::Range};
use operational_transform::OperationSeq;
use tokio::sync::{watch, mpsc, broadcast};
use tonic::async_trait;
use super::{leading_noop, tailing_noop, ControllerWorker};
use crate::errors::IgnorableError;
use crate::factory::OperationFactory;
pub struct TextChange {
pub span: Range<usize>,
pub content: String,
}
#[async_trait]
pub trait OperationControllerSubscriber {
async fn poll(&mut self) -> Option<TextChange>;
async fn apply(&self, op: OperationSeq);
}
pub struct OperationControllerHandle {
content: watch::Receiver<String>,
operations: mpsc::Sender<OperationSeq>,
original: Arc<broadcast::Sender<OperationSeq>>,
stream: broadcast::Receiver<OperationSeq>,
}
impl Clone for OperationControllerHandle {
fn clone(&self) -> Self {
OperationControllerHandle {
content: self.content.clone(),
operations: self.operations.clone(),
original: self.original.clone(),
stream: self.original.subscribe(),
}
}
}
#[async_trait]
impl OperationFactory for OperationControllerHandle {
fn content(&self) -> String {
self.content.borrow().clone()
}
}
#[async_trait]
impl OperationControllerSubscriber for OperationControllerHandle {
async fn poll(&mut self) -> Option<TextChange> {
let op = self.stream.recv().await.ok()?;
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();
Some(TextChange { span, content })
}
async fn apply(&self, op: OperationSeq) {
self.operations.send(op).await
.unwrap_or_warn("could not apply+send operation")
}
}
#[async_trait]
pub(crate) trait OperationControllerEditor {
async fn edit(&mut self, path: String, op: OperationSeq) -> bool;
async fn recv(&mut self) -> Option<OperationSeq>;
}
pub(crate) struct OperationControllerWorker<C : OperationControllerEditor> {
pub(crate) content: watch::Sender<String>,
pub(crate) operations: mpsc::Receiver<OperationSeq>,
pub(crate) stream: Arc<broadcast::Sender<OperationSeq>>,
pub(crate) queue: VecDeque<OperationSeq>,
receiver: watch::Receiver<String>,
sender: mpsc::Sender<OperationSeq>,
client: C,
buffer: String,
path: String,
}
#[async_trait]
impl<C : OperationControllerEditor + Send> ControllerWorker<OperationControllerHandle> for OperationControllerWorker<C> {
fn subscribe(&self) -> OperationControllerHandle {
OperationControllerHandle {
content: self.receiver.clone(),
operations: self.sender.clone(),
original: self.stream.clone(),
stream: self.stream.subscribe(),
}
}
async fn work(mut self) {
loop {
let op = tokio::select! {
Some(operation) = self.client.recv() => {
let mut out = operation;
for op in self.queue.iter_mut() {
(*op, out) = op.transform(&out).unwrap();
}
self.stream.send(out.clone()).unwrap();
out
},
Some(op) = self.operations.recv() => {
self.queue.push_back(op.clone());
op
},
else => break
};
self.buffer = op.apply(&self.buffer).unwrap();
self.content.send(self.buffer.clone()).unwrap();
while let Some(op) = self.queue.get(0) {
if !self.client.edit(self.path.clone(), op.clone()).await { break }
self.queue.pop_front();
}
}
}
}
impl<C : OperationControllerEditor> OperationControllerWorker<C> {
pub fn new(client: C, buffer: String, path: String) -> Self {
let (txt_tx, txt_rx) = watch::channel(buffer.clone());
let (op_tx, op_rx) = mpsc::channel(64);
let (s_tx, _s_rx) = broadcast::channel(64);
OperationControllerWorker {
content: txt_tx,
operations: op_rx,
stream: Arc::new(s_tx),
receiver: txt_rx,
sender: op_tx,
queue: VecDeque::new(),
client, buffer, path
}
}
}

104
src/controller/cursor.rs Normal file
View file

@ -0,0 +1,104 @@
use std::sync::Arc;
use tokio::sync::{mpsc, broadcast};
use tonic::async_trait;
use crate::{proto::{Position, Cursor}, errors::IgnorableError, controller::ControllerWorker};
#[async_trait]
pub trait CursorSubscriber {
async fn send(&self, path: &str, start: Position, end: Position);
async fn poll(&mut self) -> Option<Cursor>;
}
pub struct CursorControllerHandle {
uid: String,
op: mpsc::Sender<Cursor>,
stream: broadcast::Receiver<Cursor>,
original: Arc<broadcast::Sender<Cursor>>,
}
impl Clone for CursorControllerHandle {
fn clone(&self) -> Self {
Self {
uid: self.uid.clone(),
op: self.op.clone(),
stream: self.original.subscribe(),
original: self.original.clone()
}
}
}
#[async_trait]
impl CursorSubscriber for CursorControllerHandle {
async fn send(&self, path: &str, start: Position, end: Position) {
self.op.send(Cursor {
user: self.uid.clone(),
buffer: path.to_string(),
start: Some(start),
end: Some(end),
}).await.unwrap_or_warn("could not send cursor op")
}
async fn poll(&mut self) -> Option<Cursor> {
match self.stream.recv().await {
Ok(x) => Some(x),
Err(e) => {
tracing::warn!("could not poll for cursor: {}", e);
None
}
}
}
}
#[async_trait]
pub(crate) trait CursorEditor {
async fn moved(&mut self, cursor: Cursor) -> bool;
async fn recv(&mut self) -> Option<Cursor>;
}
pub(crate) struct CursorControllerWorker<C : CursorEditor> {
uid: String,
producer: mpsc::Sender<Cursor>,
op: mpsc::Receiver<Cursor>,
channel: Arc<broadcast::Sender<Cursor>>,
client: C,
}
impl<C : CursorEditor> CursorControllerWorker<C> {
pub(crate) fn new(uid: String, client: C) -> Self {
let (op_tx, op_rx) = mpsc::channel(64);
let (cur_tx, _cur_rx) = broadcast::channel(64);
CursorControllerWorker {
uid, client,
producer: op_tx,
op: op_rx,
channel: Arc::new(cur_tx),
}
}
}
#[async_trait]
impl<C : CursorEditor + Send> ControllerWorker<CursorControllerHandle> for CursorControllerWorker<C> {
fn subscribe(&self) -> CursorControllerHandle {
CursorControllerHandle {
uid: self.uid.clone(),
op: self.producer.clone(),
stream: self.channel.subscribe(),
original: self.channel.clone(),
}
}
async fn work(mut self) {
loop {
tokio::select!{
Some(cur) = self.client.recv() => self.channel.send(cur).unwrap_or_warn("could not broadcast event"),
Some(op) = self.op.recv() => { self.client.moved(op).await; },
else => break,
}
}
}
}

View file

@ -1,13 +1,16 @@
pub mod factory;
pub mod processor;
pub mod controller;
pub mod buffer;
pub mod cursor;
use std::ops::Range;
use operational_transform::{Operation, OperationSeq};
pub use processor::OperationProcessor;
pub use controller::OperationController;
pub use factory::OperationFactory;
use tonic::async_trait;
#[async_trait]
pub trait ControllerWorker<T> {
fn subscribe(&self) -> T;
async fn work(self);
}
pub const fn leading_noop(seq: &[Operation]) -> u64 { count_noop(seq.first()) }
pub const fn tailing_noop(seq: &[Operation]) -> u64 { count_noop(seq.last()) }

View file

@ -1,9 +1,4 @@
use std::sync::Arc;
use tokio::sync::{mpsc, broadcast};
use tonic::async_trait;
use crate::{proto::{Position, Cursor}, errors::IgnorableError};
use crate::proto::{Position, Cursor};
impl From::<Position> for (i32, i32) {
fn from(pos: Position) -> (i32, i32) {
@ -26,99 +21,3 @@ impl Cursor {
self.end.clone().unwrap_or((0, 0).into())
}
}
#[async_trait]
pub trait CursorSubscriber {
async fn send(&self, path: &str, start: Position, end: Position);
async fn poll(&mut self) -> Option<Cursor>;
}
pub struct CursorControllerHandle {
uid: String,
op: mpsc::Sender<Cursor>,
stream: broadcast::Receiver<Cursor>,
original: Arc<broadcast::Sender<Cursor>>,
}
impl Clone for CursorControllerHandle {
fn clone(&self) -> Self {
Self {
uid: self.uid.clone(),
op: self.op.clone(),
stream: self.original.subscribe(),
original: self.original.clone()
}
}
}
#[async_trait]
impl CursorSubscriber for CursorControllerHandle {
async fn send(&self, path: &str, start: Position, end: Position) {
self.op.send(Cursor {
user: self.uid.clone(),
buffer: path.to_string(),
start: Some(start),
end: Some(end),
}).await.unwrap_or_warn("could not send cursor op")
}
async fn poll(&mut self) -> Option<Cursor> {
match self.stream.recv().await {
Ok(x) => Some(x),
Err(e) => {
tracing::warn!("could not poll for cursor: {}", e);
None
}
}
}
}
#[async_trait]
pub(crate) trait CursorProvider<T>
where T : CursorSubscriber {
fn subscribe(&self) -> T;
fn broadcast(&self, op: Cursor);
async fn wait(&mut self) -> Option<Cursor>;
}
pub(crate) struct CursorControllerWorker {
uid: String,
producer: mpsc::Sender<Cursor>,
op: mpsc::Receiver<Cursor>,
channel: Arc<broadcast::Sender<Cursor>>,
}
impl CursorControllerWorker {
pub(crate) fn new(uid: String) -> Self {
let (op_tx, op_rx) = mpsc::channel(64);
let (cur_tx, _cur_rx) = broadcast::channel(64);
CursorControllerWorker {
uid,
producer: op_tx,
op: op_rx,
channel: Arc::new(cur_tx),
}
}
}
#[async_trait]
impl CursorProvider<CursorControllerHandle> for CursorControllerWorker {
fn broadcast(&self, op: Cursor) {
self.channel.send(op).unwrap_or_warn("could not broadcast cursor event")
}
async fn wait(&mut self) -> Option<Cursor> {
self.op.recv().await
}
fn subscribe(&self) -> CursorControllerHandle {
CursorControllerHandle {
uid: self.uid.clone(),
op: self.producer.clone(),
stream: self.channel.subscribe(),
original: self.channel.clone(),
}
}
}

View file

@ -4,18 +4,19 @@ use similar::{TextDiff, ChangeTag};
pub trait OperationFactory {
fn content(&self) -> String;
fn replace(&self, txt: &str) -> OperationSeq {
fn replace(&self, txt: &str) -> Option<OperationSeq> {
self.delta(0, txt, self.content().len())
}
fn delta(&self, skip: usize, txt: &str, tail: usize) -> OperationSeq {
fn delta(&self, skip: usize, txt: &str, tail: usize) -> Option<OperationSeq> {
let mut out = OperationSeq::default();
let content = self.content();
let tail_index = content.len() - tail;
let content_slice = &content[skip..tail];
if content_slice == txt {
return out; // TODO this won't work, should we return a noop instead?
// if slice equals given text, no operation should be taken
return None;
}
out.retain(skip as u64);
@ -32,7 +33,7 @@ pub trait OperationFactory {
out.retain(tail_index as u64);
out
Some(out)
}
fn insert(&self, txt: &str, pos: u64) -> OperationSeq {

View file

@ -1,8 +1,9 @@
pub mod proto;
pub mod client;
pub mod operation;
pub mod controller;
pub mod cursor;
pub mod errors;
pub mod factory;
pub use tonic;
pub use tokio;

View file

@ -1,118 +0,0 @@
use std::{sync::Mutex, collections::VecDeque, ops::Range};
use operational_transform::{OperationSeq, OTError};
use tokio::sync::watch;
use tracing::error;
use super::{OperationFactory, OperationProcessor, op_effective_range};
use crate::errors::IgnorableError;
#[derive(Debug)]
pub struct OperationController {
text: Mutex<String>,
queue: Mutex<VecDeque<OperationSeq>>,
last: Mutex<watch::Receiver<OperationSeq>>,
notifier: watch::Sender<OperationSeq>,
changed: Mutex<watch::Receiver<Range<u64>>>,
changed_notifier: watch::Sender<Range<u64>>,
run: watch::Receiver<bool>,
stop: watch::Sender<bool>,
}
impl OperationController {
pub fn new(content: String) -> Self {
let (tx, rx) = watch::channel(OperationSeq::default());
let (done, wait) = watch::channel(0..0);
let (stop, run) = watch::channel(true);
OperationController {
text: Mutex::new(content),
queue: Mutex::new(VecDeque::new()),
last: Mutex::new(rx),
notifier: tx,
changed: Mutex::new(wait),
changed_notifier: done,
run, stop,
}
}
pub async fn wait(&self) -> Range<u64> {
let mut blocker = self.changed.lock().unwrap().clone();
// TODO less jank way
blocker.changed().await.unwrap_or_warn("waiting for changed content #1");
blocker.changed().await.unwrap_or_warn("waiting for changed content #2");
let span = blocker.borrow().clone();
span
}
pub async fn poll(&self) -> Option<OperationSeq> {
let len = self.queue.lock().unwrap().len();
if len == 0 {
let mut recv = self.last.lock().unwrap().clone();
// TODO less jank way
recv.changed().await.unwrap_or_warn("wairing for op changes #1"); // acknowledge current state
recv.changed().await.unwrap_or_warn("wairing for op changes #2"); // wait for a change in state
}
Some(self.queue.lock().unwrap().get(0)?.clone())
}
pub async fn ack(&self) -> Option<OperationSeq> {
self.queue.lock().unwrap().pop_front()
}
pub fn stop(&self) -> bool {
match self.stop.send(false) {
Ok(()) => {
self.changed_notifier.send(0..0).unwrap_or_warn("unlocking downstream for stop");
self.notifier.send(OperationSeq::default()).unwrap_or_warn("unlocking upstream for stop");
true
},
Err(e) => {
error!("could not send stop signal to workers: {}", e);
false
}
}
}
pub fn run(&self) -> bool {
*self.run.borrow()
}
fn operation(&self, op: &OperationSeq) -> Result<Range<u64>, OTError> {
let txt = self.content();
let res = op.apply(&txt)?;
*self.text.lock().unwrap() = res;
Ok(op_effective_range(op))
}
fn transform(&self, mut op: OperationSeq) -> Result<OperationSeq, OTError> {
let mut queue = self.queue.lock().unwrap();
for el in queue.iter_mut() {
(op, *el) = op.transform(el)?;
}
Ok(op)
}
}
impl OperationFactory for OperationController {
fn content(&self) -> String {
self.text.lock().unwrap().clone()
}
}
impl OperationProcessor for OperationController {
fn apply(&self, op: OperationSeq) -> Result<Range<u64>, OTError> {
let span = self.operation(&op)?;
self.queue.lock().unwrap().push_back(op.clone());
self.notifier.send(op).unwrap_or_warn("notifying of applied change");
Ok(span)
}
fn process(&self, mut op: OperationSeq) -> Result<Range<u64>, OTError> {
op = self.transform(op)?;
let span = self.operation(&op)?;
self.changed_notifier.send(span.clone()).unwrap_or_warn("notifying of changed content");
Ok(span)
}
}

View file

@ -1,10 +0,0 @@
use std::ops::Range;
use operational_transform::{OperationSeq, OTError};
use crate::operation::factory::OperationFactory;
pub trait OperationProcessor : OperationFactory {
fn apply(&self, op: OperationSeq) -> Result<Range<u64>, OTError>;
fn process(&self, op: OperationSeq) -> Result<Range<u64>, OTError>;
}