hexedtech/codemp
3
0
Fork 0
mirror of https://github.com/hexedtech/codemp.git synced 2024-11-21 23:04:49 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix: properly update branch version

Browse source
This commit is contained in:
əlemi 2024-10-27 19:38:50 +01:00 committed by alemi.dev
parent adb9c1c1d8
commit 79204972f3
2 changed files with 21 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
src/buffer/controller.rs
View file

@ -56,7 +56,7 @@ pub(crate) struct BufferControllerInner {
pub(crate) ops_in: mpsc::UnboundedSender<TextChange>, pub(crate) ops_in: mpsc::UnboundedSender<TextChange>,
pub(crate) poller: mpsc::UnboundedSender<oneshot::Sender<()>>, pub(crate) poller: mpsc::UnboundedSender<oneshot::Sender<()>>,
pub(crate) content_request: mpsc::Sender<oneshot::Sender<String>>, pub(crate) content_request: mpsc::Sender<oneshot::Sender<String>>,
pub(crate) delta_request: mpsc::Sender<(LocalVersion, oneshot::Sender<Option<BufferUpdate>>)>, pub(crate) delta_request: mpsc::Sender<oneshot::Sender<Option<BufferUpdate>>>,
pub(crate) callback: watch::Sender<Option<ControllerCallback<BufferController>>>, pub(crate) callback: watch::Sender<Option<ControllerCallback<BufferController>>>,
pub(crate) ack_tx: mpsc::UnboundedSender<LocalVersion>, pub(crate) ack_tx: mpsc::UnboundedSender<LocalVersion>,
} }
@ -93,7 +93,7 @@ impl AsyncReceiver<BufferUpdate> for BufferController {
} }
let (tx, rx) = oneshot::channel(); let (tx, rx) = oneshot::channel();
self.0.delta_request.send((last_update, tx)).await?; self.0.delta_request.send(tx).await?;
Ok(rx.await?) Ok(rx.await?)
} }

40
src/buffer/worker.rs
View file

@ -1,5 +1,6 @@
use std::sync::Arc; use std::sync::Arc;
use diamond_types::list::encoding::EncodeOptions;
use diamond_types::list::{Branch, OpLog}; use diamond_types::list::{Branch, OpLog};
use diamond_types::LocalVersion; use diamond_types::LocalVersion;
use tokio::sync::{mpsc, oneshot, watch}; use tokio::sync::{mpsc, oneshot, watch};
@ -25,7 +26,7 @@ struct BufferWorker {
poller: mpsc::UnboundedReceiver<oneshot::Sender<()>>, poller: mpsc::UnboundedReceiver<oneshot::Sender<()>>,
pollers: Vec<oneshot::Sender<()>>, pollers: Vec<oneshot::Sender<()>>,
content_checkout: mpsc::Receiver<oneshot::Sender<String>>, content_checkout: mpsc::Receiver<oneshot::Sender<String>>,
delta_req: mpsc::Receiver<(LocalVersion, oneshot::Sender<Option<BufferUpdate>>)>, delta_req: mpsc::Receiver<oneshot::Sender<Option<BufferUpdate>>>,
controller: std::sync::Weak<BufferControllerInner>, controller: std::sync::Weak<BufferControllerInner>,
callback: watch::Receiver<Option<ControllerCallback<BufferController>>>, callback: watch::Receiver<Option<ControllerCallback<BufferController>>>,
oplog: OpLog, oplog: OpLog,
@ -107,20 +108,22 @@ impl BufferController {
tokio::select! { tokio::select! {
biased; biased;
// received a new poller, add it to collection
res = worker.poller.recv() => match res {
None => break tracing::error!("poller channel closed"),
Some(tx) => worker.pollers.push(tx),
},
// received new change ack, merge editor branch up to that version // received new change ack, merge editor branch up to that version
res = worker.ack_rx.recv() => match res { res = worker.ack_rx.recv() => match res {
None => break tracing::error!("ack channel closed"), None => break tracing::error!("ack channel closed"),
Some(v) => { Some(v) => {
worker.branch.merge(&worker.oplog, &v) worker.branch.merge(&worker.oplog, &v);
worker.local_version.send(worker.branch.local_version())
.unwrap_or_warn("could not ack local version");
}, },
}, },
// received a new poller, add it to collection
res = worker.poller.recv() => match res {
None => break tracing::error!("poller channel closed"),
Some(tx) => worker.pollers.push(tx),
},
// received a text change from editor // received a text change from editor
res = worker.ops_in.recv() => match res { res = worker.ops_in.recv() => match res {
None => break tracing::debug!("stopping: editor closed channel"), None => break tracing::debug!("stopping: editor closed channel"),
@ -137,7 +140,7 @@ impl BufferController {
// controller is ready to apply change and recv(), calculate it and send it back // controller is ready to apply change and recv(), calculate it and send it back
res = worker.delta_req.recv() => match res { res = worker.delta_req.recv() => match res {
None => break tracing::error!("no more active controllers: can't send changes"), None => break tracing::error!("no more active controllers: can't send changes"),
Some((last_ver, tx)) => worker.handle_delta_request(last_ver, tx).await, Some(tx) => worker.handle_delta_request(tx).await,
}, },
// received a request for full CRDT content // received a request for full CRDT content
@ -145,8 +148,11 @@ impl BufferController {
None => break tracing::error!("no more active controllers: can't update content"), None => break tracing::error!("no more active controllers: can't update content"),
Some(tx) => { Some(tx) => {
worker.branch.merge(&worker.oplog, worker.oplog.local_version_ref()); worker.branch.merge(&worker.oplog, worker.oplog.local_version_ref());
worker.local_version.send(worker.branch.local_version())
.unwrap_or_warn("could not checkout local version");
let content = worker.branch.content().to_string(); let content = worker.branch.content().to_string();
tx.send(content).unwrap_or_warn("checkout request dropped"); tx.send(content)
.unwrap_or_warn("checkout request dropped");
}, },
} }
} }
@ -184,7 +190,7 @@ impl BufferWorker {
if change.is_delete() || change.is_insert() { if change.is_delete() || change.is_insert() {
tx.send(Operation { tx.send(Operation {
data: self.oplog.encode_from(Default::default(), &last_ver), data: self.oplog.encode_from(EncodeOptions::default(), &last_ver),
}) })
.await .await
.unwrap_or_warn("failed to send change!"); .unwrap_or_warn("failed to send change!");
@ -221,11 +227,8 @@ impl BufferWorker {
} }
} }
async fn handle_delta_request( async fn handle_delta_request(&mut self, tx: oneshot::Sender<Option<BufferUpdate>>) {
&mut self, let last_ver = self.branch.local_version();
last_ver: LocalVersion,
tx: oneshot::Sender<Option<BufferUpdate>>,
) {
if let Some((lv, Some(dtop))) = self if let Some((lv, Some(dtop))) = self
.oplog .oplog
.iter_xf_operations_from(&last_ver, self.oplog.local_version_ref()) .iter_xf_operations_from(&last_ver, self.oplog.local_version_ref())
@ -235,8 +238,6 @@ impl BufferWorker {
// this step_ver will be the version after we apply the operation // this step_ver will be the version after we apply the operation
// we give it to the controller so that he knows where it's at. // we give it to the controller so that he knows where it's at.
let step_ver = self.oplog.version_union(&[lv.end - 1], &last_ver); let step_ver = self.oplog.version_union(&[lv.end - 1], &last_ver);
self.branch.merge(&self.oplog, &step_ver);
let new_local_v = self.branch.local_version();
let hash = if self.timer.step() { let hash = if self.timer.step() {
Some(crate::ext::hash(self.branch.content().to_string())) Some(crate::ext::hash(self.branch.content().to_string()))
@ -282,9 +283,6 @@ impl BufferWorker {
}, },
}, },
}; };
self.local_version
.send(new_local_v)
.unwrap_or_warn("could not update local version");
tx.send(Some(tc)) tx.send(Some(tc))
.unwrap_or_warn("could not update ops channel -- is controller dead?"); .unwrap_or_warn("could not update ops channel -- is controller dead?");
} else { } else {