2023-07-09 03:44:27 +02:00
|
|
|
use operational_transform::OperationSeq;
|
2023-08-11 15:33:40 +02:00
|
|
|
use tonic::{transport::Channel, Status, Streaming, async_trait};
|
|
|
|
use uuid::Uuid;
|
|
|
|
|
|
|
|
use crate::{
|
|
|
|
ControllerWorker,
|
|
|
|
cursor::tracker::{CursorTracker, CursorTrackerWorker},
|
|
|
|
buffer::controller::{OperationControllerHandle, OperationControllerEditor, OperationControllerWorker},
|
|
|
|
proto::{buffer_client::BufferClient, BufferPayload, RawOp, OperationRequest},
|
|
|
|
};
|
|
|
|
|
|
|
|
#[derive(Clone)]
|
|
|
|
pub struct BufferController {
|
|
|
|
id: String,
|
|
|
|
client: BufferClient<Channel>,
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
impl From::<BufferClient<Channel>> for BufferController {
|
|
|
|
fn from(value: BufferClient<Channel>) -> Self {
|
|
|
|
BufferController { id: Uuid::new_v4().to_string(), client: value }
|
|
|
|
}
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
impl BufferController {
|
|
|
|
pub async fn new(dest: &str) -> Result<Self, tonic::transport::Error> {
|
|
|
|
Ok(BufferClient::connect(dest.to_string()).await?.into())
|
|
|
|
}
|
2023-07-30 22:58:24 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
pub fn id(&self) -> &str { &self.id }
|
2023-07-30 22:58:24 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
pub async fn create(&mut self, path: &str, content: Option<&str>) -> Result<bool, Status> {
|
|
|
|
let req = BufferPayload {
|
|
|
|
path: path.to_string(),
|
|
|
|
content: content.map(|x| x.to_string()),
|
|
|
|
user: self.id.clone(),
|
|
|
|
};
|
2023-07-09 03:44:27 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
let res = self.client.create(req).await?;
|
2023-07-09 03:44:27 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
Ok(res.into_inner().accepted)
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
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)
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
2023-07-30 17:48:55 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
pub async fn attach(&mut self, path: &str) -> Result<OperationControllerHandle, Status> {
|
|
|
|
let req = BufferPayload {
|
|
|
|
path: path.to_string(),
|
|
|
|
content: None,
|
|
|
|
user: self.id.clone(),
|
|
|
|
};
|
2023-07-09 03:44:27 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
let content = self.client.sync(req.clone())
|
|
|
|
.await?
|
|
|
|
.into_inner()
|
|
|
|
.content
|
|
|
|
.unwrap_or("".into());
|
2023-07-09 03:44:27 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
let stream = self.client.attach(req).await?.into_inner();
|
2023-07-09 03:44:27 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
let controller = OperationControllerWorker::new((self.clone(), stream), &content, path);
|
|
|
|
let factory = controller.subscribe();
|
|
|
|
|
|
|
|
tokio::spawn(async move {
|
|
|
|
tracing::debug!("buffer worker started");
|
|
|
|
controller.work().await;
|
|
|
|
tracing::debug!("buffer worker stopped");
|
|
|
|
});
|
|
|
|
|
|
|
|
Ok(factory)
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
2023-08-11 15:33:40 +02:00
|
|
|
}
|
2023-07-09 03:44:27 +02:00
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
#[async_trait]
|
|
|
|
impl OperationControllerEditor for (BufferController, Streaming<RawOp>) {
|
|
|
|
async fn edit(&mut self, path: String, op: OperationSeq) -> bool {
|
|
|
|
let req = OperationRequest {
|
|
|
|
hash: "".into(),
|
|
|
|
opseq: serde_json::to_string(&op).unwrap(),
|
|
|
|
path,
|
|
|
|
user: self.0.id().to_string(),
|
|
|
|
};
|
|
|
|
match self.0.client.edit(req).await {
|
|
|
|
Ok(res) => res.into_inner().accepted,
|
|
|
|
Err(e) => {
|
|
|
|
tracing::error!("error sending edit: {}", e);
|
|
|
|
false
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-08-11 15:33:40 +02:00
|
|
|
async fn recv(&mut self) -> Option<OperationSeq> {
|
|
|
|
match self.1.message().await {
|
|
|
|
Ok(Some(op)) => Some(serde_json::from_str(&op.opseq).unwrap()),
|
|
|
|
Ok(None) => None,
|
|
|
|
Err(e) => {
|
|
|
|
tracing::error!("could not receive edit from server: {}", e);
|
|
|
|
None
|
|
|
|
}
|
2023-07-09 03:44:27 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|