2023-04-11 14:01:11 +02:00
|
|
|
/// TODO better name for this file
|
|
|
|
|
2023-04-11 22:35:37 +02:00
|
|
|
use std::sync::Arc;
|
2023-04-11 14:01:11 +02:00
|
|
|
use tracing::{error, warn};
|
|
|
|
use uuid::Uuid;
|
|
|
|
|
|
|
|
use crate::{
|
2023-04-11 22:35:37 +02:00
|
|
|
opfactory::AsyncFactory,
|
2023-04-11 14:01:11 +02:00
|
|
|
proto::{buffer_client::BufferClient, BufferPayload, OperationRequest, RawOp},
|
|
|
|
tonic::{transport::Channel, Status, Streaming},
|
|
|
|
};
|
|
|
|
|
|
|
|
impl From::<BufferClient<Channel>> for CodempClient {
|
|
|
|
fn from(x: BufferClient<Channel>) -> CodempClient {
|
|
|
|
CodempClient {
|
|
|
|
id: Uuid::new_v4(),
|
|
|
|
client:x,
|
2023-04-11 22:35:37 +02:00
|
|
|
factory: Arc::new(AsyncFactory::new(None)),
|
2023-04-11 14:01:11 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[derive(Clone)]
|
|
|
|
pub struct CodempClient {
|
|
|
|
id: Uuid,
|
|
|
|
client: BufferClient<Channel>,
|
2023-04-11 22:35:37 +02:00
|
|
|
factory: Arc<AsyncFactory>,
|
2023-04-11 14:01:11 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
impl CodempClient {
|
|
|
|
pub async fn create(&mut self, path: String, content: Option<String>) -> Result<bool, Status> {
|
|
|
|
Ok(
|
|
|
|
self.client.create(
|
|
|
|
BufferPayload {
|
|
|
|
path,
|
|
|
|
content,
|
|
|
|
user: self.id.to_string(),
|
|
|
|
}
|
|
|
|
)
|
|
|
|
.await?
|
|
|
|
.into_inner()
|
|
|
|
.accepted
|
|
|
|
)
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn insert(&mut self, path: String, txt: String, pos: u64) -> Result<bool, Status> {
|
2023-04-11 22:35:37 +02:00
|
|
|
match self.factory.insert(txt, pos).await {
|
2023-04-11 14:01:11 +02:00
|
|
|
Ok(op) => {
|
|
|
|
Ok(
|
|
|
|
self.client.edit(
|
|
|
|
OperationRequest {
|
|
|
|
path,
|
|
|
|
hash: "".into(),
|
|
|
|
opseq: serde_json::to_string(&op).unwrap(),
|
|
|
|
user: self.id.to_string(),
|
|
|
|
}
|
|
|
|
)
|
|
|
|
.await?
|
|
|
|
.into_inner()
|
|
|
|
.accepted
|
|
|
|
)
|
|
|
|
},
|
|
|
|
Err(e) => Err(Status::internal(format!("invalid operation: {}", e))),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-04-11 14:24:40 +02:00
|
|
|
pub async fn delete(&mut self, path: String, pos: u64, count: u64) -> Result<bool, Status> {
|
2023-04-11 22:35:37 +02:00
|
|
|
match self.factory.delete(pos, count).await {
|
2023-04-11 14:24:40 +02:00
|
|
|
Ok(op) => {
|
|
|
|
Ok(
|
|
|
|
self.client.edit(
|
|
|
|
OperationRequest {
|
|
|
|
path,
|
|
|
|
hash: "".into(),
|
|
|
|
opseq: serde_json::to_string(&op).unwrap(),
|
|
|
|
user: self.id.to_string(),
|
|
|
|
}
|
|
|
|
)
|
|
|
|
.await?
|
|
|
|
.into_inner()
|
|
|
|
.accepted
|
|
|
|
)
|
|
|
|
},
|
|
|
|
Err(e) => Err(Status::internal(format!("invalid operation: {}", e))),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-04-11 14:01:11 +02:00
|
|
|
pub async fn attach<F : Fn(String) -> () + Send + 'static>(&mut self, path: String, callback: F) -> Result<(), Status> {
|
|
|
|
let stream = self.client.attach(
|
|
|
|
BufferPayload {
|
|
|
|
path,
|
|
|
|
content: None,
|
|
|
|
user: self.id.to_string(),
|
|
|
|
}
|
|
|
|
)
|
|
|
|
.await?
|
|
|
|
.into_inner();
|
|
|
|
|
|
|
|
let factory = self.factory.clone();
|
|
|
|
tokio::spawn(async move { Self::worker(stream, factory, callback).await } );
|
|
|
|
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2023-04-11 22:35:37 +02:00
|
|
|
pub async fn sync(&mut self, path: String) -> Result<String, Status> {
|
|
|
|
let res = self.client.sync(
|
|
|
|
BufferPayload {
|
|
|
|
path, content: None, user: self.id.to_string(),
|
|
|
|
}
|
|
|
|
).await?;
|
|
|
|
Ok(res.into_inner().content.unwrap_or("".into()))
|
|
|
|
}
|
|
|
|
|
|
|
|
async fn worker<F : Fn(String) -> ()>(mut stream: Streaming<RawOp>, factory: Arc<AsyncFactory>, callback: F) {
|
2023-04-11 14:01:11 +02:00
|
|
|
loop {
|
|
|
|
match stream.message().await {
|
2023-04-11 22:35:37 +02:00
|
|
|
Err(e) => break error!("error receiving change: {}", e),
|
2023-04-11 14:01:11 +02:00
|
|
|
Ok(v) => match v {
|
2023-04-11 22:35:37 +02:00
|
|
|
None => break warn!("stream closed"),
|
2023-04-11 14:01:11 +02:00
|
|
|
Some(operation) => {
|
2023-04-11 17:12:22 +02:00
|
|
|
match serde_json::from_str(&operation.opseq) {
|
|
|
|
Err(e) => break error!("could not deserialize opseq: {}", e),
|
2023-04-11 22:35:37 +02:00
|
|
|
Ok(op) => match factory.process(op).await {
|
|
|
|
Err(e) => break error!("desynched: {}", e),
|
|
|
|
Ok(x) => callback(x),
|
|
|
|
},
|
2023-04-11 14:01:11 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|