hexedtech/codemp-vscode
3
0
Fork 0
mirror of https://github.com/hexedtech/codemp-vscode.git synced 2024-11-21 15:04:49 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

chore: removed rest of the project:

Browse source
This commit is contained in:
əlemi 2023-08-16 23:29:45 +02:00
parent 45a5667e5a
commit 27048efac9
30 changed files with 276 additions and 1937 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
client/vscode/.bundle.sh → .bundle.sh
View file

40
Cargo.toml
View file

@ -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"] }

5
build.rs
View file

@ -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(())
}

16
client/nvim/Cargo.toml
View file

@ -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"

190
client/nvim/codemp.lua
View file

@ -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', '<BS>', function() pcall(M.delete, path, cursor_offset(), 1) return '<BS>' end, {expr = true, buffer = buffer}) -- TODO log errors
vim.keymap.set('i', '<Del>', function() pcall(M.delete, path, cursor_offset() + 1, 1) return '<Del>' 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', '<BS>', { buffer = buffer })
vim.keymap.del('i', '<Del>', { 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

318
client/nvim/src/main.rs
View file

@ -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<Mutex<BTreeMap<String, OperationControllerHandle>>>,
cursors: Arc<Mutex<BTreeMap<String, CursorControllerHandle>>>,
}
fn nullable_optional_str(args: &[Value], index: usize) -> Option<String> {
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<i64> {
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<OperationControllerHandle> {
Some(self.factories.lock().unwrap().get(path)?.clone())
}
fn cursor_controller(&self, path: &String) -> Option<CursorControllerHandle> {
Some(self.cursors.lock().unwrap().get(path)?.clone())
}
}
#[async_trait::async_trait]
impl Handler for NeovimHandler {
type Writer = Compat<tokio::io::Stdout>;
async fn handle_request(
&self,
name: String,
args: Vec<Value>,
nvim: Neovim<Compat<tokio::io::Stdout>>,
) -> Result<Value, Value> {
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<String> = _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<String> = _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<Value>,
_nvim: Neovim<Compat<tokio::io::Stdout>>,
) {
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<String>,
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
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(())
}

22
client/vscode/Cargo.toml
View file

@ -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"] }

270
client/vscode/src/lib.rs
View file

@ -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(())
}

0
client/vscode/package.json → package.json
View file

35
proto/buffer.proto
View file

@ -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;
}

30
proto/cursor.proto
View file

@ -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;
}

19
server/Cargo.toml
View file

@ -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"] }

101
server/src/buffer/actor.rs
View file

@ -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");
},
}
},
}
}
}
}

2
server/src/buffer/mod.rs
View file

@ -1,2 +0,0 @@
pub mod actor;
pub mod service;

113
server/src/buffer/service.rs
View file

@ -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
server/src/cursor/mod.rs
View file

@ -1 +0,0 @@
pub mod service;

52
server/src/cursor/service.rs
View file

@ -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,
}
}
}

50
server/src/main.rs
View file

@ -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(())
}

41
src/buffer/controller.rs
View file

@ -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?)
}
}

84
src/buffer/factory.rs
View file

@ -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
}
}

11
src/buffer/mod.rs
View file

@ -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,
}

114
src/buffer/worker.rs
View file

@ -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
View file

@ -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() })
}
}
}

51
src/cursor/controller.rs
View file

@ -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")))
// }
// }
// }
// }
}

26
src/cursor/mod.rs
View file

@ -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())
}
}

54
src/cursor/worker.rs
View file

@ -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,
}
}
}
}

77
src/errors.rs
View file

@ -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 }
}
}

0
client/vscode/src/extension.js → src/extension.js
View file

81
src/instance.rs
View file

@ -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() })
}
}

293
src/lib.rs
View file

@ -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<Runtime> = 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<T> {
type Controller : Controller<T>;
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<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)
}
#[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>;
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(())
}