mirror of
https://github.com/hexedtech/codemp-nvim.git
synced 2024-11-22 15:34:53 +01:00
feat: reworked client, added static instance
This commit is contained in:
parent
96217d1a1a
commit
f8e77f0827
13 changed files with 378 additions and 305 deletions
|
@ -3,22 +3,26 @@ syntax = "proto3";
|
||||||
package codemp.cursor;
|
package codemp.cursor;
|
||||||
|
|
||||||
service Cursor {
|
service Cursor {
|
||||||
rpc Moved (CursorPosition) returns (MovedResponse);
|
rpc Moved (CursorEvent) returns (MovedResponse);
|
||||||
rpc Listen (UserIdentity) returns (stream CursorPosition);
|
rpc Listen (UserIdentity) returns (stream CursorEvent);
|
||||||
}
|
}
|
||||||
|
|
||||||
message MovedResponse {}
|
message MovedResponse {}
|
||||||
|
|
||||||
message RowColumn {
|
message RowCol {
|
||||||
int32 row = 1;
|
int32 row = 1;
|
||||||
int32 col = 2;
|
int32 col = 2;
|
||||||
}
|
}
|
||||||
|
|
||||||
message CursorPosition {
|
message CursorPosition {
|
||||||
|
string buffer = 1;
|
||||||
|
RowCol start = 2;
|
||||||
|
RowCol end = 3;
|
||||||
|
}
|
||||||
|
|
||||||
|
message CursorEvent {
|
||||||
string user = 1;
|
string user = 1;
|
||||||
string buffer = 2;
|
CursorPosition position = 2;
|
||||||
RowColumn start = 3;
|
|
||||||
RowColumn end = 4;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
message UserIdentity {
|
message UserIdentity {
|
||||||
|
|
|
@ -1,89 +1,41 @@
|
||||||
use tonic::{transport::Channel, Status};
|
use operational_transform::OperationSeq;
|
||||||
use uuid::Uuid;
|
use tokio::sync::{watch, mpsc, broadcast, Mutex};
|
||||||
|
use tonic::async_trait;
|
||||||
|
|
||||||
use crate::{
|
use crate::{Controller, CodempError};
|
||||||
ControllerWorker,
|
use crate::buffer::factory::{leading_noop, tailing_noop, OperationFactory};
|
||||||
buffer::handle::{BufferHandle, OperationControllerWorker},
|
|
||||||
proto::{buffer_client::BufferClient, BufferPayload},
|
use super::TextChange;
|
||||||
};
|
|
||||||
|
|
||||||
#[derive(Clone)]
|
|
||||||
pub struct BufferController {
|
pub struct BufferController {
|
||||||
id: String,
|
content: watch::Receiver<String>,
|
||||||
client: BufferClient<Channel>,
|
operations: mpsc::Sender<OperationSeq>,
|
||||||
|
stream: Mutex<broadcast::Receiver<OperationSeq>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl From::<BufferClient<Channel>> for BufferController {
|
#[async_trait]
|
||||||
fn from(value: BufferClient<Channel>) -> Self {
|
impl OperationFactory for BufferController {
|
||||||
BufferController { id: Uuid::new_v4().to_string(), client: value }
|
fn content(&self) -> String {
|
||||||
|
self.content.borrow().clone()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl BufferController {
|
#[async_trait]
|
||||||
pub async fn new(dest: &str) -> Result<Self, tonic::transport::Error> {
|
impl Controller<TextChange> for BufferController {
|
||||||
Ok(BufferClient::connect(dest.to_string()).await?.into())
|
type Input = OperationSeq;
|
||||||
|
|
||||||
|
async fn recv(&self) -> Result<TextChange, CodempError> {
|
||||||
|
let op = self.stream.lock().await.recv().await?;
|
||||||
|
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();
|
||||||
|
Ok(TextChange { span, content })
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn id(&self) -> &str { &self.id }
|
async fn send(&self, op: OperationSeq) -> Result<(), CodempError> {
|
||||||
|
Ok(self.operations.send(op).await?)
|
||||||
pub async fn create(&mut self, path: &str, content: Option<&str>) -> Result<(), Status> {
|
|
||||||
let req = BufferPayload {
|
|
||||||
path: path.to_string(),
|
|
||||||
content: content.map(|x| x.to_string()),
|
|
||||||
user: self.id.clone(),
|
|
||||||
};
|
|
||||||
|
|
||||||
self.client.create(req).await?;
|
|
||||||
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
// pub async fn listen(&mut self) -> Result<CursorTracker, Status> {
|
|
||||||
// let req = BufferPayload {
|
|
||||||
// path: "".into(),
|
|
||||||
// content: None,
|
|
||||||
// user: self.id.clone(),
|
|
||||||
// };
|
|
||||||
|
|
||||||
// let stream = self.client.listen(req).await?.into_inner();
|
|
||||||
|
|
||||||
// let controller = CursorTrackerWorker::new(self.id().to_string());
|
|
||||||
// let handle = controller.subscribe();
|
|
||||||
// let client = self.client.clone();
|
|
||||||
|
|
||||||
// tokio::spawn(async move {
|
|
||||||
// tracing::debug!("cursor worker started");
|
|
||||||
// controller.work(stream, client).await;
|
|
||||||
// tracing::debug!("cursor worker stopped");
|
|
||||||
// });
|
|
||||||
|
|
||||||
// Ok(handle)
|
|
||||||
// }
|
|
||||||
|
|
||||||
pub async fn attach(&mut self, path: &str) -> Result<BufferHandle, Status> {
|
|
||||||
let req = BufferPayload {
|
|
||||||
path: path.to_string(),
|
|
||||||
content: None,
|
|
||||||
user: self.id.clone(),
|
|
||||||
};
|
|
||||||
|
|
||||||
let content = self.client.sync(req.clone())
|
|
||||||
.await?
|
|
||||||
.into_inner()
|
|
||||||
.content;
|
|
||||||
|
|
||||||
let stream = self.client.attach(req).await?.into_inner();
|
|
||||||
|
|
||||||
let controller = OperationControllerWorker::new(self.id().to_string(), &content, path);
|
|
||||||
let factory = controller.subscribe();
|
|
||||||
let client = self.client.clone();
|
|
||||||
|
|
||||||
tokio::spawn(async move {
|
|
||||||
tracing::debug!("buffer worker started");
|
|
||||||
controller.work(client, stream).await;
|
|
||||||
tracing::debug!("buffer worker stopped");
|
|
||||||
});
|
|
||||||
|
|
||||||
Ok(factory)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,3 +1,11 @@
|
||||||
pub mod factory;
|
use std::ops::Range;
|
||||||
|
|
||||||
|
pub(crate) mod worker;
|
||||||
pub mod controller;
|
pub mod controller;
|
||||||
pub mod handle;
|
pub mod factory;
|
||||||
|
|
||||||
|
|
||||||
|
pub struct TextChange {
|
||||||
|
pub span: Range<usize>,
|
||||||
|
pub content: String,
|
||||||
|
}
|
||||||
|
|
|
@ -1,4 +1,4 @@
|
||||||
use std::{sync::Arc, collections::VecDeque, ops::Range};
|
use std::{sync::Arc, collections::VecDeque};
|
||||||
|
|
||||||
use operational_transform::OperationSeq;
|
use operational_transform::OperationSeq;
|
||||||
use tokio::sync::{watch, mpsc, broadcast, Mutex};
|
use tokio::sync::{watch, mpsc, broadcast, Mutex};
|
||||||
|
@ -7,48 +7,13 @@ use tonic::{async_trait, Streaming};
|
||||||
|
|
||||||
use crate::proto::{OperationRequest, RawOp};
|
use crate::proto::{OperationRequest, RawOp};
|
||||||
use crate::proto::buffer_client::BufferClient;
|
use crate::proto::buffer_client::BufferClient;
|
||||||
use crate::{ControllerWorker, Controller, CodempError};
|
use crate::ControllerWorker;
|
||||||
use crate::buffer::factory::{leading_noop, tailing_noop, OperationFactory};
|
|
||||||
|
|
||||||
pub struct TextChange {
|
use super::TextChange;
|
||||||
pub span: Range<usize>,
|
use super::controller::BufferController;
|
||||||
pub content: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct BufferHandle {
|
|
||||||
content: watch::Receiver<String>,
|
|
||||||
operations: mpsc::Sender<OperationSeq>,
|
|
||||||
stream: Mutex<broadcast::Receiver<OperationSeq>>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
pub(crate) struct BufferControllerWorker {
|
||||||
impl OperationFactory for BufferHandle {
|
|
||||||
fn content(&self) -> String {
|
|
||||||
self.content.borrow().clone()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Controller<TextChange> for BufferHandle {
|
|
||||||
type Input = OperationSeq;
|
|
||||||
|
|
||||||
async fn recv(&self) -> Result<TextChange, CodempError> {
|
|
||||||
let op = self.stream.lock().await.recv().await?;
|
|
||||||
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();
|
|
||||||
Ok(TextChange { span, content })
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn send(&self, op: OperationSeq) -> Result<(), CodempError> {
|
|
||||||
Ok(self.operations.send(op).await?)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub(crate) struct OperationControllerWorker {
|
|
||||||
uid: String,
|
uid: String,
|
||||||
pub(crate) content: watch::Sender<String>,
|
pub(crate) content: watch::Sender<String>,
|
||||||
pub(crate) operations: mpsc::Receiver<OperationSeq>,
|
pub(crate) operations: mpsc::Receiver<OperationSeq>,
|
||||||
|
@ -60,12 +25,12 @@ pub(crate) struct OperationControllerWorker {
|
||||||
path: String,
|
path: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl OperationControllerWorker {
|
impl BufferControllerWorker {
|
||||||
pub fn new(uid: String, buffer: &str, path: &str) -> Self {
|
pub fn new(uid: String, buffer: &str, path: &str) -> Self {
|
||||||
let (txt_tx, txt_rx) = watch::channel(buffer.to_string());
|
let (txt_tx, txt_rx) = watch::channel(buffer.to_string());
|
||||||
let (op_tx, op_rx) = mpsc::channel(64);
|
let (op_tx, op_rx) = mpsc::channel(64);
|
||||||
let (s_tx, _s_rx) = broadcast::channel(64);
|
let (s_tx, _s_rx) = broadcast::channel(64);
|
||||||
OperationControllerWorker {
|
BufferControllerWorker {
|
||||||
uid,
|
uid,
|
||||||
content: txt_tx,
|
content: txt_tx,
|
||||||
operations: op_rx,
|
operations: op_rx,
|
||||||
|
@ -80,13 +45,13 @@ impl OperationControllerWorker {
|
||||||
}
|
}
|
||||||
|
|
||||||
#[async_trait]
|
#[async_trait]
|
||||||
impl ControllerWorker<TextChange> for OperationControllerWorker {
|
impl ControllerWorker<TextChange> for BufferControllerWorker {
|
||||||
type Controller = BufferHandle;
|
type Controller = BufferController;
|
||||||
type Tx = BufferClient<Channel>;
|
type Tx = BufferClient<Channel>;
|
||||||
type Rx = Streaming<RawOp>;
|
type Rx = Streaming<RawOp>;
|
||||||
|
|
||||||
fn subscribe(&self) -> BufferHandle {
|
fn subscribe(&self) -> BufferController {
|
||||||
BufferHandle {
|
BufferController {
|
||||||
content: self.receiver.clone(),
|
content: self.receiver.clone(),
|
||||||
operations: self.sender.clone(),
|
operations: self.sender.clone(),
|
||||||
stream: Mutex::new(self.stream.subscribe()),
|
stream: Mutex::new(self.stream.subscribe()),
|
117
src/client.rs
Normal file
117
src/client.rs
Normal file
|
@ -0,0 +1,117 @@
|
||||||
|
use std::{sync::Arc, collections::BTreeMap};
|
||||||
|
|
||||||
|
use tonic::transport::Channel;
|
||||||
|
|
||||||
|
use crate::{
|
||||||
|
cursor::{worker::CursorControllerWorker, controller::CursorController},
|
||||||
|
proto::{
|
||||||
|
buffer_client::BufferClient, cursor_client::CursorClient, UserIdentity, BufferPayload,
|
||||||
|
},
|
||||||
|
CodempError, ControllerWorker, buffer::{controller::BufferController, worker::BufferControllerWorker},
|
||||||
|
};
|
||||||
|
|
||||||
|
|
||||||
|
pub struct CodempClient {
|
||||||
|
id: String,
|
||||||
|
client: ServiceClients,
|
||||||
|
workspace: Option<Workspace>,
|
||||||
|
}
|
||||||
|
|
||||||
|
struct ServiceClients {
|
||||||
|
buffer: BufferClient<Channel>,
|
||||||
|
cursor: CursorClient<Channel>,
|
||||||
|
}
|
||||||
|
|
||||||
|
struct Workspace {
|
||||||
|
cursor: Arc<CursorController>,
|
||||||
|
buffers: BTreeMap<String, Arc<BufferController>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
impl CodempClient {
|
||||||
|
pub async fn new(dst: &str) -> Result<Self, tonic::transport::Error> {
|
||||||
|
let buffer = BufferClient::connect(dst.to_string()).await?;
|
||||||
|
let cursor = CursorClient::connect(dst.to_string()).await?;
|
||||||
|
let id = uuid::Uuid::new_v4().to_string();
|
||||||
|
|
||||||
|
Ok(CodempClient { id, client: ServiceClients { buffer, cursor}, workspace: None })
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn get_cursor(&self) -> Option<Arc<CursorController>> {
|
||||||
|
Some(self.workspace?.cursor.clone())
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn get_buffer(&self, path: &str) -> Option<Arc<BufferController>> {
|
||||||
|
self.workspace?.buffers.get(path).cloned()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn join(&mut self, _session: &str) -> Result<Arc<CursorController>, CodempError> {
|
||||||
|
// 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
|
||||||
|
let stream = self.client.cursor.listen(UserIdentity { id: "".into() }).await?.into_inner();
|
||||||
|
|
||||||
|
let controller = CursorControllerWorker::new(self.id.clone());
|
||||||
|
let client = self.client.cursor.clone();
|
||||||
|
|
||||||
|
let handle = Arc::new(controller.subscribe());
|
||||||
|
|
||||||
|
tokio::spawn(async move {
|
||||||
|
tracing::debug!("cursor worker started");
|
||||||
|
controller.work(client, stream).await;
|
||||||
|
tracing::debug!("cursor worker stopped");
|
||||||
|
});
|
||||||
|
|
||||||
|
self.workspace = Some(
|
||||||
|
Workspace {
|
||||||
|
cursor: handle.clone(),
|
||||||
|
buffers: BTreeMap::new()
|
||||||
|
}
|
||||||
|
);
|
||||||
|
|
||||||
|
Ok(handle)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn create(&mut self, path: &str, content: Option<&str>) -> Result<(), CodempError> {
|
||||||
|
if let Some(workspace) = &self.workspace {
|
||||||
|
self.client.buffer
|
||||||
|
.create(BufferPayload {
|
||||||
|
user: self.id.clone(),
|
||||||
|
path: path.to_string(),
|
||||||
|
content: content.map(|x| x.to_string()),
|
||||||
|
}).await?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
} else {
|
||||||
|
Err(CodempError::InvalidState { msg: "join a workspace first".into() })
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn attach(&mut self, path: &str, content: Option<&str>) -> Result<Arc<BufferController>, CodempError> {
|
||||||
|
if let Some(workspace) = &mut self.workspace {
|
||||||
|
let mut client = self.client.buffer.clone();
|
||||||
|
let req = BufferPayload {
|
||||||
|
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 controller = BufferControllerWorker::new(self.id.clone(), &content, path);
|
||||||
|
let handler = Arc::new(controller.subscribe());
|
||||||
|
|
||||||
|
let _path = path.to_string();
|
||||||
|
tokio::spawn(async move {
|
||||||
|
tracing::debug!("buffer[{}] worker started", _path);
|
||||||
|
controller.work(client, stream).await;
|
||||||
|
tracing::debug!("buffer[{}] worker stopped", _path);
|
||||||
|
});
|
||||||
|
|
||||||
|
workspace.buffers.insert(path.to_string(), handler.clone());
|
||||||
|
|
||||||
|
Ok(handler)
|
||||||
|
} else {
|
||||||
|
Err(CodempError::InvalidState { msg: "join a workspace first".into() })
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
51
src/cursor/controller.rs
Normal file
51
src/cursor/controller.rs
Normal file
|
@ -0,0 +1,51 @@
|
||||||
|
use tokio::sync::{mpsc, broadcast::{self, error::RecvError}, Mutex};
|
||||||
|
use tonic::async_trait;
|
||||||
|
|
||||||
|
use crate::{proto::{CursorPosition, CursorEvent}, CodempError, Controller};
|
||||||
|
|
||||||
|
pub struct CursorController {
|
||||||
|
uid: String,
|
||||||
|
op: mpsc::Sender<CursorEvent>,
|
||||||
|
stream: Mutex<broadcast::Receiver<CursorEvent>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl Controller<CursorEvent> for CursorController {
|
||||||
|
type Input = CursorPosition;
|
||||||
|
|
||||||
|
async fn send(&self, cursor: CursorPosition) -> Result<(), CodempError> {
|
||||||
|
Ok(self.op.send(CursorEvent {
|
||||||
|
user: self.uid.clone(),
|
||||||
|
position: Some(cursor),
|
||||||
|
}).await?)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TODO is this cancelable? so it can be used in tokio::select!
|
||||||
|
// TODO is the result type overkill? should be an option?
|
||||||
|
async fn recv(&self) -> Result<CursorEvent, CodempError> {
|
||||||
|
let mut stream = self.stream.lock().await;
|
||||||
|
match stream.recv().await {
|
||||||
|
Ok(x) => Ok(x),
|
||||||
|
Err(RecvError::Closed) => Err(CodempError::Channel { send: false }),
|
||||||
|
Err(RecvError::Lagged(n)) => {
|
||||||
|
tracing::error!("cursor channel lagged behind, skipping {} events", n);
|
||||||
|
Ok(stream.recv().await.expect("could not receive after lagging"))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// fn try_poll(&self) -> Option<Option<CursorPosition>> {
|
||||||
|
// match self.stream.try_lock() {
|
||||||
|
// Err(_) => None,
|
||||||
|
// Ok(mut x) => match x.try_recv() {
|
||||||
|
// Ok(x) => Some(Some(x)),
|
||||||
|
// Err(TryRecvError::Empty) => None,
|
||||||
|
// Err(TryRecvError::Closed) => Some(None),
|
||||||
|
// Err(TryRecvError::Lagged(n)) => {
|
||||||
|
// tracing::error!("cursor channel lagged behind, skipping {} events", n);
|
||||||
|
// Some(Some(x.try_recv().expect("could not receive after lagging")))
|
||||||
|
// }
|
||||||
|
// }
|
||||||
|
// }
|
||||||
|
// }
|
||||||
|
}
|
|
@ -1,25 +1,26 @@
|
||||||
pub mod tracker;
|
pub(crate) mod worker;
|
||||||
|
pub mod controller;
|
||||||
|
|
||||||
use crate::proto::{RowColumn, CursorPosition};
|
use crate::proto::{RowCol, CursorPosition};
|
||||||
|
|
||||||
impl From::<RowColumn> for (i32, i32) {
|
impl From::<RowCol> for (i32, i32) {
|
||||||
fn from(pos: RowColumn) -> (i32, i32) {
|
fn from(pos: RowCol) -> (i32, i32) {
|
||||||
(pos.row, pos.col)
|
(pos.row, pos.col)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl From::<(i32, i32)> for RowColumn {
|
impl From::<(i32, i32)> for RowCol {
|
||||||
fn from((row, col): (i32, i32)) -> Self {
|
fn from((row, col): (i32, i32)) -> Self {
|
||||||
RowColumn { row, col }
|
RowCol { row, col }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CursorPosition {
|
impl CursorPosition {
|
||||||
pub fn start(&self) -> RowColumn {
|
pub fn start(&self) -> RowCol {
|
||||||
self.start.clone().unwrap_or((0, 0).into())
|
self.start.clone().unwrap_or((0, 0).into())
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn end(&self) -> RowColumn {
|
pub fn end(&self) -> RowCol {
|
||||||
self.end.clone().unwrap_or((0, 0).into())
|
self.end.clone().unwrap_or((0, 0).into())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,101 +0,0 @@
|
||||||
use std::sync::Arc;
|
|
||||||
|
|
||||||
use tokio::sync::{mpsc, broadcast::{self, error::RecvError}, Mutex};
|
|
||||||
use tonic::{Streaming, transport::Channel, async_trait};
|
|
||||||
|
|
||||||
use crate::{proto::{CursorPosition, cursor_client::CursorClient, RowColumn}, errors::IgnorableError, CodempError, Controller, ControllerWorker};
|
|
||||||
|
|
||||||
pub struct CursorTracker {
|
|
||||||
uid: String,
|
|
||||||
op: mpsc::Sender<CursorPosition>,
|
|
||||||
stream: Mutex<broadcast::Receiver<CursorPosition>>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl Controller<CursorPosition> for CursorTracker {
|
|
||||||
type Input = (String, RowColumn, RowColumn);
|
|
||||||
|
|
||||||
async fn send(&self, (buffer, start, end): Self::Input) -> Result<(), CodempError> {
|
|
||||||
Ok(self.op.send(CursorPosition {
|
|
||||||
user: self.uid.clone(),
|
|
||||||
start: Some(start),
|
|
||||||
end: Some(end),
|
|
||||||
buffer,
|
|
||||||
}).await?)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TODO is this cancelable? so it can be used in tokio::select!
|
|
||||||
// TODO is the result type overkill? should be an option?
|
|
||||||
async fn recv(&self) -> Result<CursorPosition, CodempError> {
|
|
||||||
let mut stream = self.stream.lock().await;
|
|
||||||
match stream.recv().await {
|
|
||||||
Ok(x) => Ok(x),
|
|
||||||
Err(RecvError::Closed) => Err(CodempError::Channel { send: false }),
|
|
||||||
Err(RecvError::Lagged(n)) => {
|
|
||||||
tracing::error!("cursor channel lagged behind, skipping {} events", n);
|
|
||||||
Ok(stream.recv().await.expect("could not receive after lagging"))
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// fn try_poll(&self) -> Option<Option<CursorPosition>> {
|
|
||||||
// match self.stream.try_lock() {
|
|
||||||
// Err(_) => None,
|
|
||||||
// Ok(mut x) => match x.try_recv() {
|
|
||||||
// Ok(x) => Some(Some(x)),
|
|
||||||
// Err(TryRecvError::Empty) => None,
|
|
||||||
// Err(TryRecvError::Closed) => Some(None),
|
|
||||||
// Err(TryRecvError::Lagged(n)) => {
|
|
||||||
// tracing::error!("cursor channel lagged behind, skipping {} events", n);
|
|
||||||
// Some(Some(x.try_recv().expect("could not receive after lagging")))
|
|
||||||
// }
|
|
||||||
// }
|
|
||||||
// }
|
|
||||||
// }
|
|
||||||
}
|
|
||||||
|
|
||||||
pub(crate) struct CursorTrackerWorker {
|
|
||||||
uid: String,
|
|
||||||
producer: mpsc::Sender<CursorPosition>,
|
|
||||||
op: mpsc::Receiver<CursorPosition>,
|
|
||||||
channel: Arc<broadcast::Sender<CursorPosition>>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl CursorTrackerWorker {
|
|
||||||
pub(crate) fn new(uid: String) -> Self {
|
|
||||||
let (op_tx, op_rx) = mpsc::channel(64);
|
|
||||||
let (cur_tx, _cur_rx) = broadcast::channel(64);
|
|
||||||
Self {
|
|
||||||
uid,
|
|
||||||
producer: op_tx,
|
|
||||||
op: op_rx,
|
|
||||||
channel: Arc::new(cur_tx),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl ControllerWorker<CursorPosition> for CursorTrackerWorker {
|
|
||||||
type Controller = CursorTracker;
|
|
||||||
type Tx = CursorClient<Channel>;
|
|
||||||
type Rx = Streaming<CursorPosition>;
|
|
||||||
|
|
||||||
fn subscribe(&self) -> CursorTracker {
|
|
||||||
CursorTracker {
|
|
||||||
uid: self.uid.clone(),
|
|
||||||
op: self.producer.clone(),
|
|
||||||
stream: Mutex::new(self.channel.subscribe()),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn work(mut self, mut tx: Self::Tx, mut rx: Self::Rx) {
|
|
||||||
loop {
|
|
||||||
tokio::select!{
|
|
||||||
Ok(Some(cur)) = rx.message() => self.channel.send(cur).unwrap_or_warn("could not broadcast event"),
|
|
||||||
Some(op) = self.op.recv() => { tx.moved(op).await.unwrap_or_warn("could not update cursor"); },
|
|
||||||
else => break,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
54
src/cursor/worker.rs
Normal file
54
src/cursor/worker.rs
Normal file
|
@ -0,0 +1,54 @@
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
|
use tokio::sync::{mpsc, broadcast::{self}, Mutex};
|
||||||
|
use tonic::{Streaming, transport::Channel, async_trait};
|
||||||
|
|
||||||
|
use crate::{proto::{cursor_client::CursorClient, CursorEvent}, errors::IgnorableError, ControllerWorker};
|
||||||
|
|
||||||
|
use super::controller::CursorController;
|
||||||
|
|
||||||
|
pub(crate) struct CursorControllerWorker {
|
||||||
|
uid: String,
|
||||||
|
producer: mpsc::Sender<CursorEvent>,
|
||||||
|
op: mpsc::Receiver<CursorEvent>,
|
||||||
|
channel: Arc<broadcast::Sender<CursorEvent>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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);
|
||||||
|
Self {
|
||||||
|
uid,
|
||||||
|
producer: op_tx,
|
||||||
|
op: op_rx,
|
||||||
|
channel: Arc::new(cur_tx),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl ControllerWorker<CursorEvent> for CursorControllerWorker {
|
||||||
|
type Controller = CursorController;
|
||||||
|
type Tx = CursorClient<Channel>;
|
||||||
|
type Rx = Streaming<CursorEvent>;
|
||||||
|
|
||||||
|
fn subscribe(&self) -> CursorController {
|
||||||
|
CursorController {
|
||||||
|
uid: self.uid.clone(),
|
||||||
|
op: self.producer.clone(),
|
||||||
|
stream: Mutex::new(self.channel.subscribe()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn work(mut self, mut tx: Self::Tx, mut rx: Self::Rx) {
|
||||||
|
loop {
|
||||||
|
tokio::select!{
|
||||||
|
Ok(Some(cur)) = rx.message() => self.channel.send(cur).unwrap_or_warn("could not broadcast event"),
|
||||||
|
Some(op) = self.op.recv() => { tx.moved(op).await.unwrap_or_warn("could not update cursor"); },
|
||||||
|
else => break,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
@ -28,6 +28,9 @@ pub enum CodempError {
|
||||||
Channel {
|
Channel {
|
||||||
send: bool
|
send: bool
|
||||||
},
|
},
|
||||||
|
InvalidState {
|
||||||
|
msg: String,
|
||||||
|
},
|
||||||
|
|
||||||
// TODO filler error, remove later
|
// TODO filler error, remove later
|
||||||
Filler {
|
Filler {
|
||||||
|
|
81
src/instance.rs
Normal file
81
src/instance.rs
Normal file
|
@ -0,0 +1,81 @@
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
|
use tokio::sync::Mutex;
|
||||||
|
|
||||||
|
use crate::{
|
||||||
|
buffer::controller::BufferController,
|
||||||
|
errors::CodempError, client::CodempClient, cursor::controller::CursorController,
|
||||||
|
};
|
||||||
|
|
||||||
|
|
||||||
|
use tokio::runtime::Runtime;
|
||||||
|
|
||||||
|
const CODEMP_DEFAULT_HOST : &str = "http://alemi.dev:50051";
|
||||||
|
|
||||||
|
lazy_static::lazy_static! {
|
||||||
|
static ref RUNTIME : Runtime = Runtime::new().expect("could not create tokio runtime");
|
||||||
|
static ref INSTANCE : Instance = Instance::default();
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct Instance {
|
||||||
|
client: Mutex<Option<CodempClient>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for Instance {
|
||||||
|
fn default() -> Self {
|
||||||
|
Instance { client: Mutex::new(None) }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TODO these methods repeat a lot of code but Mutex makes it hard to simplify
|
||||||
|
|
||||||
|
impl Instance {
|
||||||
|
pub async fn connect(&self, addr: &str) -> Result<(), CodempError> {
|
||||||
|
*self.client.lock().await = Some(CodempClient::new(addr).await?);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn join(&self, session: &str) -> Result<(), CodempError> {
|
||||||
|
self.client
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.as_mut()
|
||||||
|
.ok_or(CodempError::InvalidState { msg: "connect first".into() })?
|
||||||
|
.join(session)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn create(&self, path: &str, content: Option<&str>) -> Result<(), CodempError> {
|
||||||
|
self.client
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.as_mut()
|
||||||
|
.ok_or(CodempError::InvalidState { msg: "connect first".into() })?
|
||||||
|
.create(path, content)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn get_cursor(&self) -> Result<Arc<CursorController>, CodempError> {
|
||||||
|
self.client
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.as_mut()
|
||||||
|
.ok_or(CodempError::InvalidState { msg: "connect first".into() })?
|
||||||
|
.get_cursor()
|
||||||
|
.ok_or(CodempError::InvalidState { msg: "join a workspace first".into() })
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn get_buffer(&self, path: &str) -> Result<Arc<BufferController>, CodempError> {
|
||||||
|
self.client
|
||||||
|
.lock()
|
||||||
|
.await
|
||||||
|
.as_mut()
|
||||||
|
.ok_or(CodempError::InvalidState { msg: "connect first".into() })?
|
||||||
|
.get_buffer(path)
|
||||||
|
.ok_or(CodempError::InvalidState { msg: "join a workspace or create requested buffer first".into() })
|
||||||
|
}
|
||||||
|
}
|
|
@ -2,9 +2,11 @@ pub mod cursor;
|
||||||
pub mod errors;
|
pub mod errors;
|
||||||
pub mod buffer;
|
pub mod buffer;
|
||||||
|
|
||||||
pub mod state;
|
|
||||||
pub mod client;
|
pub mod client;
|
||||||
|
|
||||||
|
#[cfg(feature = "static")]
|
||||||
|
pub mod instance;
|
||||||
|
|
||||||
pub use tonic;
|
pub use tonic;
|
||||||
pub use tokio;
|
pub use tokio;
|
||||||
pub use operational_transform as ot;
|
pub use operational_transform as ot;
|
||||||
|
|
64
src/state.rs
64
src/state.rs
|
@ -1,64 +0,0 @@
|
||||||
use std::{collections::BTreeMap, sync::Arc};
|
|
||||||
|
|
||||||
use tokio::sync::RwLock;
|
|
||||||
|
|
||||||
use crate::{
|
|
||||||
buffer::{controller::BufferController, handle::BufferHandle},
|
|
||||||
errors::CodempError,
|
|
||||||
};
|
|
||||||
|
|
||||||
|
|
||||||
#[cfg(feature = "static")]
|
|
||||||
pub mod instance {
|
|
||||||
use tokio::runtime::Runtime;
|
|
||||||
use super::Workspace;
|
|
||||||
|
|
||||||
const CODEMP_DEFAULT_HOST : &str = "http://fantabos.co:50051";
|
|
||||||
|
|
||||||
lazy_static::lazy_static! {
|
|
||||||
static ref RUNTIME : Runtime = Runtime::new().expect("could not create tokio runtime");
|
|
||||||
static ref WORKSPACE : Workspace = RUNTIME.block_on(
|
|
||||||
Workspace::new(&std::env::var("CODEMP_HOST").unwrap_or(CODEMP_DEFAULT_HOST.into()))
|
|
||||||
).expect("could not create codemp workspace");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub struct Workspace {
|
|
||||||
buffers: RwLock<BTreeMap<Box<str>, Arc<BufferHandle>>>,
|
|
||||||
// cursor: Arc<CursorTracker>,
|
|
||||||
client: BufferController,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl Workspace {
|
|
||||||
pub async fn new(dest: &str) -> Result<Self, CodempError> {
|
|
||||||
let client = BufferController::new(dest).await?;
|
|
||||||
// let cursor = Arc::new(client.listen().await?);
|
|
||||||
Ok(
|
|
||||||
Workspace {
|
|
||||||
buffers: RwLock::new(BTreeMap::new()),
|
|
||||||
// cursor,
|
|
||||||
client,
|
|
||||||
}
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Cursor
|
|
||||||
// pub async fn cursor(&self) -> Arc<CursorTracker> {
|
|
||||||
// self.cursor.clone()
|
|
||||||
// }
|
|
||||||
|
|
||||||
// Buffer
|
|
||||||
pub async fn buffer(&self, path: &str) -> Option<Arc<BufferHandle>> {
|
|
||||||
self.buffers.read().await.get(path).cloned()
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn create(&self, path: &str, content: Option<&str>) -> Result<(), CodempError> {
|
|
||||||
Ok(self.client.clone().create(path, content).await?)
|
|
||||||
}
|
|
||||||
|
|
||||||
pub async fn attach(&self, path: &str) -> Result<(), CodempError> {
|
|
||||||
let controller = self.client.clone().attach(path).await?;
|
|
||||||
self.buffers.write().await.insert(path.into(), Arc::new(controller));
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
}
|
|
Loading…
Reference in a new issue