mirror of
https://github.com/hexedtech/codemp-nvim.git
synced 2024-11-22 07:24:52 +01:00
chore: removed rest of the project
This commit is contained in:
parent
45a5667e5a
commit
6fa161c0d5
30 changed files with 10 additions and 1636 deletions
38
Cargo.toml
38
Cargo.toml
|
|
@ -1,34 +1,16 @@
|
|||
[workspace]
|
||||
members = ["client/nvim", "client/vscode", "server"]
|
||||
|
||||
[package]
|
||||
name = "codemp"
|
||||
version = "0.3.1"
|
||||
name = "codemp-nvim"
|
||||
version = "0.2.0"
|
||||
edition = "2021"
|
||||
|
||||
[lib]
|
||||
name = "codemp"
|
||||
|
||||
[dependencies]
|
||||
# core
|
||||
codemp = { path = "../.." }
|
||||
tracing = "0.1"
|
||||
tonic = { version = "0.9", features = ["tls", "tls-roots"] }
|
||||
prost = { version = "0.11.8", optional = true }
|
||||
md5 = "0.7.0"
|
||||
tracing-subscriber = "0.3"
|
||||
uuid = { version = "1.3.1", features = ["v4"] }
|
||||
operational-transform = { version = "0.6", features = ["serde"] }
|
||||
tokio = { version = "1.0", features = ["macros", "rt-multi-thread", "sync", "full"], optional = false }
|
||||
tokio-stream = { version = "0.1", optional = false }
|
||||
serde = { version = "1", optional = false }
|
||||
serde_json = { version = "1", optional = false }
|
||||
tracing-subscriber = { version = "0.3", optional = true }
|
||||
similar = { version = "2.2", features = ["inline"] }
|
||||
lazy_static = { version = "1.4", optional = true }
|
||||
|
||||
[build-dependencies]
|
||||
tonic-build = "0.9"
|
||||
|
||||
[features]
|
||||
default = ["proto", "static"]
|
||||
proto = ["dep:prost"]
|
||||
static = ["dep:lazy_static"]
|
||||
serde = "1"
|
||||
serde_json = "1"
|
||||
rmpv = "1"
|
||||
clap = { version = "4.2.1", features = ["derive"] }
|
||||
nvim-rs = { version = "0.5", features = ["use_tokio"] }
|
||||
async-trait = "0.1.68"
|
||||
|
|
|
|||
5
build.rs
5
build.rs
|
|
@ -1,5 +0,0 @@
|
|||
fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
tonic_build::compile_protos("proto/buffer.proto")?;
|
||||
tonic_build::compile_protos("proto/cursor.proto")?;
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -1,16 +0,0 @@
|
|||
[package]
|
||||
name = "codemp-nvim"
|
||||
version = "0.2.0"
|
||||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
codemp = { path = "../.." }
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = "0.3"
|
||||
uuid = { version = "1.3.1", features = ["v4"] }
|
||||
serde = "1"
|
||||
serde_json = "1"
|
||||
rmpv = "1"
|
||||
clap = { version = "4.2.1", features = ["derive"] }
|
||||
nvim-rs = { version = "0.5", features = ["use_tokio"] }
|
||||
async-trait = "0.1.68"
|
||||
|
|
@ -1,13 +0,0 @@
|
|||
#!/bin/sh
|
||||
|
||||
rm codemp.vsix
|
||||
mkdir -p .vsix/extension
|
||||
cp package.json .vsix/extension/package.json
|
||||
cp README.md .vsix/extension/README.md
|
||||
mkdir .vsix/extension/out
|
||||
cp -R src/*.js .vsix/extension/out
|
||||
cp -R codemp.node .vsix/extension/out/codemp.node
|
||||
cd .vsix/
|
||||
zip ../codemp.vsix -r *
|
||||
cd ..
|
||||
rm -rf .vsix/
|
||||
|
|
@ -1,22 +0,0 @@
|
|||
[package]
|
||||
name = "codemp-vscode"
|
||||
version = "0.0.1"
|
||||
description = "VSCode extension for CodeMP"
|
||||
edition = "2021"
|
||||
exclude = ["index.node"]
|
||||
|
||||
[lib]
|
||||
crate-type = ["cdylib"]
|
||||
|
||||
[dependencies]
|
||||
codemp = { path = "../.." }
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = "0.3"
|
||||
uuid = { version = "1.3.1", features = ["v4"] }
|
||||
once_cell = "1"
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
rmpv = "1"
|
||||
clap = { version = "4.2.1", features = ["derive"] }
|
||||
async-trait = "0.1.68"
|
||||
neon = { version = "0.10.1", default-features = false, features = ["channel-api", "napi-6", "promise-api"] }
|
||||
|
|
@ -1,38 +0,0 @@
|
|||
{
|
||||
"name": "codemp-vscode",
|
||||
"version": "0.0.1",
|
||||
"description": "VSCode extension for CodeMP",
|
||||
"main": "./out/extension.js",
|
||||
"engines": {
|
||||
"vscode": "^1.32.0"
|
||||
},
|
||||
"scripts": {
|
||||
"build": "cargo-cp-artifact --artifact cdylib codemp-vscode codemp.node -- cargo build --release --message-format=json-render-diagnostics",
|
||||
"install": "npm run build",
|
||||
"test": "cargo test"
|
||||
},
|
||||
"devDependencies": {
|
||||
"cargo-cp-artifact": "^0.1"
|
||||
},
|
||||
"contributes": {
|
||||
"commands": [
|
||||
{
|
||||
"command": "codemp.connect",
|
||||
"title": "Connect to CodeMP"
|
||||
},
|
||||
{
|
||||
"command": "codemp.join",
|
||||
"title": "Join remote session"
|
||||
},
|
||||
{
|
||||
"command": "codemp.share",
|
||||
"title": "Share local session"
|
||||
}
|
||||
]
|
||||
},
|
||||
"activationEvents": [
|
||||
"onCommand:codemp.connect",
|
||||
"onCommand:codemp.join",
|
||||
"onCommand:codemp.share"
|
||||
]
|
||||
}
|
||||
|
|
@ -1,146 +0,0 @@
|
|||
const vscode = require("vscode");
|
||||
const codemp = require("./codemp.node");
|
||||
|
||||
var CLIENT = null
|
||||
var CONTROLLER
|
||||
var CURSOR
|
||||
var DECORATION = null
|
||||
var OP_CACHE = new Set()
|
||||
|
||||
async function activate(context) {
|
||||
context.subscriptions.push(
|
||||
vscode.commands.registerCommand("codemp.connect", connect),
|
||||
vscode.commands.registerCommand("codemp.share", share),
|
||||
vscode.commands.registerCommand("codemp.join", join),
|
||||
)
|
||||
}
|
||||
|
||||
async function connect() {
|
||||
let host = await vscode.window.showInputBox({prompt: "server host (default to http://fantabos.co:50051)"})
|
||||
if (host === undefined) return // user cancelled with ESC
|
||||
if (host.length == 0) host = "http://fantabos.co:50051"
|
||||
CLIENT = await codemp.connect(host)
|
||||
vscode.window.showInformationMessage(`Connected to codemp @[${host}]`);
|
||||
}
|
||||
|
||||
async function share() {
|
||||
if (CLIENT === null) {
|
||||
vscode.window.showErrorMessage("No connected client");
|
||||
}
|
||||
|
||||
let path = await vscode.window.showInputBox({prompt: "buffer uri (default to file path)"})
|
||||
if (path === undefined) return // user cancelled with ESC
|
||||
if (path.length == 0) path = doc.uri.toString()
|
||||
|
||||
let doc = vscode.window.activeTextEditor.document;
|
||||
|
||||
try {
|
||||
if (!await CLIENT.create(path, doc.getText())) {
|
||||
vscode.window.showErrorMessage("Could not share buffer");
|
||||
}
|
||||
|
||||
await _attach(path)
|
||||
|
||||
vscode.window.showInformationMessage(`Shared document on buffer "${path}"`);
|
||||
} catch (err) {
|
||||
vscode.window.showErrorMessage("Error sharing: " + err)
|
||||
}
|
||||
}
|
||||
|
||||
async function join() {
|
||||
if (CLIENT === null) {
|
||||
vscode.window.showErrorMessage("No connected client");
|
||||
}
|
||||
|
||||
let path = await vscode.window.showInputBox({prompt: "buffer uri"})
|
||||
|
||||
try {
|
||||
let controller = await _attach(path)
|
||||
|
||||
vscode.window.showInformationMessage(`Joined buffer "${path}"`);
|
||||
|
||||
let editor = vscode.window.activeTextEditor
|
||||
|
||||
let range = new vscode.Range(
|
||||
editor.document.positionAt(0),
|
||||
editor.document.positionAt(editor.document.getText().length)
|
||||
)
|
||||
let content = controller.content()
|
||||
OP_CACHE.add((range, content))
|
||||
editor.edit(editBuilder => editBuilder.replace(range, content))
|
||||
} catch (err) {
|
||||
vscode.window.showErrorMessage("error joining " + err)
|
||||
}
|
||||
}
|
||||
|
||||
function _order_tuples(a, b) {
|
||||
if (a[0] < b[0]) return (a, b)
|
||||
if (a[0] > b[0]) return (b, a)
|
||||
if (a[1] < b[1]) return (a, b)
|
||||
return (b, a)
|
||||
}
|
||||
|
||||
async function _attach(path) {
|
||||
let editor = vscode.window.activeTextEditor
|
||||
let doc = editor.document;
|
||||
|
||||
CURSOR = await CLIENT.listen()
|
||||
CURSOR.callback((usr, path, start, end) => {
|
||||
try {
|
||||
if (DECORATION != null) {
|
||||
DECORATION.dispose()
|
||||
DECORATION = null
|
||||
}
|
||||
const range_start = new vscode.Position(start[0] - 1, start[1]);
|
||||
const range_end = new vscode.Position(end[0] - 1, end[1]);
|
||||
const decorationRange = new vscode.Range(range_start, range_end);
|
||||
DECORATION = vscode.window.createTextEditorDecorationType(
|
||||
{backgroundColor: 'red', color: 'white'}
|
||||
)
|
||||
editor.setDecorations(DECORATION, [decorationRange])
|
||||
} catch (err) {
|
||||
vscode.window.showErrorMessage("error setting cursor decoration: " + err)
|
||||
}
|
||||
})
|
||||
vscode.window.onDidChangeTextEditorSelection(async (e) => {
|
||||
let buf = e.textEditor.document.uri.toString()
|
||||
let selection = e.selections[0] // TODO there may be more than one cursor!!
|
||||
let anchor = [selection.anchor.line+1, selection.anchor.character]
|
||||
let position = [selection.active.line+1, selection.active.character+1]
|
||||
// (anchor, position) = _order_tuples(anchor, position)
|
||||
await CURSOR.send(buf, anchor, position)
|
||||
})
|
||||
|
||||
CONTROLLER = await CLIENT.attach(path)
|
||||
CONTROLLER.callback((start, end, text) => {
|
||||
try {
|
||||
let range = new vscode.Range(
|
||||
editor.document.positionAt(start),
|
||||
editor.document.positionAt(end)
|
||||
)
|
||||
OP_CACHE.add((range, text))
|
||||
editor.edit(editBuilder => editBuilder.replace(range, text))
|
||||
} catch (err) {
|
||||
vscode.window.showErrorMessage("could not set buffer: " + err)
|
||||
}
|
||||
})
|
||||
vscode.workspace.onDidChangeTextDocument(async (e) => {
|
||||
if (e.document != doc) return
|
||||
for (let change of e.contentChanges) {
|
||||
if (OP_CACHE.has((change.range, change.text))) {
|
||||
OP_CACHE.delete((change.range, change.text))
|
||||
continue
|
||||
}
|
||||
try {
|
||||
await CONTROLLER.apply(change.rangeOffset, change.text, change.rangeOffset + change.rangeLength)
|
||||
} catch (err) {
|
||||
vscode.window.showErrorMessage("failed sending change: " + err)
|
||||
}
|
||||
}
|
||||
})
|
||||
return CONTROLLER
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
activate,
|
||||
}
|
||||
|
|
@ -1,270 +0,0 @@
|
|||
use std::sync::Arc;
|
||||
|
||||
use neon::prelude::*;
|
||||
use once_cell::sync::OnceCell;
|
||||
use codemp::{
|
||||
cursor::controller::{CursorSubscriber, CursorControllerHandle},
|
||||
buffer::{
|
||||
controller::{OperationControllerHandle, OperationControllerSubscriber},
|
||||
client::CodempClient,
|
||||
factory::OperationFactory,
|
||||
},
|
||||
proto::buffer_client::BufferClient,
|
||||
};
|
||||
use codemp::tokio::{runtime::Runtime, sync::Mutex};
|
||||
|
||||
fn runtime<'a, C: Context<'a>>(cx: &mut C) -> NeonResult<&'static Runtime> {
|
||||
static RUNTIME: OnceCell<Runtime> = OnceCell::new();
|
||||
|
||||
RUNTIME.get_or_try_init(|| {
|
||||
Runtime::new()
|
||||
.or_else(|err| cx.throw_error(err.to_string()))
|
||||
})
|
||||
}
|
||||
|
||||
fn tuple<'a, C: Context<'a>>(cx: &mut C, a: i32, b: i32) -> NeonResult<Handle<'a, JsArray>> {
|
||||
let obj = cx.empty_array();
|
||||
let a_val = cx.number(a);
|
||||
obj.set(cx, 0, a_val)?;
|
||||
let b_val = cx.number(b);
|
||||
obj.set(cx, 1, b_val)?;
|
||||
Ok(obj)
|
||||
}
|
||||
|
||||
fn unpack_tuple<'a, C: Context<'a>>(cx: &mut C, arr: Handle<'a, JsArray>) -> NeonResult<(i32, i32)> {
|
||||
Ok((
|
||||
arr.get::<JsNumber, _, u32>(cx, 0)?.value(cx) as i32,
|
||||
arr.get::<JsNumber, _, u32>(cx, 1)?.value(cx) as i32,
|
||||
))
|
||||
}
|
||||
|
||||
struct ClientHandle(Arc<Mutex<CodempClient>>);
|
||||
impl Finalize for ClientHandle {}
|
||||
|
||||
fn connect(mut cx: FunctionContext) -> JsResult<JsPromise> {
|
||||
let host = cx.argument::<JsString>(0).ok().map(|x| x.value(&mut cx));
|
||||
|
||||
let (deferred, promise) = cx.promise();
|
||||
let channel = cx.channel();
|
||||
|
||||
runtime(&mut cx)?.spawn(async move {
|
||||
match BufferClient::connect(host.unwrap_or("".into())).await {
|
||||
Err(e) => deferred.settle_with(&channel, move |mut cx| cx.throw_error::<String, neon::handle::Handle<JsString>>(format!("{}", e))),
|
||||
Ok(c) => deferred.settle_with(&channel, |mut cx| {
|
||||
let obj = cx.empty_object();
|
||||
let boxed_value = cx.boxed(ClientHandle(Arc::new(Mutex::new(c.into()))));
|
||||
obj.set(&mut cx, "boxed", boxed_value)?;
|
||||
let method_create = JsFunction::new(&mut cx, create_client)?;
|
||||
obj.set(&mut cx, "create", method_create)?;
|
||||
let method_listen = JsFunction::new(&mut cx, listen_client)?;
|
||||
obj.set(&mut cx, "listen", method_listen)?;
|
||||
let method_attach = JsFunction::new(&mut cx, attach_client)?;
|
||||
obj.set(&mut cx, "attach", method_attach)?;
|
||||
Ok(obj)
|
||||
}),
|
||||
}
|
||||
});
|
||||
|
||||
Ok(promise)
|
||||
}
|
||||
|
||||
fn create_client(mut cx: FunctionContext) -> JsResult<JsPromise> {
|
||||
let path = cx.argument::<JsString>(0)?.value(&mut cx);
|
||||
let content = cx.argument::<JsString>(1).ok().map(|x| x.value(&mut cx));
|
||||
let this = cx.this();
|
||||
let boxed : Handle<JsBox<ClientHandle>> = this.get(&mut cx, "boxed")?;
|
||||
|
||||
let rc = boxed.0.clone();
|
||||
let (deferred, promise) = cx.promise();
|
||||
let channel = cx.channel();
|
||||
|
||||
runtime(&mut cx)?.spawn(async move {
|
||||
match rc.lock().await.create(&path, content.as_deref()).await {
|
||||
Ok(accepted) => deferred.settle_with(&channel, move |mut cx| Ok(cx.boolean(accepted))),
|
||||
Err(e) => deferred.settle_with(&channel, move |mut cx| cx.throw_error::<String, neon::handle::Handle<JsString>>(e.to_string())),
|
||||
}
|
||||
});
|
||||
|
||||
Ok(promise)
|
||||
}
|
||||
|
||||
fn listen_client(mut cx: FunctionContext) -> JsResult<JsPromise> {
|
||||
let this = cx.this();
|
||||
let boxed : Handle<JsBox<ClientHandle>> = this.get(&mut cx, "boxed")?;
|
||||
|
||||
let rc = boxed.0.clone();
|
||||
let (deferred, promise) = cx.promise();
|
||||
let channel = cx.channel();
|
||||
|
||||
runtime(&mut cx)?.spawn(async move {
|
||||
match rc.lock().await.listen().await {
|
||||
Ok(controller) => {
|
||||
deferred.settle_with(&channel, move |mut cx| {
|
||||
let obj = cx.empty_object();
|
||||
let boxed_value = cx.boxed(CursorEventsHandle(controller));
|
||||
obj.set(&mut cx, "boxed", boxed_value)?;
|
||||
let callback_method = JsFunction::new(&mut cx, callback_cursor)?;
|
||||
obj.set(&mut cx, "callback", callback_method)?;
|
||||
let send_method = JsFunction::new(&mut cx, send_cursor)?;
|
||||
obj.set(&mut cx, "send", send_method)?;
|
||||
Ok(obj)
|
||||
})
|
||||
},
|
||||
Err(e) => deferred.settle_with(&channel, move |mut cx| cx.throw_error::<String, neon::handle::Handle<JsString>>(e.to_string())),
|
||||
}
|
||||
});
|
||||
|
||||
Ok(promise)
|
||||
}
|
||||
|
||||
fn attach_client(mut cx: FunctionContext) -> JsResult<JsPromise> {
|
||||
let this = cx.this();
|
||||
let boxed : Handle<JsBox<ClientHandle>> = this.get(&mut cx, "boxed")?;
|
||||
let path = cx.argument::<JsString>(0)?.value(&mut cx);
|
||||
|
||||
let rc = boxed.0.clone();
|
||||
let (deferred, promise) = cx.promise();
|
||||
let channel = cx.channel();
|
||||
|
||||
runtime(&mut cx)?.spawn(async move {
|
||||
match rc.lock().await.attach(&path).await {
|
||||
Ok(controller) => {
|
||||
deferred.settle_with(&channel, move |mut cx| {
|
||||
let obj = cx.empty_object();
|
||||
let boxed_value = cx.boxed(OperationControllerJs(controller));
|
||||
obj.set(&mut cx, "boxed", boxed_value)?;
|
||||
let apply_method = JsFunction::new(&mut cx, apply_operation)?;
|
||||
obj.set(&mut cx, "apply", apply_method)?;
|
||||
let content_method = JsFunction::new(&mut cx, content_operation)?;
|
||||
obj.set(&mut cx, "content", content_method)?;
|
||||
let callback_method = JsFunction::new(&mut cx, callback_operation)?;
|
||||
obj.set(&mut cx, "callback", callback_method)?;
|
||||
Ok(obj)
|
||||
})
|
||||
},
|
||||
Err(e) => deferred.settle_with(&channel, move |mut cx| cx.throw_error::<String, neon::handle::Handle<JsString>>(e.to_string())),
|
||||
}
|
||||
});
|
||||
|
||||
Ok(promise)
|
||||
}
|
||||
|
||||
|
||||
struct OperationControllerJs(OperationControllerHandle);
|
||||
impl Finalize for OperationControllerJs {}
|
||||
|
||||
fn apply_operation(mut cx: FunctionContext) -> JsResult<JsPromise> {
|
||||
let this = cx.this();
|
||||
let boxed : Handle<JsBox<OperationControllerJs>> = this.get(&mut cx, "boxed")?;
|
||||
let skip = cx.argument::<JsNumber>(0)?.value(&mut cx).round() as usize;
|
||||
let text = cx.argument::<JsString>(1)?.value(&mut cx);
|
||||
let tail = cx.argument::<JsNumber>(2)?.value(&mut cx).round() as usize;
|
||||
|
||||
let rc = boxed.0.clone();
|
||||
let (deferred, promise) = cx.promise();
|
||||
let channel = cx.channel();
|
||||
|
||||
runtime(&mut cx)?.spawn(async move {
|
||||
if let Some(op) = rc.delta(skip, text.as_str(), tail) {
|
||||
rc.apply(op).await;
|
||||
deferred.settle_with(&channel, move |mut cx| Ok(cx.boolean(true)));
|
||||
} else {
|
||||
deferred.settle_with(&channel, move |mut cx| Ok(cx.undefined()));
|
||||
}
|
||||
});
|
||||
|
||||
Ok(promise)
|
||||
}
|
||||
|
||||
fn content_operation(mut cx: FunctionContext) -> JsResult<JsString> {
|
||||
let this = cx.this();
|
||||
let boxed : Handle<JsBox<OperationControllerJs>> = this.get(&mut cx, "boxed")?;
|
||||
Ok(cx.string(boxed.0.content()))
|
||||
}
|
||||
|
||||
fn callback_operation(mut cx: FunctionContext) -> JsResult<JsUndefined> {
|
||||
let this = cx.this();
|
||||
let boxed : Handle<JsBox<OperationControllerJs>> = this.get(&mut cx, "boxed")?;
|
||||
let callback = Arc::new(cx.argument::<JsFunction>(0)?.root(&mut cx));
|
||||
|
||||
let rc = boxed.0.clone();
|
||||
let channel = cx.channel();
|
||||
|
||||
// TODO when garbage collecting OperationController stop this worker
|
||||
runtime(&mut cx)?.spawn(async move {
|
||||
while let Some(edit) = rc.poll().await {
|
||||
let cb = callback.clone();
|
||||
channel.send(move |mut cx| {
|
||||
cb.to_inner(&mut cx)
|
||||
.call_with(&cx)
|
||||
.arg(cx.number(edit.span.start as i32))
|
||||
.arg(cx.number(edit.span.end as i32))
|
||||
.arg(cx.string(edit.content))
|
||||
.apply::<JsUndefined, _>(&mut cx)?;
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
Ok(cx.undefined())
|
||||
}
|
||||
|
||||
struct CursorEventsHandle(CursorControllerHandle);
|
||||
impl Finalize for CursorEventsHandle {}
|
||||
|
||||
fn callback_cursor(mut cx: FunctionContext) -> JsResult<JsUndefined> {
|
||||
let this = cx.this();
|
||||
let boxed : Handle<JsBox<CursorEventsHandle>> = this.get(&mut cx, "boxed")?;
|
||||
let callback = Arc::new(cx.argument::<JsFunction>(0)?.root(&mut cx));
|
||||
|
||||
let rc = boxed.0.clone();
|
||||
let channel = cx.channel();
|
||||
|
||||
// TODO when garbage collecting OperationController stop this worker
|
||||
runtime(&mut cx)?.spawn(async move {
|
||||
while let Some(op) = rc.poll().await {
|
||||
let cb = callback.clone();
|
||||
channel.send(move |mut cx| {
|
||||
cb.to_inner(&mut cx)
|
||||
.call_with(&cx)
|
||||
.arg(cx.string(&op.user))
|
||||
.arg(cx.string(&op.buffer))
|
||||
.arg(tuple(&mut cx, op.start().row, op.start().col)?)
|
||||
.arg(tuple(&mut cx, op.end().row, op.end().col)?)
|
||||
.apply::<JsUndefined, _>(&mut cx)?;
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
|
||||
});
|
||||
|
||||
Ok(cx.undefined())
|
||||
}
|
||||
|
||||
fn send_cursor(mut cx: FunctionContext) -> JsResult<JsPromise> {
|
||||
let this = cx.this();
|
||||
let boxed : Handle<JsBox<CursorEventsHandle>> = this.get(&mut cx, "boxed")?;
|
||||
let path = cx.argument::<JsString>(0)?.value(&mut cx);
|
||||
let start_obj = cx.argument::<JsArray>(1)?;
|
||||
let start = unpack_tuple(&mut cx, start_obj)?;
|
||||
let end_obj = cx.argument::<JsArray>(2)?;
|
||||
let end = unpack_tuple(&mut cx, end_obj)?;
|
||||
|
||||
let rc = boxed.0.clone();
|
||||
let (deferred, promise) = cx.promise();
|
||||
let channel = cx.channel();
|
||||
|
||||
runtime(&mut cx)?.spawn(async move {
|
||||
rc.send(&path, start.into(), end.into()).await;
|
||||
deferred.settle_with(&channel, |mut cx| Ok(cx.undefined()))
|
||||
});
|
||||
|
||||
Ok(promise)
|
||||
}
|
||||
|
||||
#[neon::main]
|
||||
fn main(mut cx: ModuleContext) -> NeonResult<()> {
|
||||
cx.export_function("connect", connect)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -1,35 +0,0 @@
|
|||
syntax = "proto3";
|
||||
|
||||
package codemp.buffer;
|
||||
|
||||
service Buffer {
|
||||
rpc Attach (BufferPayload) returns (stream RawOp);
|
||||
rpc Edit (OperationRequest) returns (BufferEditResponse);
|
||||
rpc Create (BufferPayload) returns (BufferCreateResponse);
|
||||
rpc Sync (BufferPayload) returns (BufferResponse);
|
||||
}
|
||||
|
||||
message BufferCreateResponse {}
|
||||
message BufferEditResponse {}
|
||||
|
||||
message RawOp {
|
||||
string opseq = 1;
|
||||
string user = 2;
|
||||
}
|
||||
|
||||
message OperationRequest {
|
||||
string path = 1;
|
||||
string hash = 2;
|
||||
string opseq = 3;
|
||||
string user = 4;
|
||||
}
|
||||
|
||||
message BufferPayload {
|
||||
string path = 1;
|
||||
string user = 2;
|
||||
optional string content = 3;
|
||||
}
|
||||
|
||||
message BufferResponse {
|
||||
string content = 2;
|
||||
}
|
||||
|
|
@ -1,30 +0,0 @@
|
|||
syntax = "proto3";
|
||||
|
||||
package codemp.cursor;
|
||||
|
||||
service Cursor {
|
||||
rpc Moved (CursorEvent) returns (MovedResponse);
|
||||
rpc Listen (UserIdentity) returns (stream CursorEvent);
|
||||
}
|
||||
|
||||
message MovedResponse {}
|
||||
|
||||
message RowCol {
|
||||
int32 row = 1;
|
||||
int32 col = 2;
|
||||
}
|
||||
|
||||
message CursorPosition {
|
||||
string buffer = 1;
|
||||
RowCol start = 2;
|
||||
RowCol end = 3;
|
||||
}
|
||||
|
||||
message CursorEvent {
|
||||
string user = 1;
|
||||
CursorPosition position = 2;
|
||||
}
|
||||
|
||||
message UserIdentity {
|
||||
string id = 1;
|
||||
}
|
||||
|
|
@ -1,19 +0,0 @@
|
|||
[package]
|
||||
name = "codemp-server"
|
||||
version = "0.2.0"
|
||||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
codemp = { path = ".." }
|
||||
tracing = "0.1"
|
||||
tracing-subscriber = "0.3"
|
||||
tonic = { version = "0.9", features = ["tls", "tls-roots"] }
|
||||
prost = "0.11.8"
|
||||
md5 = "0.7.0"
|
||||
uuid = { version = "1.3.1", features = ["v4"] }
|
||||
operational-transform = { version = "0.6", features = ["serde"] }
|
||||
tokio = { version = "1.0", features = ["macros", "rt-multi-thread", "sync", "full"] }
|
||||
tokio-stream = "0.1"
|
||||
serde = "1"
|
||||
serde_json = "1"
|
||||
clap = { version = "4.2.1", features = ["derive"] }
|
||||
|
|
@ -1,101 +0,0 @@
|
|||
use codemp::{proto::{RawOp, OperationRequest}, errors::IgnorableError};
|
||||
use tokio::sync::{mpsc, broadcast, watch, oneshot};
|
||||
use tracing::{error, warn};
|
||||
// use md5::Digest;
|
||||
|
||||
use operational_transform::OperationSeq;
|
||||
|
||||
pub trait BufferStore<T> {
|
||||
fn get(&self, key: &T) -> Option<&BufferHandle>;
|
||||
fn put(&mut self, key: T, handle: BufferHandle) -> Option<BufferHandle>;
|
||||
|
||||
fn handle(&mut self, key: T, content: Option<String>) {
|
||||
let handle = BufferHandle::new(content);
|
||||
self.put(key, handle);
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct BufferHandle {
|
||||
pub edit: mpsc::Sender<(oneshot::Sender<bool>, OperationRequest)>,
|
||||
events: broadcast::Sender<RawOp>,
|
||||
// pub digest: watch::Receiver<Digest>,
|
||||
pub content: watch::Receiver<String>,
|
||||
}
|
||||
|
||||
impl BufferHandle {
|
||||
fn new(init: Option<String>) -> Self {
|
||||
let init_val = init.unwrap_or("".into());
|
||||
let (edits_tx, edits_rx) = mpsc::channel(64); // TODO hardcoded size
|
||||
let (events_tx, _events_rx) = broadcast::channel(64); // TODO hardcoded size
|
||||
// let (digest_tx, digest_rx) = watch::channel(md5::compute(&init_val));
|
||||
let (content_tx, content_rx) = watch::channel(init_val.clone());
|
||||
|
||||
let events_tx_clone = events_tx.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let worker = BufferWorker {
|
||||
store: init_val,
|
||||
edits: edits_rx,
|
||||
events: events_tx_clone,
|
||||
// digest: digest_tx,
|
||||
content: content_tx,
|
||||
};
|
||||
worker.work().await
|
||||
});
|
||||
|
||||
BufferHandle {
|
||||
edit: edits_tx,
|
||||
events: events_tx,
|
||||
// digest: digest_rx,
|
||||
content: content_rx,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn subscribe(&self) -> broadcast::Receiver<RawOp> {
|
||||
self.events.subscribe()
|
||||
}
|
||||
}
|
||||
|
||||
struct BufferWorker {
|
||||
store: String,
|
||||
edits: mpsc::Receiver<(oneshot::Sender<bool>, OperationRequest)>,
|
||||
events: broadcast::Sender<RawOp>,
|
||||
// digest: watch::Sender<Digest>,
|
||||
content: watch::Sender<String>,
|
||||
}
|
||||
|
||||
impl BufferWorker {
|
||||
async fn work(mut self) {
|
||||
loop {
|
||||
match self.edits.recv().await {
|
||||
None => break warn!("channel closed"),
|
||||
Some((ack, v)) => match serde_json::from_str::<OperationSeq>(&v.opseq) {
|
||||
Err(e) => {
|
||||
ack.send(false).unwrap_or_warn("could not reject undeserializable opseq");
|
||||
error!("could not deserialize opseq: {}", e);
|
||||
},
|
||||
Ok(op) => match op.apply(&self.store) {
|
||||
Err(e) => {
|
||||
ack.send(false).unwrap_or_warn("could not reject unappliable opseq");
|
||||
error!("coult not apply OpSeq '{:?}' on '{}' : {}", v, self.store, e); // TODO
|
||||
}
|
||||
Ok(res) => {
|
||||
self.store = res;
|
||||
let msg = RawOp {
|
||||
opseq: v.opseq,
|
||||
user: v.user
|
||||
};
|
||||
// if let Err(e) = self.digest.send(md5::compute(&self.store)) {
|
||||
// error!("could not update digest: {}", e);
|
||||
// }
|
||||
ack.send(true).unwrap_or_warn("could not accept opseq");
|
||||
self.content.send(self.store.clone()).unwrap_or_warn("could not update content");
|
||||
self.events.send(msg).unwrap_or_warn("could not broadcast OpSeq");
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,2 +0,0 @@
|
|||
pub mod actor;
|
||||
pub mod service;
|
||||
|
|
@ -1,113 +0,0 @@
|
|||
use std::{pin::Pin, sync::{Arc, RwLock}, collections::HashMap};
|
||||
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
use tokio_stream::{Stream, wrappers::ReceiverStream}; // TODO example used this?
|
||||
|
||||
use codemp::proto::{buffer_server::Buffer, RawOp, BufferPayload, BufferResponse, OperationRequest, BufferEditResponse, BufferCreateResponse};
|
||||
use tracing::info;
|
||||
|
||||
use super::actor::{BufferHandle, BufferStore};
|
||||
|
||||
type OperationStream = Pin<Box<dyn Stream<Item = Result<RawOp, Status>> + Send>>;
|
||||
|
||||
struct BufferMap {
|
||||
store: HashMap<String, BufferHandle>,
|
||||
}
|
||||
|
||||
impl From::<HashMap<String, BufferHandle>> for BufferMap {
|
||||
fn from(value: HashMap<String, BufferHandle>) -> Self {
|
||||
BufferMap { store: value }
|
||||
}
|
||||
}
|
||||
|
||||
impl BufferStore<String> for BufferMap {
|
||||
fn get(&self, key: &String) -> Option<&BufferHandle> {
|
||||
self.store.get(key)
|
||||
}
|
||||
fn put(&mut self, key: String, handle: BufferHandle) -> Option<BufferHandle> {
|
||||
self.store.insert(key, handle)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct BufferService {
|
||||
map: Arc<RwLock<BufferMap>>,
|
||||
}
|
||||
|
||||
impl Default for BufferService {
|
||||
fn default() -> BufferService {
|
||||
BufferService {
|
||||
map: Arc::new(RwLock::new(HashMap::new().into())),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Buffer for BufferService {
|
||||
type AttachStream = OperationStream;
|
||||
|
||||
async fn attach(&self, req: Request<BufferPayload>) -> Result<Response<OperationStream>, Status> {
|
||||
let request = req.into_inner();
|
||||
let myself = request.user;
|
||||
match self.map.read().unwrap().get(&request.path) {
|
||||
None => Err(Status::not_found("path not found")),
|
||||
Some(handle) => {
|
||||
let (tx, rx) = mpsc::channel(128);
|
||||
let mut sub = handle.subscribe();
|
||||
tokio::spawn(async move {
|
||||
while let Ok(v) = sub.recv().await {
|
||||
if v.user == myself { continue }
|
||||
tx.send(Ok(v)).await.unwrap(); // TODO unnecessary channel?
|
||||
}
|
||||
});
|
||||
let output_stream = ReceiverStream::new(rx);
|
||||
info!("registered new subscriber on buffer");
|
||||
Ok(Response::new(Box::pin(output_stream)))
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
async fn edit(&self, req:Request<OperationRequest>) -> Result<Response<BufferEditResponse>, Status> {
|
||||
let request = req.into_inner();
|
||||
let tx = match self.map.read().unwrap().get(&request.path) {
|
||||
Some(handle) => {
|
||||
// if format!("{:x}", *handle.digest.borrow()) != request.hash {
|
||||
// return Ok(Response::new(BufferResponse { accepted : false } ));
|
||||
// }
|
||||
handle.edit.clone()
|
||||
},
|
||||
None => return Err(Status::not_found("path not found")),
|
||||
};
|
||||
info!("sending edit to buffer: {}", request.opseq);
|
||||
let (ack, status) = oneshot::channel();
|
||||
match tx.send((ack, request)).await {
|
||||
Err(e) => Err(Status::internal(format!("error sending edit to buffer actor: {}", e))),
|
||||
Ok(()) => {
|
||||
match status.await {
|
||||
Ok(_accepted) => Ok(Response::new(BufferEditResponse { })),
|
||||
Err(e) => Err(Status::internal(format!("error receiving edit result: {}", e))),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn create(&self, req:Request<BufferPayload>) -> Result<Response<BufferCreateResponse>, Status> {
|
||||
let request = req.into_inner();
|
||||
let _handle = self.map.write().unwrap().handle(request.path, request.content);
|
||||
info!("created new buffer");
|
||||
Ok(Response::new(BufferCreateResponse { }))
|
||||
}
|
||||
|
||||
async fn sync(&self, req: Request<BufferPayload>) -> Result<Response<BufferResponse>, Status> {
|
||||
let request = req.into_inner();
|
||||
match self.map.read().unwrap().get(&request.path) {
|
||||
None => Err(Status::not_found("requested buffer does not exist")),
|
||||
Some(buf) => {
|
||||
info!("synching buffer");
|
||||
let answ = BufferResponse { content: buf.content.borrow().clone() };
|
||||
Ok(Response::new(answ))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1 +0,0 @@
|
|||
pub mod service;
|
||||
|
|
@ -1,52 +0,0 @@
|
|||
use std::pin::Pin;
|
||||
|
||||
use tokio::sync::{mpsc, broadcast};
|
||||
use tonic::{Request, Response, Status};
|
||||
|
||||
use tokio_stream::{Stream, wrappers::ReceiverStream}; // TODO example used this?
|
||||
|
||||
use codemp::proto::{cursor_server::Cursor, UserIdentity, CursorPosition, MovedResponse};
|
||||
use tracing::info;
|
||||
|
||||
type CursorStream = Pin<Box<dyn Stream<Item = Result<CursorPosition, Status>> + Send>>;
|
||||
|
||||
pub struct CursorService {
|
||||
cursor: broadcast::Sender<CursorPosition>,
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
impl Cursor for CursorService {
|
||||
type ListenStream = CursorStream;
|
||||
|
||||
async fn listen(&self, req: Request<UserIdentity>) -> Result<Response<CursorStream>, Status> {
|
||||
let mut sub = self.cursor.subscribe();
|
||||
let myself = req.into_inner().id;
|
||||
let (tx, rx) = mpsc::channel(128);
|
||||
tokio::spawn(async move {
|
||||
while let Ok(v) = sub.recv().await {
|
||||
if v.user == myself { continue }
|
||||
tx.send(Ok(v)).await.unwrap(); // TODO unnecessary channel?
|
||||
}
|
||||
});
|
||||
let output_stream = ReceiverStream::new(rx);
|
||||
info!("registered new subscriber to cursor updates");
|
||||
Ok(Response::new(Box::pin(output_stream)))
|
||||
}
|
||||
|
||||
async fn moved(&self, req:Request<CursorPosition>) -> Result<Response<MovedResponse>, Status> {
|
||||
match self.cursor.send(req.into_inner()) {
|
||||
Ok(_) => Ok(Response::new(MovedResponse { })),
|
||||
Err(e) => Err(Status::internal(format!("could not broadcast cursor update: {}", e))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for CursorService {
|
||||
fn default() -> Self {
|
||||
let (cur_tx, _cur_rx) = broadcast::channel(64); // TODO hardcoded capacity
|
||||
// TODO don't drop receiver because sending event when there are no receivers throws an error
|
||||
CursorService {
|
||||
cursor: cur_tx,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,50 +0,0 @@
|
|||
//! # codemp Server
|
||||
//!
|
||||
//! The codemp server itself, in charge of handling the global state, merging operations from
|
||||
//! all clients and synching everyone's cursor.
|
||||
//!
|
||||
|
||||
use clap::Parser;
|
||||
use codemp::proto::buffer_server::BufferServer;
|
||||
use codemp::proto::cursor_server::CursorServer;
|
||||
use tracing::info;
|
||||
use tonic::transport::Server;
|
||||
|
||||
mod buffer;
|
||||
mod cursor;
|
||||
|
||||
use crate::buffer::service::BufferService;
|
||||
use crate::cursor::service::CursorService;
|
||||
|
||||
#[derive(Parser, Debug)]
|
||||
struct CliArgs {
|
||||
|
||||
/// address to listen on
|
||||
#[arg(long, default_value = "[::1]:50051")]
|
||||
host: String,
|
||||
|
||||
/// enable debug log level
|
||||
#[arg(long, default_value_t = false)]
|
||||
debug: bool,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
let args = CliArgs::parse();
|
||||
|
||||
tracing_subscriber::fmt()
|
||||
.with_writer(std::io::stdout)
|
||||
.with_max_level(if args.debug { tracing::Level::DEBUG } else { tracing::Level::INFO })
|
||||
.init();
|
||||
|
||||
info!(">> codemp server");
|
||||
info!("binding on {}", args.host);
|
||||
|
||||
Server::builder()
|
||||
.add_service(BufferServer::new(BufferService::default()))
|
||||
.add_service(CursorServer::new(CursorService::default()))
|
||||
.serve(args.host.parse()?)
|
||||
.await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -1,41 +0,0 @@
|
|||
use operational_transform::OperationSeq;
|
||||
use tokio::sync::{watch, mpsc, broadcast, Mutex};
|
||||
use tonic::async_trait;
|
||||
|
||||
use crate::{Controller, CodempError};
|
||||
use crate::buffer::factory::{leading_noop, tailing_noop, OperationFactory};
|
||||
|
||||
use super::TextChange;
|
||||
|
||||
pub struct BufferController {
|
||||
content: watch::Receiver<String>,
|
||||
operations: mpsc::Sender<OperationSeq>,
|
||||
stream: Mutex<broadcast::Receiver<OperationSeq>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl OperationFactory for BufferController {
|
||||
fn content(&self) -> String {
|
||||
self.content.borrow().clone()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Controller<TextChange> for BufferController {
|
||||
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?)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,84 +0,0 @@
|
|||
use std::ops::Range;
|
||||
|
||||
use operational_transform::{OperationSeq, Operation};
|
||||
use similar::{TextDiff, ChangeTag};
|
||||
|
||||
pub const fn leading_noop(seq: &[Operation]) -> u64 { count_noop(seq.first()) }
|
||||
pub const fn tailing_noop(seq: &[Operation]) -> u64 { count_noop(seq.last()) }
|
||||
|
||||
const fn count_noop(op: Option<&Operation>) -> u64 {
|
||||
match op {
|
||||
None => 0,
|
||||
Some(Operation::Retain(n)) => *n,
|
||||
Some(_) => 0,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn op_effective_range(op: &OperationSeq) -> Range<u64> {
|
||||
let first = leading_noop(op.ops());
|
||||
let last = op.base_len() as u64 - tailing_noop(op.ops());
|
||||
first..last
|
||||
}
|
||||
|
||||
pub trait OperationFactory {
|
||||
fn content(&self) -> String;
|
||||
|
||||
fn replace(&self, txt: &str) -> Option<OperationSeq> {
|
||||
self.delta(0, txt, self.content().len())
|
||||
}
|
||||
|
||||
fn delta(&self, start: usize, txt: &str, end: usize) -> Option<OperationSeq> {
|
||||
let mut out = OperationSeq::default();
|
||||
let content = self.content();
|
||||
let tail_skip = content.len() - end;
|
||||
let content_slice = &content[start..tail_skip];
|
||||
|
||||
if content_slice == txt {
|
||||
// if slice equals given text, no operation should be taken
|
||||
return None;
|
||||
}
|
||||
|
||||
out.retain(start as u64);
|
||||
|
||||
let diff = TextDiff::from_chars(content_slice, txt);
|
||||
|
||||
for change in diff.iter_all_changes() {
|
||||
match change.tag() {
|
||||
ChangeTag::Equal => out.retain(1),
|
||||
ChangeTag::Delete => out.delete(1),
|
||||
ChangeTag::Insert => out.insert(change.value()),
|
||||
}
|
||||
}
|
||||
|
||||
out.retain(tail_skip as u64);
|
||||
|
||||
Some(out)
|
||||
}
|
||||
|
||||
fn insert(&self, txt: &str, pos: u64) -> OperationSeq {
|
||||
let mut out = OperationSeq::default();
|
||||
let total = self.content().len() as u64;
|
||||
out.retain(pos);
|
||||
out.insert(txt);
|
||||
out.retain(total - pos);
|
||||
out
|
||||
}
|
||||
|
||||
fn delete(&self, pos: u64, count: u64) -> OperationSeq {
|
||||
let mut out = OperationSeq::default();
|
||||
let len = self.content().len() as u64;
|
||||
out.retain(pos - count);
|
||||
out.delete(count);
|
||||
out.retain(len - pos);
|
||||
out
|
||||
}
|
||||
|
||||
fn cancel(&self, pos: u64, count: u64) -> OperationSeq {
|
||||
let mut out = OperationSeq::default();
|
||||
let len = self.content().len() as u64;
|
||||
out.retain(pos);
|
||||
out.delete(count);
|
||||
out.retain(len - (pos+count));
|
||||
out
|
||||
}
|
||||
}
|
||||
|
|
@ -1,11 +0,0 @@
|
|||
use std::ops::Range;
|
||||
|
||||
pub(crate) mod worker;
|
||||
pub mod controller;
|
||||
pub mod factory;
|
||||
|
||||
|
||||
pub struct TextChange {
|
||||
pub span: Range<usize>,
|
||||
pub content: String,
|
||||
}
|
||||
|
|
@ -1,114 +0,0 @@
|
|||
use std::{sync::Arc, collections::VecDeque};
|
||||
|
||||
use operational_transform::OperationSeq;
|
||||
use tokio::sync::{watch, mpsc, broadcast, Mutex};
|
||||
use tonic::transport::Channel;
|
||||
use tonic::{async_trait, Streaming};
|
||||
|
||||
use crate::proto::{OperationRequest, RawOp};
|
||||
use crate::proto::buffer_client::BufferClient;
|
||||
use crate::ControllerWorker;
|
||||
|
||||
use super::TextChange;
|
||||
use super::controller::BufferController;
|
||||
|
||||
|
||||
pub(crate) struct BufferControllerWorker {
|
||||
uid: String,
|
||||
pub(crate) content: watch::Sender<String>,
|
||||
pub(crate) operations: mpsc::Receiver<OperationSeq>,
|
||||
pub(crate) stream: Arc<broadcast::Sender<OperationSeq>>,
|
||||
pub(crate) queue: VecDeque<OperationSeq>,
|
||||
receiver: watch::Receiver<String>,
|
||||
sender: mpsc::Sender<OperationSeq>,
|
||||
buffer: String,
|
||||
path: String,
|
||||
}
|
||||
|
||||
impl BufferControllerWorker {
|
||||
pub fn new(uid: String, buffer: &str, path: &str) -> Self {
|
||||
let (txt_tx, txt_rx) = watch::channel(buffer.to_string());
|
||||
let (op_tx, op_rx) = mpsc::channel(64);
|
||||
let (s_tx, _s_rx) = broadcast::channel(64);
|
||||
BufferControllerWorker {
|
||||
uid,
|
||||
content: txt_tx,
|
||||
operations: op_rx,
|
||||
stream: Arc::new(s_tx),
|
||||
receiver: txt_rx,
|
||||
sender: op_tx,
|
||||
queue: VecDeque::new(),
|
||||
buffer: buffer.to_string(),
|
||||
path: path.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ControllerWorker<TextChange> for BufferControllerWorker {
|
||||
type Controller = BufferController;
|
||||
type Tx = BufferClient<Channel>;
|
||||
type Rx = Streaming<RawOp>;
|
||||
|
||||
fn subscribe(&self) -> BufferController {
|
||||
BufferController {
|
||||
content: self.receiver.clone(),
|
||||
operations: self.sender.clone(),
|
||||
stream: Mutex::new(self.stream.subscribe()),
|
||||
}
|
||||
}
|
||||
|
||||
async fn work(mut self, mut tx: Self::Tx, mut rx: Self::Rx) {
|
||||
loop {
|
||||
let op = tokio::select! {
|
||||
Some(operation) = recv_opseq(&mut rx) => {
|
||||
let mut out = operation;
|
||||
for op in self.queue.iter_mut() {
|
||||
(*op, out) = op.transform(&out).unwrap();
|
||||
}
|
||||
self.stream.send(out.clone()).unwrap();
|
||||
out
|
||||
},
|
||||
Some(op) = self.operations.recv() => {
|
||||
self.queue.push_back(op.clone());
|
||||
op
|
||||
},
|
||||
else => break
|
||||
};
|
||||
self.buffer = op.apply(&self.buffer).unwrap();
|
||||
self.content.send(self.buffer.clone()).unwrap();
|
||||
|
||||
while let Some(op) = self.queue.get(0) {
|
||||
if !send_opseq(&mut tx, self.uid.clone(), self.path.clone(), op.clone()).await { break }
|
||||
self.queue.pop_front();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn send_opseq(tx: &mut BufferClient<Channel>, uid: String, path: String, op: OperationSeq) -> bool {
|
||||
let req = OperationRequest {
|
||||
hash: "".into(),
|
||||
opseq: serde_json::to_string(&op).unwrap(),
|
||||
path,
|
||||
user: uid,
|
||||
};
|
||||
match tx.edit(req).await {
|
||||
Ok(_) => true,
|
||||
Err(e) => {
|
||||
tracing::error!("error sending edit: {}", e);
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn recv_opseq(rx: &mut Streaming<RawOp>) -> Option<OperationSeq> {
|
||||
match rx.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
|
||||
}
|
||||
}
|
||||
}
|
||||
117
src/client.rs
117
src/client.rs
|
|
@ -1,117 +0,0 @@
|
|||
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() })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,51 +0,0 @@
|
|||
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,26 +0,0 @@
|
|||
pub(crate) mod worker;
|
||||
pub mod controller;
|
||||
|
||||
use crate::proto::{RowCol, CursorPosition};
|
||||
|
||||
impl From::<RowCol> for (i32, i32) {
|
||||
fn from(pos: RowCol) -> (i32, i32) {
|
||||
(pos.row, pos.col)
|
||||
}
|
||||
}
|
||||
|
||||
impl From::<(i32, i32)> for RowCol {
|
||||
fn from((row, col): (i32, i32)) -> Self {
|
||||
RowCol { row, col }
|
||||
}
|
||||
}
|
||||
|
||||
impl CursorPosition {
|
||||
pub fn start(&self) -> RowCol {
|
||||
self.start.clone().unwrap_or((0, 0).into())
|
||||
}
|
||||
|
||||
pub fn end(&self) -> RowCol {
|
||||
self.end.clone().unwrap_or((0, 0).into())
|
||||
}
|
||||
}
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
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,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1,77 +0,0 @@
|
|||
use std::{error::Error, fmt::Display};
|
||||
|
||||
use tokio::sync::{mpsc, broadcast};
|
||||
use tonic::{Status, Code};
|
||||
use tracing::warn;
|
||||
|
||||
pub trait IgnorableError {
|
||||
fn unwrap_or_warn(self, msg: &str);
|
||||
}
|
||||
|
||||
impl<T, E> IgnorableError for Result<T, E>
|
||||
where E : std::fmt::Display {
|
||||
fn unwrap_or_warn(self, msg: &str) {
|
||||
match self {
|
||||
Ok(_) => {},
|
||||
Err(e) => warn!("{}: {}", msg, e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TODO split this into specific errors for various parts of the library
|
||||
#[derive(Debug)]
|
||||
pub enum CodempError {
|
||||
Transport {
|
||||
status: Code,
|
||||
message: String,
|
||||
},
|
||||
Channel {
|
||||
send: bool
|
||||
},
|
||||
InvalidState {
|
||||
msg: String,
|
||||
},
|
||||
|
||||
// TODO filler error, remove later
|
||||
Filler {
|
||||
message: String,
|
||||
},
|
||||
}
|
||||
|
||||
impl Error for CodempError {}
|
||||
|
||||
impl Display for CodempError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Transport { status, message } => write!(f, "Transport error: ({}) {}", status, message),
|
||||
Self::Channel { send } => write!(f, "Channel error (send:{})", send),
|
||||
_ => write!(f, "Unknown error"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Status> for CodempError {
|
||||
fn from(status: Status) -> Self {
|
||||
CodempError::Transport { status: status.code(), message: status.message().to_string() }
|
||||
}
|
||||
}
|
||||
|
||||
impl From<tonic::transport::Error> for CodempError {
|
||||
fn from(err: tonic::transport::Error) -> Self {
|
||||
CodempError::Transport {
|
||||
status: Code::Unknown, message: format!("underlying transport error: {:?}", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> From<mpsc::error::SendError<T>> for CodempError {
|
||||
fn from(_value: mpsc::error::SendError<T>) -> Self {
|
||||
CodempError::Channel { send: true }
|
||||
}
|
||||
}
|
||||
|
||||
impl From<broadcast::error::RecvError> for CodempError {
|
||||
fn from(_value: broadcast::error::RecvError) -> Self {
|
||||
CodempError::Channel { send: false }
|
||||
}
|
||||
}
|
||||
|
|
@ -1,81 +0,0 @@
|
|||
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() })
|
||||
}
|
||||
}
|
||||
39
src/lib.rs
39
src/lib.rs
|
|
@ -1,39 +0,0 @@
|
|||
pub mod cursor;
|
||||
pub mod errors;
|
||||
pub mod buffer;
|
||||
|
||||
pub mod client;
|
||||
|
||||
#[cfg(feature = "static")]
|
||||
pub mod instance;
|
||||
|
||||
pub use tonic;
|
||||
pub use tokio;
|
||||
pub use operational_transform as ot;
|
||||
|
||||
#[cfg(feature = "proto")]
|
||||
#[allow(non_snake_case)]
|
||||
pub mod proto {
|
||||
tonic::include_proto!("codemp.buffer");
|
||||
tonic::include_proto!("codemp.cursor");
|
||||
}
|
||||
|
||||
pub use errors::CodempError;
|
||||
|
||||
#[tonic::async_trait] // TODO move this somewhere?
|
||||
pub(crate) trait ControllerWorker<T> {
|
||||
type Controller : Controller<T>;
|
||||
type Tx;
|
||||
type Rx;
|
||||
|
||||
fn subscribe(&self) -> Self::Controller;
|
||||
async fn work(self, tx: Self::Tx, rx: Self::Rx);
|
||||
}
|
||||
|
||||
#[tonic::async_trait]
|
||||
pub trait Controller<T> {
|
||||
type Input;
|
||||
|
||||
async fn send(&self, x: Self::Input) -> Result<(), CodempError>;
|
||||
async fn recv(&self) -> Result<T, CodempError>;
|
||||
}
|
||||
Loading…
Reference in a new issue