mirror of
https://github.com/hexedtech/codemp.git
synced 2024-11-22 15:24:48 +01:00
fix: try_recv() channel error : delta may be None
basically if there was no change to report, the oneshot would not get updated which is bad. so we put back the version we got and send a None (the channel now has nullable TextChange). so basically its always a try_recv, but its fine since recv is implemented with try_recv + poll anyway
This commit is contained in:
parent
fdcfc611b1
commit
0154e5a032
2 changed files with 11 additions and 4 deletions
|
@ -15,6 +15,8 @@ use crate::api::TextChange;
|
||||||
|
|
||||||
use crate::ext::InternallyMutable;
|
use crate::ext::InternallyMutable;
|
||||||
|
|
||||||
|
use super::worker::DeltaRequest;
|
||||||
|
|
||||||
/// the buffer controller implementation
|
/// the buffer controller implementation
|
||||||
///
|
///
|
||||||
/// for each controller a worker exists, managing outgoing and inbound
|
/// for each controller a worker exists, managing outgoing and inbound
|
||||||
|
@ -52,7 +54,7 @@ pub(crate) struct BufferControllerInner {
|
||||||
pub(crate) poller: mpsc::UnboundedSender<oneshot::Sender<()>>,
|
pub(crate) poller: mpsc::UnboundedSender<oneshot::Sender<()>>,
|
||||||
pub(crate) stopper: mpsc::UnboundedSender<()>, // just exist
|
pub(crate) stopper: mpsc::UnboundedSender<()>, // just exist
|
||||||
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<(LocalVersion, TextChange)>)>,
|
pub(crate) delta_request: mpsc::Sender<DeltaRequest>,
|
||||||
pub(crate) callback: watch::Sender<Option<ControllerCallback<BufferController>>>,
|
pub(crate) callback: watch::Sender<Option<ControllerCallback<BufferController>>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -85,7 +87,7 @@ impl Controller<TextChange> for BufferController {
|
||||||
self.0.delta_request.send((last_update, tx)).await?;
|
self.0.delta_request.send((last_update, tx)).await?;
|
||||||
let (v, change) = rx.await?;
|
let (v, change) = rx.await?;
|
||||||
self.0.last_update.set(v);
|
self.0.last_update.set(v);
|
||||||
Ok(Some(change))
|
Ok(change)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// enqueue a text change for processing
|
/// enqueue a text change for processing
|
||||||
|
|
|
@ -14,6 +14,9 @@ use codemp_proto::buffer::{BufferEvent, Operation};
|
||||||
|
|
||||||
use super::controller::{BufferController, BufferControllerInner};
|
use super::controller::{BufferController, BufferControllerInner};
|
||||||
|
|
||||||
|
pub(crate) type DeltaOp = (LocalVersion, Option<TextChange>);
|
||||||
|
pub(crate) type DeltaRequest = (LocalVersion, oneshot::Sender<DeltaOp>);
|
||||||
|
|
||||||
pub(crate) struct BufferWorker {
|
pub(crate) struct BufferWorker {
|
||||||
user_id: Uuid,
|
user_id: Uuid,
|
||||||
latest_version: watch::Sender<diamond_types::LocalVersion>,
|
latest_version: watch::Sender<diamond_types::LocalVersion>,
|
||||||
|
@ -21,7 +24,7 @@ pub(crate) 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<(LocalVersion, TextChange)>)>,
|
delta_req: mpsc::Receiver<DeltaRequest>,
|
||||||
stop: mpsc::UnboundedReceiver<()>,
|
stop: mpsc::UnboundedReceiver<()>,
|
||||||
controller: BufferController,
|
controller: BufferController,
|
||||||
callback: watch::Receiver<Option<ControllerCallback<BufferController>>>,
|
callback: watch::Receiver<Option<ControllerCallback<BufferController>>>,
|
||||||
|
@ -181,7 +184,9 @@ impl ControllerWorker<TextChange> for BufferWorker {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
tx.send((new_local_v, tc)).unwrap_or_warn("could not update ops channel -- is controller dead?");
|
tx.send((new_local_v, Some(tc))).unwrap_or_warn("could not update ops channel -- is controller dead?");
|
||||||
|
} else {
|
||||||
|
tx.send((last_ver, None)).unwrap_or_warn("could not update ops channel -- is controller dead?");
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
|
|
Loading…
Reference in a new issue