hexedtech/codemp
3
0
Fork 0
mirror of https://github.com/hexedtech/codemp.git synced 2024-10-22 17:59:20 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat: implemented CRDT engine (merge branch 'woot')

Browse source
This commit is contained in:
əlemi 2023-11-23 20:46:46 +01:00
commit 0bd8f0541d
15 changed files with 330 additions and 452 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
Cargo.toml
View file

@ -9,8 +9,8 @@ name = "codemp"
[dependencies] [dependencies]
# core # core
tracing = "0.1" tracing = "0.1"
# ot # woot
operational-transform = { version = "0.6", features = ["serde"], optional = true } codemp-woot = { git = "ssh://git@github.com/codewithotherpeopleandchangenamelater/woot.git", tag = "v0.1.0", optional = true }
# proto # proto
tonic = { version = "0.9", features = ["tls", "tls-roots"], optional = true } tonic = { version = "0.9", features = ["tls", "tls-roots"], optional = true }
prost = { version = "0.11.8", optional = true } prost = { version = "0.11.8", optional = true }
@ -31,8 +31,8 @@ tonic-build = "0.9"
[features] [features]
default = ["client"] default = ["client"]
api = ["ot", "dep:similar", "dep:tokio", "dep:async-trait"] api = ["woot", "dep:similar", "dep:tokio", "dep:async-trait"]
ot = ["dep:operational-transform"] woot = ["dep:codemp-woot"]
proto = ["dep:prost", "dep:tonic"] proto = ["dep:prost", "dep:tonic"]
client = ["proto", "api", "dep:tokio", "dep:tokio-stream", "dep:uuid", "dep:md5", "dep:serde_json"] client = ["proto", "api", "dep:tokio", "dep:tokio-stream", "dep:uuid", "dep:md5", "dep:serde_json"]
global = ["client", "dep:lazy_static"] global = ["client", "dep:lazy_static"]

59
src/api/change.rs Normal file
View file

@ -0,0 +1,59 @@
//! # TextChange
//!
//! an editor-friendly representation of a text change in a buffer
//! to easily interface with codemp from various editors
/// an editor-friendly representation of a text change in a buffer
///
/// this represent a range in the previous state of the string and a new content which should be
/// replaced to it, allowing to represent any combination of deletions, insertions or replacements
///
/// bulk and widespread operations will result in a TextChange effectively sending the whole new
/// buffer, but small changes are efficient and easy to create or apply
///
/// ### examples
/// to insert 'a' after 4th character we should send a
/// `TextChange { span: 4..4, content: "a".into() }`
///
/// to delete a the fourth character we should send a
/// `TextChange { span: 3..4, content: "".into() }`
///
#[derive(Clone, Debug, Default)]
pub struct TextChange {
/// range of text change, as char indexes in buffer previous state
pub span: std::ops::Range<usize>,
/// new content of text inside span
pub content: String,
}
impl TextChange {
/// create a new TextChange from the difference of given strings
pub fn from_diff(before: &str, after: &str) -> TextChange {
let diff = similar::TextDiff::from_chars(before, after);
let mut start = 0;
let mut end = 0;
let mut from_beginning = true;
for op in diff.ops() {
match op {
similar::DiffOp::Equal { .. } => {
if from_beginning {
start += 1
} else {
end += 1
}
},
_ => {
end = 0;
from_beginning = false;
}
}
}
let end_before = before.len() - end;
let end_after = after.len() - end;
TextChange {
span: start..end_before,
content: after[start..end_after].to_string(),
}
}
}

52
src/api/controller.rs
View file

@ -4,7 +4,6 @@
//! server //! server
use crate::Result; use crate::Result;
use std::sync::Arc;
#[async_trait::async_trait] #[async_trait::async_trait]
pub(crate) trait ControllerWorker<T : Sized + Send + Sync> { pub(crate) trait ControllerWorker<T : Sized + Send + Sync> {
@ -21,20 +20,22 @@ pub(crate) trait ControllerWorker<T : Sized + Send + Sync> {
/// this generic trait is implemented by actors managing stream procedures. /// this generic trait is implemented by actors managing stream procedures.
/// events can be enqueued for dispatching without blocking ([Controller::send]), and an async blocking /// events can be enqueued for dispatching without blocking ([Controller::send]), and an async blocking
/// api ([Controller::recv]) is provided to wait for server events. Additional sync blocking /// api ([Controller::recv]) is provided to wait for server events. Additional sync blocking
/// ([Controller::blocking_recv]) and callback-based ([Controller::callback]) are implemented. /// ([Controller::blocking_recv]) is implemented if feature `sync` is enabled.
/// ///
/// * if possible, prefer a pure [Controller::recv] consumer /// * if possible, prefer a pure [Controller::recv] consumer, awaiting for events
/// * a second possibility in preference is using a [Controller::callback] /// * if async is not feasible a [Controller::poll]/[Controller::try_recv] approach is possible
/// * if neither is feasible a [Controller::poll]/[Controller::try_recv] approach is available
#[async_trait::async_trait] #[async_trait::async_trait]
pub trait Controller<T : Sized + Send + Sync> : Sized + Send + Sync { pub trait Controller<T : Sized + Send + Sync> : Sized + Send + Sync {
/// type of upstream values, used in [Self::send] /// type of upstream values, used in [Self::send]
type Input; type Input;
/// enqueue a new value to be sent /// enqueue a new value to be sent to all other users
///
/// success or failure of this function does not imply validity of sent operation,
/// because it's integrated asynchronously on the background worker
fn send(&self, x: Self::Input) -> Result<()>; fn send(&self, x: Self::Input) -> Result<()>;
/// get next value from stream, blocking until one is available /// get next value from other users, blocking until one is available
/// ///
/// this is just an async trait function wrapped by `async_trait`: /// this is just an async trait function wrapped by `async_trait`:
/// ///
@ -48,7 +49,7 @@ pub trait Controller<T : Sized + Send + Sync> : Sized + Send + Sync {
Ok(self.try_recv()?.expect("no message available after polling")) Ok(self.try_recv()?.expect("no message available after polling"))
} }
/// block until next value is added to the stream without removing any element /// block until next value is available without consuming it
/// ///
/// this is just an async trait function wrapped by `async_trait`: /// this is just an async trait function wrapped by `async_trait`:
/// ///
@ -56,40 +57,15 @@ pub trait Controller<T : Sized + Send + Sync> : Sized + Send + Sync {
async fn poll(&self) -> Result<()>; async fn poll(&self) -> Result<()>;
/// attempt to receive a value without blocking, return None if nothing is available /// attempt to receive a value without blocking, return None if nothing is available
///
/// note that this function does not circumvent race conditions, returning errors if it would
/// block. it's usually safe to ignore such errors and retry
fn try_recv(&self) -> Result<Option<T>>; fn try_recv(&self) -> Result<Option<T>>;
/// sync variant of [Self::recv], blocking invoking thread /// sync variant of [Self::recv], blocking invoking thread
/// this calls [Controller::recv] inside a [tokio::runtime::Runtime::block_on]
#[cfg(feature = "sync")]
fn blocking_recv(&self, rt: &tokio::runtime::Handle) -> Result<T> { fn blocking_recv(&self, rt: &tokio::runtime::Handle) -> Result<T> {
rt.block_on(self.recv()) rt.block_on(self.recv())
} }
/// register a callback to be called for each received stream value
///
/// this will spawn a new task on given runtime invoking [Self::recv] in loop and calling given
/// callback for each received value. a stop channel should be provided, and first value sent
/// into it will stop the worker loop.
///
/// note: creating a callback handler will hold an Arc reference to the given controller,
/// preventing it from being dropped (and likely disconnecting). using the stop channel is
/// important for proper cleanup
fn callback<F>(
self: &Arc<Self>,
rt: &tokio::runtime::Handle,
mut stop: tokio::sync::mpsc::UnboundedReceiver<()>,
mut cb: F
) where
Self : 'static,
F : FnMut(T) + Sync + Send + 'static
{
let _self = self.clone();
rt.spawn(async move {
loop {
tokio::select! {
Ok(data) = _self.recv() => cb(data),
Some(()) = stop.recv() => break,
else => break,
}
}
});
}
} }

165
src/api/factory.rs
View file

@ -1,165 +0,0 @@
//! ### factory
//!
//! a helper trait that any string container can implement, which generates opseqs
//!
//! an OperationFactory trait implementation is provided for `String` and `Arc<str>`, but plugin developers
//! should implement their own operation factory interfacing directly with the editor
//! buffer when possible.
use std::ops::Range;
use operational_transform::{OperationSeq, Operation};
use similar::{TextDiff, ChangeTag};
/// calculate leading no-ops in given opseq
pub const fn leading_noop(seq: &[Operation]) -> u64 { count_noop(seq.first()) }
/// calculate tailing no-ops in given opseq
pub const fn tailing_noop(seq: &[Operation]) -> u64 { count_noop(seq.last()) }
const fn count_noop(op: Option<&Operation>) -> u64 {
match op {
None => 0,
Some(Operation::Retain(n)) => *n,
Some(_) => 0,
}
}
/// return the range on which the operation seq is actually applying its changes
pub fn op_effective_range(op: &OperationSeq) -> Range<u64> {
let first = leading_noop(op.ops());
let last = op.base_len() as u64 - tailing_noop(op.ops());
first..last
}
/// a helper trait that any string container can implement, which generates opseqs
///
/// all operations are to be considered mutating current state, obtainable with
/// [OperationFactory::content]. generating an operation has no effect on internal state
///
/// ### examples
///
/// ```rust
/// use codemp::api::OperationFactory;
///
/// let mut factory = String::new();
/// let operation = factory.ins("asd", 0);
/// factory = operation.apply(&factory)?;
/// assert_eq!(factory, "asd");
/// # Ok::<(), codemp::ot::OTError>(())
/// ```
///
/// use [OperationFactory::ins] to add new characters at a specific index
///
/// ```rust
/// # use codemp::api::OperationFactory;
/// # let mut factory = String::from("asd");
/// factory = factory.ins(" dsa", 3).apply(&factory)?;
/// assert_eq!(factory, "asd dsa");
/// # Ok::<(), codemp::ot::OTError>(())
/// ```
///
/// use [OperationFactory::diff] to arbitrarily change text at any position
///
/// ```rust
/// # use codemp::api::OperationFactory;
/// # let mut factory = String::from("asd dsa");
/// factory = factory
/// .diff(2, " xxx ", 5)
/// .expect("replaced region is equal to origin")
/// .apply(&factory)?;
/// assert_eq!(factory, "as xxx sa");
/// # Ok::<(), codemp::ot::OTError>(())
/// ```
///
/// use [OperationFactory::del] to remove characters from given index
///
/// ```rust
/// # use codemp::api::OperationFactory;
/// # let mut factory = String::from("as xxx sa");
/// factory = factory.del(2, 5).apply(&factory)?;
/// assert_eq!(factory, "assa");
/// # Ok::<(), codemp::ot::OTError>(())
/// ```
///
/// use [OperationFactory::replace] to completely replace buffer content
///
/// ```rust
/// # use codemp::api::OperationFactory;
/// # let mut factory = String::from("assa");
/// factory = factory.replace("from scratch")
/// .expect("replace is equal to origin")
/// .apply(&factory)?;
/// assert_eq!(factory, "from scratch");
/// # Ok::<(), codemp::ot::OTError>(())
/// ```
pub trait OperationFactory {
/// the current content of the buffer
fn content(&self) -> String;
/// completely replace the buffer with given text
fn replace(&self, txt: &str) -> Option<OperationSeq> {
self.diff(0, txt, self.content().len())
}
/// transform buffer in range [start..end] with given text
fn diff(&self, start: usize, txt: &str, end: usize) -> Option<OperationSeq> {
let mut out = OperationSeq::default();
let content = self.content();
let tail_skip = content.len() - end; // TODO len is number of bytes, not chars
let content_slice = &content[start..end];
if content_slice == txt {
// if slice equals given text, no operation should be taken
return None;
}
out.retain(start as u64);
let diff = TextDiff::from_chars(content_slice, txt);
for change in diff.iter_all_changes() {
match change.tag() {
ChangeTag::Equal => out.retain(1),
ChangeTag::Delete => out.delete(1),
ChangeTag::Insert => out.insert(change.value()),
}
}
out.retain(tail_skip as u64);
Some(out)
}
/// insert given chars at target position
fn ins(&self, txt: &str, pos: u64) -> OperationSeq {
let mut out = OperationSeq::default();
let total = self.content().len() as u64;
out.retain(pos);
out.insert(txt);
out.retain(total - pos);
out
}
/// delete n characters forward at given position
fn del(&self, pos: u64, count: u64) -> OperationSeq {
let mut out = OperationSeq::default();
let len = self.content().len() as u64;
out.retain(pos);
out.delete(count);
out.retain(len - (pos+count));
out
}
}
impl OperationFactory for String {
fn content(&self) -> String {
self.clone()
}
}
impl OperationFactory for std::sync::Arc<str> {
fn content(&self) -> String {
self.to_string()
}
}

6
src/api/mod.rs
View file

@ -7,8 +7,8 @@
/// a generic async provider for bidirectional communication /// a generic async provider for bidirectional communication
pub mod controller; pub mod controller;
/// a helper trait to generate operation sequences /// a generic representation of a text change
pub mod factory; pub mod change;
pub use controller::Controller; pub use controller::Controller;
pub use factory::OperationFactory; pub use change::TextChange;

87
src/buffer/controller.rs
View file

@ -3,14 +3,16 @@
//! a controller implementation for buffer actions //! a controller implementation for buffer actions
use operational_transform::OperationSeq; use std::sync::Arc;
use tokio::sync::{watch, mpsc, Mutex, oneshot};
use tokio::sync::oneshot;
use tokio::sync::{watch, mpsc, RwLock};
use tonic::async_trait; use tonic::async_trait;
use crate::errors::IgnorableError; use crate::errors::IgnorableError;
use crate::{api::Controller, Error}; use crate::api::Controller;
use super::TextChange; use crate::api::TextChange;
/// the buffer controller implementation /// the buffer controller implementation
/// ///
@ -24,29 +26,31 @@ use super::TextChange;
/// queues, transforming outbound delayed ops and applying remote changes /// queues, transforming outbound delayed ops and applying remote changes
/// to the local buffer /// to the local buffer
/// ///
/// this controller implements [crate::api::OperationFactory], allowing to produce
/// Operation Sequences easily
///
/// upon dropping this handle will stop the associated worker /// upon dropping this handle will stop the associated worker
#[derive(Debug)] #[derive(Debug, Clone)]
pub struct BufferController { pub struct BufferController {
/// unique identifier of buffer
pub name: String,
content: watch::Receiver<String>, content: watch::Receiver<String>,
operations: mpsc::UnboundedSender<OperationSeq>, seen: Arc<RwLock<String>>,
last_op: Mutex<watch::Receiver<()>>, operations: mpsc::UnboundedSender<TextChange>,
stream: mpsc::UnboundedSender<oneshot::Sender<Option<TextChange>>>, poller: mpsc::Sender<oneshot::Sender<()>>,
stop: mpsc::UnboundedSender<()>, _stop: Arc<StopOnDrop>, // just exist
} }
impl BufferController { impl BufferController {
pub(crate) fn new( pub(crate) fn new(
name: String,
content: watch::Receiver<String>, content: watch::Receiver<String>,
operations: mpsc::UnboundedSender<OperationSeq>, operations: mpsc::UnboundedSender<TextChange>,
stream: mpsc::UnboundedSender<oneshot::Sender<Option<TextChange>>>, poller: mpsc::Sender<oneshot::Sender<()>>,
stop: mpsc::UnboundedSender<()>, stop: mpsc::UnboundedSender<()>,
last_op: Mutex<watch::Receiver<()>>,
) -> Self { ) -> Self {
BufferController { BufferController {
last_op, content, operations, stream, stop, name,
content, operations, poller,
_stop: Arc::new(StopOnDrop(stop)),
seen: Arc::new(RwLock::new("".into())),
} }
} }
@ -55,40 +59,53 @@ impl BufferController {
} }
} }
impl Drop for BufferController { #[derive(Debug)]
struct StopOnDrop(mpsc::UnboundedSender<()>);
impl Drop for StopOnDrop {
fn drop(&mut self) { fn drop(&mut self) {
self.stop.send(()).unwrap_or_warn("could not send stop message to worker"); self.0.send(()).unwrap_or_warn("could not send stop message to worker");
} }
} }
#[async_trait] #[async_trait]
impl Controller<TextChange> for BufferController { impl Controller<TextChange> for BufferController {
type Input = OperationSeq; type Input = TextChange;
async fn poll(&self) -> Result<(), Error> { async fn poll(&self) -> crate::Result<()> {
Ok(self.last_op.lock().await.changed().await?) let (tx, rx) = oneshot::channel::<()>();
self.poller.send(tx);
Ok(rx.await.map_err(|_| crate::Error::Channel { send: false })?)
} }
fn try_recv(&self) -> Result<Option<TextChange>, Error> { fn try_recv(&self) -> crate::Result<Option<TextChange>> {
let (tx, rx) = oneshot::channel(); let seen = match self.seen.try_read() {
self.stream.send(tx)?; Err(_) => return Err(crate::Error::Deadlocked),
rx.blocking_recv() Ok(x) => x.clone(),
.map_err(|_| Error::Channel { send: false }) };
let actual = self.content.borrow().clone();
if seen == actual {
return Ok(None);
}
let change = TextChange::from_diff(&seen, &actual);
match self.seen.try_write() {
Err(_) => return Err(crate::Error::Deadlocked),
Ok(mut w) => *w = actual,
};
Ok(Some(change))
} }
async fn recv(&self) -> Result<TextChange, Error> { async fn recv(&self) -> crate::Result<TextChange> {
self.poll().await?; self.poll().await?;
let (tx, rx) = oneshot::channel(); let cur = self.seen.read().await.clone();
self.stream.send(tx)?; let change = TextChange::from_diff(&cur, &self.content.borrow());
Ok( let mut seen = self.seen.write().await;
rx.await *seen = self.content.borrow().clone();
.map_err(|_| Error::Channel { send: false })? Ok(change)
.expect("empty channel after polling")
)
} }
/// enqueue an opseq for processing /// enqueue an opseq for processing
fn send(&self, op: OperationSeq) -> Result<(), Error> { fn send(&self, op: TextChange) -> crate::Result<()> {
Ok(self.operations.send(op)?) Ok(self.operations.send(op)?)
} }
} }

15
src/buffer/mod.rs
View file

@ -12,18 +12,3 @@ pub mod controller;
pub(crate) mod worker; pub(crate) mod worker;
pub use controller::BufferController as Controller; pub use controller::BufferController as Controller;
/// an editor-friendly representation of a text change in a buffer
///
/// TODO move in proto
#[derive(Clone, Debug, Default)]
pub struct TextChange {
/// range of text change, as byte indexes in buffer
pub span: std::ops::Range<usize>,
/// content of text change, as string
pub content: String,
/// content after this text change
/// note that this field will probably be dropped, don't rely on it
pub after: String
}

217
src/buffer/worker.rs
View file

@ -1,64 +1,65 @@
use std::collections::VecDeque; use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use operational_transform::OperationSeq; use similar::{TextDiff, ChangeTag};
use tokio::sync::{watch, mpsc, oneshot, Mutex}; use tokio::sync::{watch, mpsc, oneshot};
use tonic::transport::Channel; use tonic::transport::Channel;
use tonic::{async_trait, Streaming}; use tonic::{async_trait, Streaming};
use woot::crdt::{Op, CRDT, TextEditor};
use woot::woot::Woot;
use crate::errors::IgnorableError;
use crate::proto::{OperationRequest, RawOp}; use crate::proto::{OperationRequest, RawOp};
use crate::proto::buffer_client::BufferClient; use crate::proto::buffer_client::BufferClient;
use crate::api::controller::ControllerWorker; use crate::api::controller::ControllerWorker;
use crate::api::factory::{leading_noop, tailing_noop}; use crate::api::TextChange;
use super::TextChange;
use super::controller::BufferController; use super::controller::BufferController;
pub(crate) struct BufferControllerWorker { pub(crate) struct BufferControllerWorker {
uid: String, uid: String,
content: watch::Sender<String>, content: watch::Sender<String>,
operations: mpsc::UnboundedReceiver<OperationSeq>, operations: mpsc::UnboundedReceiver<TextChange>,
stream: mpsc::UnboundedReceiver<oneshot::Sender<Option<TextChange>>>,
stream_requestor: mpsc::UnboundedSender<oneshot::Sender<Option<TextChange>>>,
receiver: watch::Receiver<String>, receiver: watch::Receiver<String>,
sender: mpsc::UnboundedSender<OperationSeq>, sender: mpsc::UnboundedSender<TextChange>,
buffer: String, buffer: Woot,
path: String, name: String,
stop: mpsc::UnboundedReceiver<()>, stop: mpsc::UnboundedReceiver<()>,
stop_control: mpsc::UnboundedSender<()>, stop_control: mpsc::UnboundedSender<()>,
new_op_tx: watch::Sender<()>, poller_rx: mpsc::Receiver<oneshot::Sender<()>>,
new_op_rx: watch::Receiver<()>, poller_tx: mpsc::Sender<oneshot::Sender<()>>,
pollers: Vec<oneshot::Sender<()>>,
} }
impl BufferControllerWorker { impl BufferControllerWorker {
pub fn new(uid: String, buffer: &str, path: &str) -> Self { pub fn new(uid: String, path: &str) -> Self {
let (txt_tx, txt_rx) = watch::channel(buffer.to_string()); let (txt_tx, txt_rx) = watch::channel("".to_string());
let (op_tx, op_rx) = mpsc::unbounded_channel(); let (op_tx, op_rx) = mpsc::unbounded_channel();
let (s_tx, s_rx) = mpsc::unbounded_channel();
let (end_tx, end_rx) = mpsc::unbounded_channel(); let (end_tx, end_rx) = mpsc::unbounded_channel();
let (notx, norx) = watch::channel(()); let (poller_tx, poller_rx) = mpsc::channel(10);
let mut hasher = DefaultHasher::new();
uid.hash(&mut hasher);
let site_id = hasher.finish() as usize;
BufferControllerWorker { BufferControllerWorker {
uid, uid, poller_rx, poller_tx,
pollers: Vec::new(),
content: txt_tx, content: txt_tx,
operations: op_rx, operations: op_rx,
stream: s_rx,
stream_requestor: s_tx,
receiver: txt_rx, receiver: txt_rx,
sender: op_tx, sender: op_tx,
buffer: buffer.to_string(), buffer: Woot::new(site_id % (2<<10), ""), // TODO remove the modulo, only for debugging!
path: path.to_string(), name: path.to_string(),
stop: end_rx, stop: end_rx,
stop_control: end_tx, stop_control: end_tx,
new_op_tx: notx,
new_op_rx: norx,
} }
} }
async fn send_op(&self, tx: &mut BufferClient<Channel>, outbound: &OperationSeq) -> crate::Result<()> { async fn send_op(&self, tx: &mut BufferClient<Channel>, outbound: &Op) -> crate::Result<()> {
let opseq = serde_json::to_string(outbound).expect("could not serialize opseq"); let opseq = serde_json::to_string(outbound).expect("could not serialize opseq");
let req = OperationRequest { let req = OperationRequest {
path: self.path.clone(), path: self.name.clone(),
hash: format!("{:x}", md5::compute(&self.buffer)), hash: format!("{:x}", md5::compute(self.buffer.view())),
op: Some(RawOp { op: Some(RawOp {
opseq, user: self.uid.clone(), opseq, user: self.uid.clone(),
}), }),
@ -76,125 +77,91 @@ impl ControllerWorker<TextChange> for BufferControllerWorker {
fn subscribe(&self) -> BufferController { fn subscribe(&self) -> BufferController {
BufferController::new( BufferController::new(
self.name.clone(),
self.receiver.clone(), self.receiver.clone(),
self.sender.clone(), self.sender.clone(),
self.stream_requestor.clone(), self.poller_tx.clone(),
self.stop_control.clone(), self.stop_control.clone(),
Mutex::new(self.new_op_rx.clone()),
) )
} }
async fn work(mut self, mut tx: Self::Tx, mut rx: Self::Rx) { async fn work(mut self, mut tx: Self::Tx, mut rx: Self::Rx) {
let mut clientside : VecDeque<OperationSeq> = VecDeque::new();
let mut serverside : VecDeque<OperationSeq> = VecDeque::new();
loop { loop {
// block until one of these is ready // block until one of these is ready
tokio::select! { tokio::select! {
biased; biased;
// received a stop request (or channel got closed) // received stop signal
res = self.stop.recv() => { _ = self.stop.recv() => break,
tracing::info!("received stop signal");
match res {
None => return tracing::warn!("stop channel closed, stopping worker"),
Some(()) => return tracing::debug!("buffer worker stopping cleanly"),
}
}
// received a new message from server (or an error) // received a new poller, add it to collection
res = rx.message() => { res = self.poller_rx.recv() => match res {
tracing::info!("received msg from server"); None => break tracing::error!("poller channel closed"),
let inbound : OperationSeq = match res { Some(tx) => self.pollers.push(tx),
Err(e) => return tracing::error!("error receiving op from server: {}", e), },
Ok(None) => return tracing::warn!("server closed operation stream"),
Ok(Some(msg)) => serde_json::from_str(&msg.opseq) // received a text change from editor
.expect("could not deserialize server opseq"), res = self.operations.recv() => match res {
}; None => break,
self.buffer = inbound.apply(&self.buffer).expect("could not apply remote opseq???"); Some(change) => {
serverside.push_back(inbound); match self.buffer.view().get(change.span.clone()) {
while let Some(mut outbound) = clientside.get(0).cloned() { None => tracing::error!("received illegal span from client"),
let mut serverside_tmp = serverside.clone(); Some(span) => {
for server_op in serverside_tmp.iter_mut() { let diff = TextDiff::from_chars(span, &change.content);
tracing::info!("transforming {:?} <-> {:?}", outbound, server_op);
(outbound, *server_op) = outbound.transform(server_op) let mut i = 0;
.expect("could not transform enqueued out with just received"); let mut ops = Vec::new();
} for diff in diff.iter_all_changes() {
match self.send_op(&mut tx, &outbound).await { match diff.tag() {
Err(e) => { tracing::warn!("could not send op even after transforming: {}", e); break; }, ChangeTag::Equal => i += 1,
Ok(()) => { ChangeTag::Delete => match self.buffer.delete(change.span.start + i) {
tracing::info!("back in sync"); Ok(op) => ops.push(op),
serverside = serverside_tmp; Err(e) => tracing::error!("could not apply deletion: {}", e),
self.buffer = outbound.apply(&self.buffer).expect("could not apply op after synching back"); },
clientside.pop_front(); 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),
}
}
},
}
}
for op in ops {
match self.send_op(&mut tx, &op).await {
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");
},
}
}
}, },
} }
} }
self.content.send(self.buffer.clone()).expect("could not broadcast buffer update");
self.new_op_tx.send(()).expect("could not activate client after new server event");
}, },
// received a new operation from client (or channel got closed) // received a message from server
res = self.operations.recv() => { res = rx.message() => match res {
tracing::info!("received op from client"); Err(_e) => break,
match res { Ok(None) => break,
None => return tracing::warn!("client closed operation stream"), Ok(Some(change)) => match serde_json::from_str::<Op>(&change.opseq) {
Some(op) => { Ok(op) => {
if clientside.is_empty() { self.buffer.merge(op);
match self.send_op(&mut tx, &op).await { self.content.send(self.buffer.view()).unwrap_or_warn("could not send buffer update");
Ok(()) => { for tx in self.pollers.drain(0..self.pollers.len()) {
self.buffer = op.apply(&self.buffer).expect("could not apply op"); tx.send(()).unwrap_or_warn("could not wake up poller");
self.content.send(self.buffer.clone()).expect("could not update buffer view");
},
Err(e) => {
tracing::warn!("server rejected op: {}", e);
clientside.push_back(op);
},
}
} else { // I GET STUCK IN THIS BRANCH AND NOTHING HAPPENS AAAAAAAAAA
clientside.push_back(op);
} }
} },
} Err(e) => tracing::error!("could not deserialize operation from server: {}", e),
},
}, },
// client requested a server operation, transform it and send it
res = self.stream.recv() => {
tracing::info!("received op REQUEST from client");
match res {
None => return tracing::error!("client closed requestor stream"),
Some(tx) => tx.send(match serverside.pop_front() {
None => {
tracing::warn!("requested change but none is available");
None
},
Some(mut operation) => {
let mut after = self.buffer.clone();
for op in clientside.iter_mut() {
(*op, operation) = match op.transform(&operation) {
Err(e) => return tracing::warn!("could not transform enqueued operation: {}", e),
Ok((x, y)) => (x, y),
};
after = match op.apply(&after) {
Err(_) => return tracing::error!("could not apply outgoing enqueued opseq to current buffer?"),
Ok(x) => x,
};
}
let skip = leading_noop(operation.ops()) as usize;
let tail = tailing_noop(operation.ops()) as usize;
let span = skip..(operation.base_len() - tail);
let content = if after.len() - tail < skip { "".into() } else { after[skip..after.len()-tail].to_string() };
let change = TextChange { span, content, after };
Some(change)
},
}).expect("client did not wait????"),
}
},
} }
} }
} }
} }

54
src/client.rs
View file

@ -4,9 +4,12 @@
use std::{sync::Arc, collections::BTreeMap}; use std::{sync::Arc, collections::BTreeMap};
use tokio::sync::mpsc;
use tokio_stream::StreamExt;
use tonic::transport::Channel; use tonic::transport::Channel;
use crate::{ use crate::{
api::Controller,
cursor::{worker::CursorControllerWorker, controller::CursorController}, cursor::{worker::CursorControllerWorker, controller::CursorController},
proto::{ proto::{
buffer_client::BufferClient, cursor_client::CursorClient, UserIdentity, BufferPayload, buffer_client::BufferClient, cursor_client::CursorClient, UserIdentity, BufferPayload,
@ -76,7 +79,7 @@ impl Client {
/// ///
/// to interact with such workspace [crate::api::Controller::send] cursor events or /// to interact with such workspace [crate::api::Controller::send] cursor events or
/// [crate::api::Controller::recv] for events on the associated [crate::cursor::Controller]. /// [crate::api::Controller::recv] for events on the associated [crate::cursor::Controller].
pub async fn join(&mut self, _session: &str) -> Result<Arc<CursorController>, Error> { pub async fn join(&mut self, _session: &str) -> crate::Result<Arc<CursorController>> {
// TODO there is no real workspace handling in codemp server so it behaves like one big global // TODO there is no real workspace handling in codemp server so it behaves like one big global
// session. I'm still creating this to start laying out the proper use flow // session. I'm still creating this to start laying out the proper use flow
let stream = self.client.cursor.listen(UserIdentity { id: "".into() }).await?.into_inner(); let stream = self.client.cursor.listen(UserIdentity { id: "".into() }).await?.into_inner();
@ -103,7 +106,7 @@ impl Client {
} }
/// create a new buffer in current workspace, with optional given content /// create a new buffer in current workspace, with optional given content
pub async fn create(&mut self, path: &str, content: Option<&str>) -> Result<(), Error> { pub async fn create(&mut self, path: &str, content: Option<&str>) -> crate::Result<()> {
if let Some(_workspace) = &self.workspace { if let Some(_workspace) = &self.workspace {
self.client.buffer self.client.buffer
.create(BufferPayload { .create(BufferPayload {
@ -120,23 +123,18 @@ impl Client {
/// attach to a buffer, starting a buffer controller and returning a new reference to it /// attach to a buffer, starting a buffer controller and returning a new reference to it
/// ///
/// to interact with such buffer [crate::api::Controller::send] operation sequences /// to interact with such buffer use [crate::api::Controller::send] or
/// or [crate::api::Controller::recv] for text events using its [crate::buffer::Controller]. /// [crate::api::Controller::recv] to exchange [crate::api::TextChange]
/// to generate operation sequences use the [crate::api::OperationFactory] pub async fn attach(&mut self, path: &str) -> crate::Result<Arc<BufferController>> {
/// methods, which are implemented on [crate::buffer::Controller], such as
/// [crate::api::OperationFactory::diff].
pub async fn attach(&mut self, path: &str) -> Result<Arc<BufferController>, Error> {
if let Some(workspace) = &mut self.workspace { if let Some(workspace) = &mut self.workspace {
let mut client = self.client.buffer.clone(); let mut client = self.client.buffer.clone();
let req = BufferPayload { let req = BufferPayload {
path: path.to_string(), user: self.id.clone(), content: None path: path.to_string(), user: self.id.clone(), content: None
}; };
let content = client.sync(req.clone()).await?.into_inner().content;
let stream = client.attach(req).await?.into_inner(); let stream = client.attach(req).await?.into_inner();
let controller = BufferControllerWorker::new(self.id.clone(), &content, path); let controller = BufferControllerWorker::new(self.id.clone(), path);
let handler = Arc::new(controller.subscribe()); let handler = Arc::new(controller.subscribe());
let _path = path.to_string(); let _path = path.to_string();
@ -153,4 +151,38 @@ impl Client {
Err(Error::InvalidState { msg: "join a workspace first".into() }) Err(Error::InvalidState { msg: "join a workspace first".into() })
} }
} }
pub async fn select_buffer(&self) -> crate::Result<String> {
match &self.workspace {
None => Err(Error::InvalidState { msg: "join workspace first".into() }),
Some(workspace) => {
let (tx, mut rx) = mpsc::unbounded_channel();
let mut tasks = Vec::new();
for (id, buffer) in workspace.buffers.iter() {
let _tx = tx.clone();
let _id = id.clone();
let _buffer = buffer.clone();
tasks.push(tokio::spawn(async move {
match _buffer.poll().await {
Ok(()) => _tx.send(Ok(_id)),
Err(_) => _tx.send(Err(Error::Channel { send: true })),
}
}))
}
loop {
match rx.recv().await {
None => return Err(Error::Channel { send: false }),
Some(Err(_)) => continue, // TODO log errors
Some(Ok(x)) => {
for t in tasks {
t.abort();
}
return Ok(x.clone());
},
}
}
}
}
}
} }

2
src/cursor/controller.rs
View file

@ -52,7 +52,7 @@ impl Controller<CursorEvent> for CursorController {
/// enqueue a cursor event to be broadcast to current workspace /// enqueue a cursor event to be broadcast to current workspace
/// will automatically invert cursor start/end if they are inverted /// will automatically invert cursor start/end if they are inverted
fn send(&self, mut cursor: CursorPosition) -> Result<(), Error> { fn send(&self, mut cursor: CursorPosition) -> Result<(), Error> {
if cursor.start() < cursor.end() { if cursor.start() > cursor.end() {
std::mem::swap(&mut cursor.start, &mut cursor.end); std::mem::swap(&mut cursor.start, &mut cursor.end);
} }
Ok(self.op.send(CursorEvent { Ok(self.op.send(CursorEvent {

11
src/cursor/mod.rs
View file

@ -26,13 +26,20 @@ impl From::<(i32, i32)> for RowCol {
} }
} }
impl RowCol {
/// create a RowCol and wrap into an Option, to help build protocol packets
pub fn wrap(row: i32, col: i32) -> Option<RowCol> {
Some(RowCol { row, col })
}
}
impl CursorPosition { impl CursorPosition {
/// extract start position, defaulting to (0,0) /// extract start position, defaulting to (0,0), to help build protocol packets
pub fn start(&self) -> RowCol { pub fn start(&self) -> RowCol {
self.start.clone().unwrap_or((0, 0).into()) self.start.clone().unwrap_or((0, 0).into())
} }
/// extract end position, defaulting to (0,0) /// extract end position, defaulting to (0,0), to help build protocol packets
pub fn end(&self) -> RowCol { pub fn end(&self) -> RowCol {
self.end.clone().unwrap_or((0, 0).into()) self.end.clone().unwrap_or((0, 0).into())
} }

3
src/errors.rs
View file

@ -61,6 +61,9 @@ pub enum Error {
msg: String, msg: String,
}, },
/// errors caused by wrong interlocking, safe to retry
Deadlocked,
/// if you see these errors someone is being lazy (: /// if you see these errors someone is being lazy (:
Filler { // TODO filler error, remove later Filler { // TODO filler error, remove later
message: String, message: String,

48
src/instance.rs
View file

@ -46,13 +46,13 @@ pub mod a_sync {
impl Instance { impl Instance {
/// connect to remote address instantiating a new client [crate::client::Client::new] /// connect to remote address instantiating a new client [crate::client::Client::new]
pub async fn connect(&self, addr: &str) -> Result<(), Error> { pub async fn connect(&self, addr: &str) -> crate::Result<()> {
*self.client.lock().await = Some(Client::new(addr).await?); *self.client.lock().await = Some(Client::new(addr).await?);
Ok(()) Ok(())
} }
/// threadsafe version of [crate::client::Client::join] /// threadsafe version of [crate::client::Client::join]
pub async fn join(&self, session: &str) -> Result<Arc<CursorController>, Error> { pub async fn join(&self, session: &str) -> crate::Result<Arc<CursorController>> {
self.client self.client
.lock().await .lock().await
.as_mut() .as_mut()
@ -62,7 +62,7 @@ pub mod a_sync {
} }
/// threadsafe version of [crate::client::Client::create] /// threadsafe version of [crate::client::Client::create]
pub async fn create(&self, path: &str, content: Option<&str>) -> Result<(), Error> { pub async fn create(&self, path: &str, content: Option<&str>) -> crate::Result<()> {
self.client self.client
.lock().await .lock().await
.as_mut() .as_mut()
@ -72,7 +72,7 @@ pub mod a_sync {
} }
/// threadsafe version of [crate::client::Client::attach] /// threadsafe version of [crate::client::Client::attach]
pub async fn attach(&self, path: &str) -> Result<Arc<BufferController>, Error> { pub async fn attach(&self, path: &str) -> crate::Result<Arc<BufferController>> {
self.client self.client
.lock().await .lock().await
.as_mut() .as_mut()
@ -82,7 +82,7 @@ pub mod a_sync {
} }
/// threadsafe version of [crate::client::Client::get_cursor] /// threadsafe version of [crate::client::Client::get_cursor]
pub async fn get_cursor(&self) -> Result<Arc<CursorController>, Error> { pub async fn get_cursor(&self) -> crate::Result<Arc<CursorController>> {
self.client self.client
.lock().await .lock().await
.as_mut() .as_mut()
@ -92,7 +92,7 @@ pub mod a_sync {
} }
/// threadsafe version of [crate::client::Client::get_buffer] /// threadsafe version of [crate::client::Client::get_buffer]
pub async fn get_buffer(&self, path: &str) -> Result<Arc<BufferController>, Error> { pub async fn get_buffer(&self, path: &str) -> crate::Result<Arc<BufferController>> {
self.client self.client
.lock().await .lock().await
.as_mut() .as_mut()
@ -102,7 +102,7 @@ pub mod a_sync {
} }
/// threadsafe version of [crate::client::Client::leave_workspace] /// threadsafe version of [crate::client::Client::leave_workspace]
pub async fn leave_workspace(&self) -> Result<(), Error> { pub async fn leave_workspace(&self) -> crate::Result<()> {
self.client self.client
.lock().await .lock().await
.as_mut() .as_mut()
@ -112,7 +112,7 @@ pub mod a_sync {
} }
/// threadsafe version of [crate::client::Client::disconnect_buffer] /// threadsafe version of [crate::client::Client::disconnect_buffer]
pub async fn disconnect_buffer(&self, path: &str) -> Result<bool, Error> { pub async fn disconnect_buffer(&self, path: &str) -> crate::Result<bool> {
let res = self.client let res = self.client
.lock().await .lock().await
.as_mut() .as_mut()
@ -120,6 +120,16 @@ pub mod a_sync {
.disconnect_buffer(path); .disconnect_buffer(path);
Ok(res) Ok(res)
} }
pub async fn select_buffer(&self) -> crate::Result<String> {
let res = self.client
.lock().await
.as_ref()
.ok_or(Error::InvalidState { msg: "connect first".into() })?
.select_buffer()
.await?;
Ok(res)
}
} }
} }
@ -135,7 +145,7 @@ pub mod sync {
buffer::controller::BufferController buffer::controller::BufferController
}; };
/// persistant session manager for codemp client /// persistent session manager for codemp client
/// ///
/// will hold a std mutex over an optional client, and drop its reference when disconnecting. /// will hold a std mutex over an optional client, and drop its reference when disconnecting.
/// also contains a tokio runtime to execute async futures on /// also contains a tokio runtime to execute async futures on
@ -157,7 +167,7 @@ pub mod sync {
} }
impl Instance { impl Instance {
fn if_client<T>(&self, op: impl FnOnce(&mut Client) -> T) -> Result<T, Error> { fn if_client<T>(&self, op: impl FnOnce(&mut Client) -> T) -> crate::Result<T> {
if let Some(c) = self.client.lock().expect("client mutex poisoned").as_mut() { if let Some(c) = self.client.lock().expect("client mutex poisoned").as_mut() {
Ok(op(c)) Ok(op(c))
} else { } else {
@ -175,38 +185,42 @@ pub mod sync {
} }
/// threadsafe and sync version of [crate::client::Client::join] /// threadsafe and sync version of [crate::client::Client::join]
pub fn join(&self, session: &str) -> Result<Arc<CursorController>, Error> { pub fn join(&self, session: &str) -> crate::Result<Arc<CursorController>> {
self.if_client(|c| self.rt().block_on(c.join(session)))? self.if_client(|c| self.rt().block_on(c.join(session)))?
} }
/// threadsafe and sync version of [crate::client::Client::create] /// threadsafe and sync version of [crate::client::Client::create]
pub fn create(&self, path: &str, content: Option<&str>) -> Result<(), Error> { pub fn create(&self, path: &str, content: Option<&str>) -> crate::Result<()> {
self.if_client(|c| self.rt().block_on(c.create(path, content)))? self.if_client(|c| self.rt().block_on(c.create(path, content)))?
} }
/// threadsafe and sync version of [crate::client::Client::attach] /// threadsafe and sync version of [crate::client::Client::attach]
pub fn attach(&self, path: &str) -> Result<Arc<BufferController>, Error> { pub fn attach(&self, path: &str) -> crate::Result<Arc<BufferController>> {
self.if_client(|c| self.rt().block_on(c.attach(path)))? self.if_client(|c| self.rt().block_on(c.attach(path)))?
} }
/// threadsafe and sync version of [crate::client::Client::get_cursor] /// threadsafe and sync version of [crate::client::Client::get_cursor]
pub fn get_cursor(&self) -> Result<Arc<CursorController>, Error> { pub fn get_cursor(&self) -> crate::Result<Arc<CursorController>> {
self.if_client(|c| c.get_cursor().ok_or(Error::InvalidState { msg: "join workspace first".into() }))? self.if_client(|c| c.get_cursor().ok_or(Error::InvalidState { msg: "join workspace first".into() }))?
} }
/// threadsafe and sync version of [crate::client::Client::get_buffer] /// threadsafe and sync version of [crate::client::Client::get_buffer]
pub fn get_buffer(&self, path: &str) -> Result<Arc<BufferController>, Error> { pub fn get_buffer(&self, path: &str) -> crate::Result<Arc<BufferController>> {
self.if_client(|c| c.get_buffer(path).ok_or(Error::InvalidState { msg: "join workspace or create requested buffer first".into() }))? self.if_client(|c| c.get_buffer(path).ok_or(Error::InvalidState { msg: "join workspace or create requested buffer first".into() }))?
} }
/// threadsafe and sync version of [crate::client::Client::leave_workspace] /// threadsafe and sync version of [crate::client::Client::leave_workspace]
pub fn leave_workspace(&self) -> Result<(), Error> { pub fn leave_workspace(&self) -> crate::Result<()> {
self.if_client(|c| c.leave_workspace()) self.if_client(|c| c.leave_workspace())
} }
/// threadsafe and sync version of [crate::client::Client::disconnect_buffer] /// threadsafe and sync version of [crate::client::Client::disconnect_buffer]
pub fn disconnect_buffer(&self, path: &str) -> Result<bool, Error> { pub fn disconnect_buffer(&self, path: &str) -> crate::Result<bool> {
self.if_client(|c| c.disconnect_buffer(path)) self.if_client(|c| c.disconnect_buffer(path))
} }
pub fn select_buffer(&self) -> crate::Result<String> {
self.if_client(|c| self.rt().block_on(c.select_buffer()))?
}
} }
} }

42
src/lib.rs
View file

@ -2,7 +2,7 @@
//! //!
//! ![just a nice pic](https://alemi.dev/img/about-slice-1.png) //! ![just a nice pic](https://alemi.dev/img/about-slice-1.png)
//! //!
//! This is the core library of the codemp project. //! > the core library of the codemp project, driving all editor plugins
//! //!
//! ## structure //! ## structure
//! The main entrypoint is the [Instance] object, that maintains a connection and can //! The main entrypoint is the [Instance] object, that maintains a connection and can
@ -14,17 +14,14 @@
//! Blocking and callback variants are also implemented. The [api::Controller] can also be used to send new //! Blocking and callback variants are also implemented. The [api::Controller] can also be used to send new
//! events to the server ([api::Controller::send]). //! events to the server ([api::Controller::send]).
//! //!
//! Each operation on a buffer is represented as an [ot::OperationSeq]. //! Each operation on a buffer is represented as an [woot::crdt::Op]. The underlying buffer is a
//! A visualization about how OperationSeqs work is available //! [WOOT CRDT](https://inria.hal.science/file/index/docid/71240/filename/RR-5580.pdf),
//! [here](http://operational-transformation.github.io/index.html), //! but to use this library it's only sufficient to know that all WOOT buffers that have received
//! but to use this library it's only sufficient to know that they can only //! the same operations converge to the same state, and that operations might not get integrated
//! be applied on buffers of some length and are transformable to be able to be //! immediately but instead deferred until compatible.
//! applied in a different order while maintaining the same result.
//!
//! To generate Operation Sequences use helper methods from module [api::factory] (trait [api::OperationFactory]).
//! //!
//! ## features //! ## features
//! * `ot` : include the underlying operational transform library (default enabled) //! * `woot` : include the underlying CRDT library and re-exports it (default enabled)
//! * `api` : include traits for core interfaces under [api] (default enabled) //! * `api` : include traits for core interfaces under [api] (default enabled)
//! * `proto` : include GRCP protocol definitions under [proto] (default enabled) //! * `proto` : include GRCP protocol definitions under [proto] (default enabled)
//! * `client`: include the local [client] implementation (default enabled) //! * `client`: include the local [client] implementation (default enabled)
@ -40,7 +37,7 @@
//! [instance::a_sync::Instance] //! [instance::a_sync::Instance]
//! //!
//! ```rust,no_run //! ```rust,no_run
//! use codemp::api::{Controller, OperationFactory}; //! use codemp::api::{Controller, TextChange};
//! # use codemp::instance::a_sync::Instance; //! # use codemp::instance::a_sync::Instance;
//! //!
//! # async fn async_example() -> codemp::Result<()> { //! # async fn async_example() -> codemp::Result<()> {
@ -62,12 +59,9 @@
//! // attach to a new buffer and execute operations on it //! // attach to a new buffer and execute operations on it
//! session.create("test.txt", None).await?; // create new buffer //! session.create("test.txt", None).await?; // create new buffer
//! let buffer = session.attach("test.txt").await?; // attach to it //! let buffer = session.attach("test.txt").await?; // attach to it
//! let text = buffer.content(); // any string can be used as operation factory //! let local_change = TextChange { span: 0..0, content: "hello!".into() };
//! buffer.send(text.ins("hello", 0))?; // insert some text //! buffer.send(local_change)?; // insert some text
//! if let Some(operation) = text.diff(4, "o world", 5) { //! let remote_change = buffer.recv().await?;
//! buffer.send(operation)?; // replace with precision, if valid
//! }
//! assert_eq!(buffer.content(), "hello world");
//! # //! #
//! # Ok(()) //! # Ok(())
//! # } //! # }
@ -85,16 +79,6 @@
//! let session = Instance::default(); // instantiate sync variant //! let session = Instance::default(); // instantiate sync variant
//! session.connect("http://alemi.dev:50051")?; // connect to server //! session.connect("http://alemi.dev:50051")?; // connect to server
//! //!
//! // join remote workspace and handle cursor events with a callback
//! let cursor = session.join("some_workspace")?; // join workspace
//! let (stop, stop_rx) = tokio::sync::mpsc::unbounded_channel(); // create stop channel
//! Arc::new(cursor).callback( // register callback
//! session.rt(), stop_rx, // pass instance runtime and stop channel receiver
//! | cursor_event | {
//! println!("received cursor event: {:?}", cursor_event);
//! }
//! );
//!
//! // attach to buffer and blockingly receive events //! // attach to buffer and blockingly receive events
//! let buffer = session.attach("test.txt")?; // attach to buffer, must already exist //! let buffer = session.attach("test.txt")?; // attach to buffer, must already exist
//! while let Ok(op) = buffer.blocking_recv(session.rt()) { // must pass runtime //! while let Ok(op) = buffer.blocking_recv(session.rt()) { // must pass runtime
@ -172,8 +156,8 @@ pub mod instance;
pub mod prelude; pub mod prelude;
/// underlying OperationalTransform library used, re-exported /// underlying OperationalTransform library used, re-exported
#[cfg(feature = "ot")] #[cfg(feature = "woot")]
pub use operational_transform as ot; pub use woot;
/// protocol types and services auto-generated by grpc /// protocol types and services auto-generated by grpc
#[cfg(feature = "proto")] #[cfg(feature = "proto")]

13
src/prelude.rs
View file

@ -7,22 +7,21 @@ pub use crate::{
Result as CodempResult, Result as CodempResult,
}; };
#[cfg(feature = "ot")] #[cfg(feature = "woot")]
pub use crate::ot::OperationSeq as CodempOperationSeq; pub use crate::woot::crdt::Op as CodempOp;
#[cfg(feature = "api")] #[cfg(feature = "api")]
pub use crate::{ pub use crate::api::{
api::Controller as CodempController, Controller as CodempController,
api::OperationFactory as CodempOperationFactory, TextChange as CodempTextChange,
}; };
#[cfg(feature = "client")] #[cfg(feature = "client")]
pub use crate::{ pub use crate::{
Instance as CodempInstance,
client::Client as CodempClient, client::Client as CodempClient,
cursor::Controller as CodempCursorController, cursor::Controller as CodempCursorController,
buffer::Controller as CodempBufferController, buffer::Controller as CodempBufferController,
buffer::TextChange as CodempTextChange,
Instance as CodempInstance,
}; };
#[cfg(feature = "proto")] #[cfg(feature = "proto")]