From 27048efac9927860d1c8ffc26b4ae68960c73fd9 Mon Sep 17 00:00:00 2001 From: alemi Date: Wed, 16 Aug 2023 23:29:45 +0200 Subject: [PATCH] chore: removed rest of the project: --- client/vscode/.bundle.sh => .bundle.sh | 0 Cargo.toml | 40 +-- build.rs | 5 - client/nvim/Cargo.toml | 16 -- client/nvim/codemp.lua | 190 ------------ client/nvim/src/main.rs | 318 --------------------- client/vscode/Cargo.toml | 22 -- client/vscode/src/lib.rs | 270 ----------------- client/vscode/package.json => package.json | 0 proto/buffer.proto | 35 --- proto/cursor.proto | 30 -- server/Cargo.toml | 19 -- server/src/buffer/actor.rs | 101 ------- server/src/buffer/mod.rs | 2 - server/src/buffer/service.rs | 113 -------- server/src/cursor/mod.rs | 1 - server/src/cursor/service.rs | 52 ---- server/src/main.rs | 50 ---- src/buffer/controller.rs | 41 --- src/buffer/factory.rs | 84 ------ src/buffer/mod.rs | 11 - src/buffer/worker.rs | 114 -------- src/client.rs | 117 -------- src/cursor/controller.rs | 51 ---- src/cursor/mod.rs | 26 -- src/cursor/worker.rs | 54 ---- src/errors.rs | 77 ----- {client/vscode/src => src}/extension.js | 0 src/instance.rs | 81 ------ src/lib.rs | 293 +++++++++++++++++-- 30 files changed, 276 insertions(+), 1937 deletions(-) rename client/vscode/.bundle.sh => .bundle.sh (100%) delete mode 100644 build.rs delete mode 100644 client/nvim/Cargo.toml delete mode 100644 client/nvim/codemp.lua delete mode 100644 client/nvim/src/main.rs delete mode 100644 client/vscode/Cargo.toml delete mode 100644 client/vscode/src/lib.rs rename client/vscode/package.json => package.json (100%) delete mode 100644 proto/buffer.proto delete mode 100644 proto/cursor.proto delete mode 100644 server/Cargo.toml delete mode 100644 server/src/buffer/actor.rs delete mode 100644 server/src/buffer/mod.rs delete mode 100644 server/src/buffer/service.rs delete mode 100644 server/src/cursor/mod.rs delete mode 100644 server/src/cursor/service.rs delete mode 100644 server/src/main.rs delete mode 100644 src/buffer/controller.rs delete mode 100644 src/buffer/factory.rs delete mode 100644 src/buffer/mod.rs delete mode 100644 src/buffer/worker.rs delete mode 100644 src/client.rs delete mode 100644 src/cursor/controller.rs delete mode 100644 src/cursor/mod.rs delete mode 100644 src/cursor/worker.rs delete mode 100644 src/errors.rs rename {client/vscode/src => src}/extension.js (100%) delete mode 100644 src/instance.rs diff --git a/client/vscode/.bundle.sh b/.bundle.sh similarity index 100% rename from client/vscode/.bundle.sh rename to .bundle.sh diff --git a/Cargo.toml b/Cargo.toml index 0e6edc6..457e394 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,34 +1,22 @@ -[workspace] -members = ["client/nvim", "client/vscode", "server"] - [package] -name = "codemp" -version = "0.3.1" +name = "codemp-vscode" +version = "0.0.1" +description = "VSCode extension for CodeMP" edition = "2021" +exclude = ["index.node"] [lib] -name = "codemp" +crate-type = ["cdylib"] [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"] +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"] } diff --git a/build.rs b/build.rs deleted file mode 100644 index 8c727c0..0000000 --- a/build.rs +++ /dev/null @@ -1,5 +0,0 @@ -fn main() -> Result<(), Box> { - tonic_build::compile_protos("proto/buffer.proto")?; - tonic_build::compile_protos("proto/cursor.proto")?; - Ok(()) -} diff --git a/client/nvim/Cargo.toml b/client/nvim/Cargo.toml deleted file mode 100644 index 59949b7..0000000 --- a/client/nvim/Cargo.toml +++ /dev/null @@ -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" diff --git a/client/nvim/codemp.lua b/client/nvim/codemp.lua deleted file mode 100644 index 099c98a..0000000 --- a/client/nvim/codemp.lua +++ /dev/null @@ -1,190 +0,0 @@ -local BINARY = vim.g.codemp_binary or "./codemp-client-nvim" - -local M = {} - -M.jobid = nil -M.create = function(path, content) return vim.rpcrequest(M.jobid, "create", path, content) end -M.insert = function(path, txt, pos) return vim.rpcrequest(M.jobid, "insert", path, txt, pos) end -M.delete = function(path, pos, count) return vim.rpcrequest(M.jobid, "delete", path, pos, count) end -M.replace = function(path, txt) return vim.rpcrequest(M.jobid, "replace", path, txt) end -M.cursor = function(path, cur) return vim.rpcrequest(M.jobid, "cursor", path, cur[1][1], cur[1][2], cur[2][1], cur[2][2]) end -M.attach = function(path) return vim.rpcrequest(M.jobid, "attach", path) end -M.listen = function(path) return vim.rpcrequest(M.jobid, "listen", path) end -M.detach = function(path) return vim.rpcrequest(M.jobid, "detach", path) end - -local function cursor_offset() - local cursor = vim.api.nvim_win_get_cursor(0) - return vim.fn.line2byte(cursor[1]) + cursor[2] - 1 -end - -local codemp_autocmds = vim.api.nvim_create_augroup("CodempAuGroup", { clear = true }) - -local function get_cursor_range() - local mode = vim.fn.mode() - if mode == "" or mode == "s" or mode == "Vs" or mode == "V" or mode == "vs" or mode == "v" then - local start = vim.fn.getpos("'<") - local finish = vim.fn.getpos("'>") - return { - { start[2], start[3] }, - { finish[2], finish[3] } - } - else - local cursor = vim.api.nvim_win_get_cursor(0) - return { - { cursor[1], cursor[2] }, - { cursor[1], cursor[2] + 1 }, - } - end -end - -local function hook_callbacks(path, buffer) - vim.api.nvim_create_autocmd( - { "InsertCharPre" }, - { - callback = function(_) - pcall(M.insert, path, vim.v.char, cursor_offset()) -- TODO log errors - end, - buffer = buffer, - group = codemp_autocmds, - } - ) - vim.api.nvim_create_autocmd( - { "CursorMoved", "CompleteDone", "InsertEnter", "InsertLeave" }, - { - callback = function(args) - local lines = vim.api.nvim_buf_get_lines(args.buf, 0, -1, false) - pcall(M.replace, path, vim.fn.join(lines, "\n")) -- TODO log errors - pcall(M.cursor, path, get_cursor_range()) -- TODO log errors - end, - buffer = buffer, - group = codemp_autocmds, - } - ) - local last_line = 0 - vim.api.nvim_create_autocmd( - { "CursorMovedI" }, - { - callback = function(args) - local cursor = get_cursor_range() - pcall(M.cursor, path, cursor) -- TODO log errors - if cursor[1][1] == last_line then - return - end - last_line = cursor[1][1] - local lines = vim.api.nvim_buf_get_lines(args.buf, 0, -1, false) - pcall(M.replace, path, vim.fn.join(lines, "\n")) -- TODO log errors - end, - buffer = buffer, - group = codemp_autocmds, - } - ) - vim.keymap.set('i', '', function() pcall(M.delete, path, cursor_offset(), 1) return '' end, {expr = true, buffer = buffer}) -- TODO log errors - vim.keymap.set('i', '', function() pcall(M.delete, path, cursor_offset() + 1, 1) return '' end, {expr = true, buffer = buffer}) -- TODO log errors -end - -local function unhook_callbacks(buffer) - vim.api.nvim_clear_autocmds({ group = codemp_autocmds, buffer = buffer }) - vim.keymap.del('i', '', { buffer = buffer }) - vim.keymap.del('i', '', { buffer = buffer }) -end - -local function auto_address(addr) - if not string.find(addr, "://") then - addr = string.format("http://%s", addr) - end - if not string.find(addr, ":", 7) then -- skip first 7 chars because 'https://' - addr = string.format("%s:50051", addr) - end - return addr -end - -vim.api.nvim_create_user_command('Connect', - function(args) - if M.jobid ~= nil and M.jobid > 0 then - print("already connected, disconnect first") - return - end - local bin_args = { BINARY } - if #args.fargs > 0 then - table.insert(bin_args, "--host") - table.insert(bin_args, auto_address(args.fargs[1])) - end - if vim.g.codemp_remote_debug then - table.insert(bin_args, "--remote-debug") - table.insert(bin_args, vim.g.codemp_remote_debug) - end - if args.bang then - table.insert(bin_args, "--debug") - end - M.jobid = vim.fn.jobstart( - bin_args, - { - rpc = true, - on_stderr = function(_, data, _) - for _, line in pairs(data) do - print(line) - end - -- print(vim.fn.join(data, "\n")) - end, - stderr_buffered = false, - env = { RUST_BACKTRACE = 1 } - } - ) - if M.jobid <= 0 then - print("[!] could not start codemp client") - end - end, -{ nargs='?', bang=true }) - -vim.api.nvim_create_user_command('Stop', - function(_) - vim.fn.jobstop(M.jobid) - M.jobid = nil - end, -{ bang=true }) - -vim.api.nvim_create_user_command('Share', - function(args) - if M.jobid == nil or M.jobid <= 0 then - print("[!] connect to codemp server first") - return - end - local path = args.fargs[1] - local bufnr = vim.api.nvim_get_current_buf() - local lines = vim.api.nvim_buf_get_lines(bufnr, 0, -1, false) - vim.opt.fileformat = "unix" - M.create(path, vim.fn.join(lines, "\n")) - hook_callbacks(path, bufnr) - M.attach(path) - M.listen(path) - end, -{ nargs=1 }) - -vim.api.nvim_create_user_command('Join', - function(args) - if M.jobid == nil or M.jobid <= 0 then - print("[!] connect to codemp server first") - return - end - local path = args.fargs[1] - local bufnr = vim.api.nvim_get_current_buf() - vim.opt.fileformat = "unix" - hook_callbacks(path, bufnr) - M.attach(path) - M.listen(path) - end, -{ nargs=1 }) - -vim.api.nvim_create_user_command('Detach', - function(args) - local bufnr = vim.api.nvim_get_current_buf() - if M.detach(args.fargs[1]) then - unhook_callbacks(bufnr) - print("[/] detached from buffer") - else - print("[!] error detaching from buffer") - end - end, -{ nargs=1 }) - -return M diff --git a/client/nvim/src/main.rs b/client/nvim/src/main.rs deleted file mode 100644 index a585225..0000000 --- a/client/nvim/src/main.rs +++ /dev/null @@ -1,318 +0,0 @@ -use std::sync::Arc; -use std::{net::TcpStream, sync::Mutex, collections::BTreeMap}; - -use codemp::client::CodempClient; -use codemp::controller::buffer::{OperationControllerHandle, OperationControllerSubscriber}; -use codemp::controller::cursor::{CursorControllerHandle, CursorSubscriber}; -use codemp::factory::OperationFactory; -use codemp::proto::buffer_client::BufferClient; -use codemp::tokio; - -use rmpv::Value; -use clap::Parser; - -use nvim_rs::{compat::tokio::Compat, create::tokio as create, Handler, Neovim}; -use tracing::{error, warn, debug, info}; - -#[derive(Clone)] -struct NeovimHandler { - client: CodempClient, - factories: Arc>>, - cursors: Arc>>, -} - -fn nullable_optional_str(args: &[Value], index: usize) -> Option { - Some(args.get(index)?.as_str()?.to_string()) -} - -fn default_empty_str(args: &[Value], index: usize) -> String { - nullable_optional_str(args, index).unwrap_or("".into()) -} - -fn nullable_optional_number(args: &[Value], index: usize) -> Option { - args.get(index)?.as_i64() -} - -fn default_zero_number(args: &[Value], index: usize) -> i64 { - nullable_optional_number(args, index).unwrap_or(0) -} - -impl NeovimHandler { - fn buffer_controller(&self, path: &String) -> Option { - Some(self.factories.lock().unwrap().get(path)?.clone()) - } - - fn cursor_controller(&self, path: &String) -> Option { - Some(self.cursors.lock().unwrap().get(path)?.clone()) - } -} - -#[async_trait::async_trait] -impl Handler for NeovimHandler { - type Writer = Compat; - - async fn handle_request( - &self, - name: String, - args: Vec, - nvim: Neovim>, - ) -> Result { - debug!("processing '{}' - {:?}", name, args); - match name.as_ref() { - "ping" => Ok(Value::from("pong")), - - "create" => { - if args.is_empty() { - return Err(Value::from("no path given")); - } - let path = default_empty_str(&args, 0); - let content = nullable_optional_str(&args, 1); - let mut c = self.client.clone(); - match c.create(path, content).await { - Ok(r) => match r { - true => Ok(Value::Nil), - false => Err(Value::from("rejected")), - }, - Err(e) => Err(Value::from(format!("could not create buffer: {}", e))), - } - }, - - "insert" => { - if args.len() < 3 { - return Err(Value::from("not enough arguments")); - } - let path = default_empty_str(&args, 0); - let txt = default_empty_str(&args, 1); - let mut pos = default_zero_number(&args, 2); - - if pos <= 0 { pos = 0 } // TODO wtf vim?? - - match self.buffer_controller(&path) { - None => Err(Value::from("no controller for given path")), - Some(controller) => { - controller.apply(controller.insert(&txt, pos as u64)).await; - Ok(Value::Nil) - }, - } - }, - - "delete" => { - if args.len() < 3 { - return Err(Value::from("not enough arguments")); - } - let path = default_empty_str(&args, 0); - let pos = default_zero_number(&args, 1) as u64; - let count = default_zero_number(&args, 2) as u64; - - match self.buffer_controller(&path) { - None => Err(Value::from("no controller for given path")), - Some(controller) => { - controller.apply(controller.delete(pos, count)).await; - Ok(Value::Nil) - } - } - }, - - "replace" => { - if args.len() < 2 { - return Err(Value::from("not enough arguments")); - } - let path = default_empty_str(&args, 0); - let txt = default_empty_str(&args, 1); - - match self.buffer_controller(&path) { - None => Err(Value::from("no controller for given path")), - Some(controller) => { - if let Some(op) = controller.replace(&txt) { - controller.apply(op).await; - } - Ok(Value::Nil) - } - } - }, - - "attach" => { - if args.is_empty() { - return Err(Value::from("no path given")); - } - let path = default_empty_str(&args, 0); - let buffer = match nvim.get_current_buf().await { - Ok(b) => b, - Err(e) => return Err(Value::from(format!("could not get current buffer: {}", e))), - }; - - let mut c = self.client.clone(); - - match c.attach(path.clone()).await { - Err(e) => Err(Value::from(format!("could not attach to stream: {}", e))), - Ok(controller) => { - let mut _controller = controller.clone(); - let lines : Vec = _controller.content().split('\n').map(|x| x.to_string()).collect(); - match buffer.set_lines(0, -1, false, lines).await { - Err(e) => Err(Value::from(format!("could not sync buffer: {}", e))), - Ok(()) => { - tokio::spawn(async move { - while let Some(_change) = _controller.poll().await { - let lines : Vec = _controller.content().split('\n').map(|x| x.to_string()).collect(); - // TODO only change lines affected! - if let Err(e) = buffer.set_lines(0, -1, false, lines).await { - error!("could not update buffer: {}", e); - } - } - }); - self.factories.lock().unwrap().insert(path, controller); - Ok(Value::Nil) - } - } - }, - } - }, - - "detach" => { - Err(Value::String("not implemented".into())) - // if args.is_empty() { - // return Err(Value::from("no path given")); - // } - // let path = default_empty_str(&args, 0); - // match self.buffer_controller(&path) { - // None => Err(Value::from("no controller for given path")), - // Some(controller) => Ok(Value::from(controller.stop())), - // } - }, - - "listen" => { - if args.is_empty() { - return Err(Value::from("no path given")); - } - let path = default_empty_str(&args, 0); - - let ns = nvim.create_namespace("Cursor").await - .map_err(|e| Value::from(format!("could not create namespace: {}", e)))?; - - let buf = nvim.get_current_buf().await - .map_err(|e| Value::from(format!("could not get current buf: {}", e)))?; - - let mut c = self.client.clone(); - match c.listen().await { - Err(e) => Err(Value::from(format!("could not listen cursors: {}", e))), - Ok(mut cursor) => { - self.cursors.lock().unwrap().insert(path, cursor.clone()); - debug!("spawning cursor processing worker"); - tokio::spawn(async move { - while let Some(cur) = cursor.poll().await { - if let Err(e) = buf.clear_namespace(ns, 0, -1).await { - error!("could not clear previous cursor highlight: {}", e); - } - let start = cur.start(); - let end = cur.end(); - let end_col = if start.row == end.row { - end.col - } else { - 0 // TODO what the fuck - }; - if let Err(e) = buf.add_highlight( - ns, "ErrorMsg", - start.row as i64 - 1, - start.col as i64, - end_col as i64 - ).await { - error!("could not create highlight for cursor: {}", e); - } - } - if let Err(e) = buf.clear_namespace(ns, 0, -1).await { - error!("could not clear previous cursor highlight: {}", e); - } - }); - Ok(Value::Nil) - }, - } - }, - - "cursor" => { - if args.len() < 3 { - return Err(Value::from("not enough args")); - } - let path = default_empty_str(&args, 0); - let row = default_zero_number(&args, 1) as i32; - let col = default_zero_number(&args, 2) as i32; - let row_end = default_zero_number(&args, 3) as i32; - let col_end = default_zero_number(&args, 4) as i32; - - match self.cursor_controller(&path) { - None => Err(Value::from("no path given")), - Some(cur) => { - cur.send(&path, (row, col).into(), (row_end, col_end).into()).await; - Ok(Value::Nil) - } - } - }, - - _ => Err(Value::from("unimplemented")), - } - } - - async fn handle_notify( - &self, - _name: String, - _args: Vec, - _nvim: Neovim>, - ) { - warn!("notify not handled"); - } -} - -#[derive(Parser, Debug)] -struct CliArgs { - /// server host to connect to - #[arg(long, default_value = "http://[::1]:50051")] - host: String, - - /// show debug level logs - #[arg(long, default_value_t = false)] - debug: bool, - - /// dump raw tracing logs into this TCP host - #[arg(long)] - remote_debug: Option, -} - - -#[tokio::main] -async fn main() -> Result<(), Box> { - let args = CliArgs::parse(); - - match args.remote_debug { - Some(host) => - tracing_subscriber::fmt() - .with_writer(Mutex::new(TcpStream::connect(host)?)) - .with_max_level(if args.debug { tracing::Level::DEBUG } else { tracing::Level::INFO }) - .init(), - - None => - tracing_subscriber::fmt() - .compact() - .without_time() - .with_ansi(false) - .with_writer(std::io::stderr) - .with_max_level(if args.debug { tracing::Level::DEBUG } else { tracing::Level::INFO }) - .init(), - } - - let client = BufferClient::connect(args.host.clone()).await?; - - let handler: NeovimHandler = NeovimHandler { - client: client.into(), - factories: Arc::new(Mutex::new(BTreeMap::new())), - cursors: Arc::new(Mutex::new(BTreeMap::new())), - }; - - let (_nvim, io_handler) = create::new_parent(handler).await; - - info!("++ codemp connected: {}", args.host); - - if let Err(e) = io_handler.await? { - error!("worker stopped with error: {}", e); - } - - Ok(()) -} diff --git a/client/vscode/Cargo.toml b/client/vscode/Cargo.toml deleted file mode 100644 index 457e394..0000000 --- a/client/vscode/Cargo.toml +++ /dev/null @@ -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"] } diff --git a/client/vscode/src/lib.rs b/client/vscode/src/lib.rs deleted file mode 100644 index 4c2794f..0000000 --- a/client/vscode/src/lib.rs +++ /dev/null @@ -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 = 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> { - 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::(cx, 0)?.value(cx) as i32, - arr.get::(cx, 1)?.value(cx) as i32, - )) -} - -struct ClientHandle(Arc>); -impl Finalize for ClientHandle {} - -fn connect(mut cx: FunctionContext) -> JsResult { - let host = cx.argument::(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::>(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 { - let path = cx.argument::(0)?.value(&mut cx); - let content = cx.argument::(1).ok().map(|x| x.value(&mut cx)); - let this = cx.this(); - let boxed : Handle> = 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::>(e.to_string())), - } - }); - - Ok(promise) -} - -fn listen_client(mut cx: FunctionContext) -> JsResult { - let this = cx.this(); - let boxed : Handle> = 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::>(e.to_string())), - } - }); - - Ok(promise) -} - -fn attach_client(mut cx: FunctionContext) -> JsResult { - let this = cx.this(); - let boxed : Handle> = this.get(&mut cx, "boxed")?; - let path = cx.argument::(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::>(e.to_string())), - } - }); - - Ok(promise) -} - - -struct OperationControllerJs(OperationControllerHandle); -impl Finalize for OperationControllerJs {} - -fn apply_operation(mut cx: FunctionContext) -> JsResult { - let this = cx.this(); - let boxed : Handle> = this.get(&mut cx, "boxed")?; - let skip = cx.argument::(0)?.value(&mut cx).round() as usize; - let text = cx.argument::(1)?.value(&mut cx); - let tail = cx.argument::(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 { - let this = cx.this(); - let boxed : Handle> = this.get(&mut cx, "boxed")?; - Ok(cx.string(boxed.0.content())) -} - -fn callback_operation(mut cx: FunctionContext) -> JsResult { - let this = cx.this(); - let boxed : Handle> = this.get(&mut cx, "boxed")?; - let callback = Arc::new(cx.argument::(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::(&mut cx)?; - Ok(()) - }); - } - }); - - Ok(cx.undefined()) -} - -struct CursorEventsHandle(CursorControllerHandle); -impl Finalize for CursorEventsHandle {} - -fn callback_cursor(mut cx: FunctionContext) -> JsResult { - let this = cx.this(); - let boxed : Handle> = this.get(&mut cx, "boxed")?; - let callback = Arc::new(cx.argument::(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::(&mut cx)?; - Ok(()) - }); - } - - }); - - Ok(cx.undefined()) -} - -fn send_cursor(mut cx: FunctionContext) -> JsResult { - let this = cx.this(); - let boxed : Handle> = this.get(&mut cx, "boxed")?; - let path = cx.argument::(0)?.value(&mut cx); - let start_obj = cx.argument::(1)?; - let start = unpack_tuple(&mut cx, start_obj)?; - let end_obj = cx.argument::(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(()) -} diff --git a/client/vscode/package.json b/package.json similarity index 100% rename from client/vscode/package.json rename to package.json diff --git a/proto/buffer.proto b/proto/buffer.proto deleted file mode 100644 index b668a44..0000000 --- a/proto/buffer.proto +++ /dev/null @@ -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; -} diff --git a/proto/cursor.proto b/proto/cursor.proto deleted file mode 100644 index 5af4fa2..0000000 --- a/proto/cursor.proto +++ /dev/null @@ -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; -} diff --git a/server/Cargo.toml b/server/Cargo.toml deleted file mode 100644 index 5fa0755..0000000 --- a/server/Cargo.toml +++ /dev/null @@ -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"] } diff --git a/server/src/buffer/actor.rs b/server/src/buffer/actor.rs deleted file mode 100644 index d7eea4e..0000000 --- a/server/src/buffer/actor.rs +++ /dev/null @@ -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 { - fn get(&self, key: &T) -> Option<&BufferHandle>; - fn put(&mut self, key: T, handle: BufferHandle) -> Option; - - fn handle(&mut self, key: T, content: Option) { - let handle = BufferHandle::new(content); - self.put(key, handle); - } -} - -#[derive(Clone)] -pub struct BufferHandle { - pub edit: mpsc::Sender<(oneshot::Sender, OperationRequest)>, - events: broadcast::Sender, - // pub digest: watch::Receiver, - pub content: watch::Receiver, -} - -impl BufferHandle { - fn new(init: Option) -> 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 { - self.events.subscribe() - } -} - -struct BufferWorker { - store: String, - edits: mpsc::Receiver<(oneshot::Sender, OperationRequest)>, - events: broadcast::Sender, - // digest: watch::Sender, - content: watch::Sender, -} - -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::(&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"); - }, - } - }, - } - } - } -} diff --git a/server/src/buffer/mod.rs b/server/src/buffer/mod.rs deleted file mode 100644 index 5c38de9..0000000 --- a/server/src/buffer/mod.rs +++ /dev/null @@ -1,2 +0,0 @@ -pub mod actor; -pub mod service; diff --git a/server/src/buffer/service.rs b/server/src/buffer/service.rs deleted file mode 100644 index 5745981..0000000 --- a/server/src/buffer/service.rs +++ /dev/null @@ -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> + Send>>; - -struct BufferMap { - store: HashMap, -} - -impl From::> for BufferMap { - fn from(value: HashMap) -> Self { - BufferMap { store: value } - } -} - -impl BufferStore for BufferMap { - fn get(&self, key: &String) -> Option<&BufferHandle> { - self.store.get(key) - } - fn put(&mut self, key: String, handle: BufferHandle) -> Option { - self.store.insert(key, handle) - } -} - -pub struct BufferService { - map: Arc>, -} - -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) -> Result, 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) -> Result, 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) -> Result, 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) -> Result, 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)) - } - } - } -} diff --git a/server/src/cursor/mod.rs b/server/src/cursor/mod.rs deleted file mode 100644 index 1f278a4..0000000 --- a/server/src/cursor/mod.rs +++ /dev/null @@ -1 +0,0 @@ -pub mod service; diff --git a/server/src/cursor/service.rs b/server/src/cursor/service.rs deleted file mode 100644 index b641c95..0000000 --- a/server/src/cursor/service.rs +++ /dev/null @@ -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> + Send>>; - -pub struct CursorService { - cursor: broadcast::Sender, -} - -#[tonic::async_trait] -impl Cursor for CursorService { - type ListenStream = CursorStream; - - async fn listen(&self, req: Request) -> Result, 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) -> Result, 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, - } - } -} diff --git a/server/src/main.rs b/server/src/main.rs deleted file mode 100644 index ec951e1..0000000 --- a/server/src/main.rs +++ /dev/null @@ -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> { - 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(()) -} diff --git a/src/buffer/controller.rs b/src/buffer/controller.rs deleted file mode 100644 index 73e967e..0000000 --- a/src/buffer/controller.rs +++ /dev/null @@ -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, - operations: mpsc::Sender, - stream: Mutex>, -} - -#[async_trait] -impl OperationFactory for BufferController { - fn content(&self) -> String { - self.content.borrow().clone() - } -} - -#[async_trait] -impl Controller for BufferController { - type Input = OperationSeq; - - async fn recv(&self) -> Result { - 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?) - } -} diff --git a/src/buffer/factory.rs b/src/buffer/factory.rs deleted file mode 100644 index 4c175e0..0000000 --- a/src/buffer/factory.rs +++ /dev/null @@ -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 { - 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 { - self.delta(0, txt, self.content().len()) - } - - fn delta(&self, start: usize, txt: &str, end: usize) -> Option { - 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 - } -} diff --git a/src/buffer/mod.rs b/src/buffer/mod.rs deleted file mode 100644 index 922e217..0000000 --- a/src/buffer/mod.rs +++ /dev/null @@ -1,11 +0,0 @@ -use std::ops::Range; - -pub(crate) mod worker; -pub mod controller; -pub mod factory; - - -pub struct TextChange { - pub span: Range, - pub content: String, -} diff --git a/src/buffer/worker.rs b/src/buffer/worker.rs deleted file mode 100644 index 43bb008..0000000 --- a/src/buffer/worker.rs +++ /dev/null @@ -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, - pub(crate) operations: mpsc::Receiver, - pub(crate) stream: Arc>, - pub(crate) queue: VecDeque, - receiver: watch::Receiver, - sender: mpsc::Sender, - 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 for BufferControllerWorker { - type Controller = BufferController; - type Tx = BufferClient; - type Rx = Streaming; - - 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, 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) -> Option { - 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 - } - } -} diff --git a/src/client.rs b/src/client.rs deleted file mode 100644 index 9691ee2..0000000 --- a/src/client.rs +++ /dev/null @@ -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, -} - -struct ServiceClients { - buffer: BufferClient, - cursor: CursorClient, -} - -struct Workspace { - cursor: Arc, - buffers: BTreeMap>, -} - - -impl CodempClient { - pub async fn new(dst: &str) -> Result { - 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> { - Some(self.workspace?.cursor.clone()) - } - - pub fn get_buffer(&self, path: &str) -> Option> { - self.workspace?.buffers.get(path).cloned() - } - - pub async fn join(&mut self, _session: &str) -> Result, 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, 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() }) - } - } -} diff --git a/src/cursor/controller.rs b/src/cursor/controller.rs deleted file mode 100644 index 7e879a7..0000000 --- a/src/cursor/controller.rs +++ /dev/null @@ -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, - stream: Mutex>, -} - -#[async_trait] -impl Controller 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 { - 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> { - // 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"))) - // } - // } - // } - // } -} diff --git a/src/cursor/mod.rs b/src/cursor/mod.rs deleted file mode 100644 index c6fbdd9..0000000 --- a/src/cursor/mod.rs +++ /dev/null @@ -1,26 +0,0 @@ -pub(crate) mod worker; -pub mod controller; - -use crate::proto::{RowCol, CursorPosition}; - -impl From:: 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()) - } -} diff --git a/src/cursor/worker.rs b/src/cursor/worker.rs deleted file mode 100644 index 42fa694..0000000 --- a/src/cursor/worker.rs +++ /dev/null @@ -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, - op: mpsc::Receiver, - channel: Arc>, -} - -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 for CursorControllerWorker { - type Controller = CursorController; - type Tx = CursorClient; - type Rx = Streaming; - - 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, - } - } - } -} - diff --git a/src/errors.rs b/src/errors.rs deleted file mode 100644 index 6b923a7..0000000 --- a/src/errors.rs +++ /dev/null @@ -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 IgnorableError for Result -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 for CodempError { - fn from(status: Status) -> Self { - CodempError::Transport { status: status.code(), message: status.message().to_string() } - } -} - -impl From for CodempError { - fn from(err: tonic::transport::Error) -> Self { - CodempError::Transport { - status: Code::Unknown, message: format!("underlying transport error: {:?}", err) - } - } -} - -impl From> for CodempError { - fn from(_value: mpsc::error::SendError) -> Self { - CodempError::Channel { send: true } - } -} - -impl From for CodempError { - fn from(_value: broadcast::error::RecvError) -> Self { - CodempError::Channel { send: false } - } -} diff --git a/client/vscode/src/extension.js b/src/extension.js similarity index 100% rename from client/vscode/src/extension.js rename to src/extension.js diff --git a/src/instance.rs b/src/instance.rs deleted file mode 100644 index 1421d29..0000000 --- a/src/instance.rs +++ /dev/null @@ -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>, -} - -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, 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, 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() }) - } -} diff --git a/src/lib.rs b/src/lib.rs index 0975600..4c2794f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,39 +1,270 @@ -pub mod cursor; -pub mod errors; -pub mod buffer; +use std::sync::Arc; -pub mod client; +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}; -#[cfg(feature = "static")] -pub mod instance; +fn runtime<'a, C: Context<'a>>(cx: &mut C) -> NeonResult<&'static Runtime> { + static RUNTIME: OnceCell = OnceCell::new(); -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"); + RUNTIME.get_or_try_init(|| { + Runtime::new() + .or_else(|err| cx.throw_error(err.to_string())) + }) } -pub use errors::CodempError; - -#[tonic::async_trait] // TODO move this somewhere? -pub(crate) trait ControllerWorker { - type Controller : Controller; - type Tx; - type Rx; - - fn subscribe(&self) -> Self::Controller; - async fn work(self, tx: Self::Tx, rx: Self::Rx); +fn tuple<'a, C: Context<'a>>(cx: &mut C, a: i32, b: i32) -> NeonResult> { + 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) } -#[tonic::async_trait] -pub trait Controller { - type Input; - - async fn send(&self, x: Self::Input) -> Result<(), CodempError>; - async fn recv(&self) -> Result; +fn unpack_tuple<'a, C: Context<'a>>(cx: &mut C, arr: Handle<'a, JsArray>) -> NeonResult<(i32, i32)> { + Ok(( + arr.get::(cx, 0)?.value(cx) as i32, + arr.get::(cx, 1)?.value(cx) as i32, + )) +} + +struct ClientHandle(Arc>); +impl Finalize for ClientHandle {} + +fn connect(mut cx: FunctionContext) -> JsResult { + let host = cx.argument::(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::>(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 { + let path = cx.argument::(0)?.value(&mut cx); + let content = cx.argument::(1).ok().map(|x| x.value(&mut cx)); + let this = cx.this(); + let boxed : Handle> = 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::>(e.to_string())), + } + }); + + Ok(promise) +} + +fn listen_client(mut cx: FunctionContext) -> JsResult { + let this = cx.this(); + let boxed : Handle> = 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::>(e.to_string())), + } + }); + + Ok(promise) +} + +fn attach_client(mut cx: FunctionContext) -> JsResult { + let this = cx.this(); + let boxed : Handle> = this.get(&mut cx, "boxed")?; + let path = cx.argument::(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::>(e.to_string())), + } + }); + + Ok(promise) +} + + +struct OperationControllerJs(OperationControllerHandle); +impl Finalize for OperationControllerJs {} + +fn apply_operation(mut cx: FunctionContext) -> JsResult { + let this = cx.this(); + let boxed : Handle> = this.get(&mut cx, "boxed")?; + let skip = cx.argument::(0)?.value(&mut cx).round() as usize; + let text = cx.argument::(1)?.value(&mut cx); + let tail = cx.argument::(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 { + let this = cx.this(); + let boxed : Handle> = this.get(&mut cx, "boxed")?; + Ok(cx.string(boxed.0.content())) +} + +fn callback_operation(mut cx: FunctionContext) -> JsResult { + let this = cx.this(); + let boxed : Handle> = this.get(&mut cx, "boxed")?; + let callback = Arc::new(cx.argument::(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::(&mut cx)?; + Ok(()) + }); + } + }); + + Ok(cx.undefined()) +} + +struct CursorEventsHandle(CursorControllerHandle); +impl Finalize for CursorEventsHandle {} + +fn callback_cursor(mut cx: FunctionContext) -> JsResult { + let this = cx.this(); + let boxed : Handle> = this.get(&mut cx, "boxed")?; + let callback = Arc::new(cx.argument::(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::(&mut cx)?; + Ok(()) + }); + } + + }); + + Ok(cx.undefined()) +} + +fn send_cursor(mut cx: FunctionContext) -> JsResult { + let this = cx.this(); + let boxed : Handle> = this.get(&mut cx, "boxed")?; + let path = cx.argument::(0)?.value(&mut cx); + let start_obj = cx.argument::(1)?; + let start = unpack_tuple(&mut cx, start_obj)?; + let end_obj = cx.argument::(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(()) }