Merge branch 'pyo3_bump' into dev

Cargo.toml

@@ -40,8 +40,7 @@ napi = { version = "2.16", features = ["full"], optional = true }
 napi-derive = { version="2.16",  optional = true}
 
 # glue (python)
-pyo3 = { version = "0.20", features = ["extension-module"], optional = true}
+pyo3 = { version = "0.22", features = ["extension-module", "experimental-async"], optional = true}
-pyo3-asyncio = { version = "0.20", features = ["tokio-runtime"], optional = true }
 
 [build-dependencies]
 # glue (js)
@@ -55,4 +54,4 @@ rust = [] # used for ci matrix
 lua = ["mlua", "tracing-subscriber"]
 java = ["lazy_static", "jni", "tracing-subscriber"]
 js = ["napi-build", "tracing-subscriber", "napi", "napi-derive"]
-python = ["pyo3", "pyo3-asyncio", "tracing-subscriber", "pyo3-build-config"]
+python = ["pyo3", "tracing-subscriber", "pyo3-build-config"]

dist/py/build.sh

@@ -7,8 +7,16 @@ TARGET_EXT="$($PYO3_PYTHON -c 'import sysconfig; print(sysconfig.get_config_var(
 maturin build -i "$PYO3_PYTHON" --out "$WHEEL_DIR"
 
 CODEMPSUBLIME_DIR="../../../codemp-sublime/bindings/"
+CODEMPTEST_DIR="../../../codemp-python-test/"
 
 wheels=($WHEEL_DIR/*.whl)
 for wheel in $wheels; do
+	echo "moving $wheel to $CODEMPSUBLIME_DIR"
 	cp $wheel "$CODEMPSUBLIME_DIR"
+	cp $wheel "$CODEMPTEST_DIR"
 done
+
+cd "$CODEMPSUBLIME_DIR"
+source .venv/bin/activate
+pip install $wheel --force-reinstall
+

dist/py/codemp.pyi

@@ -1,65 +1,110 @@
-from typing import Tuple
+from typing import Tuple, Optional, Callable
 
-class PyLogger:
+class Driver:
-	def __init__(self, debug) -> None: ...
+	"""
-	async def listen(self) -> str | None: ...
+	this is akin to a big red button with a white "STOP" on top of it.
+	it is used to stop the runtime.
+	"""
+	def stop(self) -> None: ...
 
 
+def init(logger_cb: Callable, debug: bool) -> Driver: ...
+
+class Promise[T]:
+	"""
+	This is a class akin to a future, which wraps a join handle from a spawned
+	task on the rust side. you may call .pyawait() on this promise to block
+	until we have a result, or return immediately if we already have one.
+	This only goes one way rust -> python.
+
+	It can either be used directly or you can wrap it inside a future python side.
+	"""
+	def wait(self) -> T: ...
+	def is_done(self) -> bool: ...
+
 class TextChange:
+	"""
+	Editor agnostic representation of a text change, it translate between internal
+	codemp text operations and editor operations
+	"""
 	start: int
 	end: int
 	content: str
 
-	def is_deletion(self) -> bool: ...
+	def is_delete(self) -> bool: ...
-	def is_addition(self) -> bool: ...
+	def is_insert(self) -> bool: ...
 	def is_empty(self) -> bool: ...
 	def apply(self, txt: str) -> str: ...
-	def from_diff(self, before: str, after: str) -> TextChange: ...
-	def index_to_rowcol(self, txt: str, index: int) -> Tuple[int, int]: ...
 
 
 class BufferController:
-	def content(self) -> str: ...
+	"""
-	def send(self, start: int, end: int, txt: str) -> None: ...
+	Handle to the controller for a specific buffer, which manages the back and forth
-	async def try_recv(self) -> TextChange | None: ...
+	of operations to and from other peers.
-	async def recv(self) -> TextChange: ...
+	"""
-	async def poll(self) -> None: ...
+	def content(self) -> Promise[str]: ...
+	def send(self,
+		start: int,
+		end: int,
+		txt: str) -> Promise[None]: ...
+	def try_recv(self) -> Optional[TextChange]: ...
+	def recv(self) -> Promise[TextChange]: ...
+	def poll(self) -> Promise[None]: ...
+	def stop(self) -> bool: ...
 
 
 
 class Cursor:
+	"""
+	An Editor agnostic cursor position representation
+	"""
 	start: Tuple[int, int]
 	end: Tuple[int, int]
 	buffer: str
-	user: str # can be an empty string
+	user: Optional[str] # can be an empty string
 
 
 class CursorController:
-	def send(self, path: str, start: Tuple[int, int], end: Tuple[int, int]) -> None: ...
+	"""
-	def try_recv(self) -> Cursor | None: ...
+	Handle to the controller for a workspace, which manages the back and forth of
-	async def recv(self) -> Cursor: ...
+	cursor movements to and from other peers
-	async def poll(self) -> None: ...
+	"""
+	def send(self,
+		path: str,
+		start: Tuple[int, int],
+		end: Tuple[int, int]) -> Promise[None]: ...
+	def try_recv(self) -> Optional[Cursor]: ...
+	def recv(self) -> Promise[Cursor]: ...
+	def poll(self) -> Promise[None]: ...
 	def stop(self) -> bool: ...
 
 
 class Workspace:
-	async def create(self, path: str) -> None: ...
+	"""
-	async def attach(self, path: str) -> BufferController: ...
+	Handle to a workspace inside codemp. It manages buffers.
+	A cursor is tied to the single workspace.
+	"""
+	def create(self, path: str) -> Promise[None]: ...
+	def attach(self, path: str) -> Promise[BufferController]: ...
 	def detach(self, path: str) -> bool: ...
-	async def fetch_buffers(self) -> None: ...
+	def fetch_buffers(self) -> Promise[None]: ...
-	async def fetch_users(self) -> None: ...
+	def fetch_users(self) -> Promise[None]: ...
-	async def list_buffer_users(self, path: str) -> list[str]: ...
+	def list_buffer_users(self, path: str) -> Promise[list[str]]: ...
-	async def delete(self, path: str) -> None: ...
+	def delete(self, path: str) -> Promise[None]: ...
 	def id(self) -> str: ...
 	def cursor(self) -> CursorController: ...
-	def buffer_by_name(self, path: str) -> BufferController | None: ...
+	def buffer_by_name(self, path: str) -> Optional[BufferController]: ...
 	def buffer_list(self) -> list[str]: ...
-	def filetree(self) -> list[str]: ...
+	def filetree(self, filter: Optional[str]) -> list[str]: ...
 
 
 class Client:
-	def __init__(self, host: str, username: str, password: str) -> None: ...
+	"""
-	async def join_workspace(self, workspace: str) -> Workspace: ...
+	Handle to the actual client that manages the session. It manages the connection
+	to a server and joining/creating new workspaces
+	"""
+	def __new__(cls, host: str, username: str, password: str) -> None: ...
+	def join_workspace(self, workspace: str) -> Promise[Workspace]: ...
 	def leave_workspace(self, workspace: str) -> bool: ...
 	def get_workspace(self, id: str) -> Workspace: ...
 	def active_workspaces(self) -> list[str]: ...

src/api/change.rs

@@ -20,8 +20,7 @@
 ///
 #[derive(Clone, Debug, Default)]
 #[cfg_attr(feature = "js", napi_derive::napi(object))]
-#[cfg_attr(feature = "python", pyo3::pyclass)]
+#[cfg_attr(feature = "python", pyo3::pyclass(get_all))]
-#[cfg_attr(feature = "python", pyo3(get_all))]
 pub struct TextChange {
 	/// range start of text change, as char indexes in buffer previous state
 	pub start: u32,
@@ -37,7 +36,10 @@ impl TextChange {
 	pub fn span(&self) -> std::ops::Range<usize> {
 		self.start as usize..self.end as usize
 	}
+}
 
+#[cfg_attr(feature = "python", pyo3::pymethods)]
+impl TextChange {
 	/// returns true if this TextChange deletes existing text
 	pub fn is_delete(&self) -> bool {
 		self.start < self.end
@@ -55,9 +57,9 @@ impl TextChange {
 
 	/// applies this text change to given text, returning a new string
 	pub fn apply(&self, txt: &str) -> String {
-		let pre_index = std::cmp::min(self.span().start, txt.len());
+		let pre_index = std::cmp::min(self.start as usize, txt.len());
 		let pre = txt.get(..pre_index).unwrap_or("").to_string();
-		let post = txt.get(self.span().end..).unwrap_or("").to_string();
+		let post = txt.get(self.end as usize..).unwrap_or("").to_string();
 		format!("{}{}{}", pre, self.content, post)
 	}
 }
@@ -70,7 +72,7 @@ mod tests {
 			start: 5,
 			end: 5,
 			content: " cruel".to_string(),
-			hash: None
+			hash: None,
 		};
 		let result = change.apply("hello world!");
 		assert_eq!(result, "hello cruel world!");
@@ -82,7 +84,7 @@ mod tests {
 			start: 5,
 			end: 11,
 			content: "".to_string(),
-			hash: None
+			hash: None,
 		};
 		let result = change.apply("hello cruel world!");
 		assert_eq!(result, "hello world!");
@@ -94,7 +96,7 @@ mod tests {
 			start: 5,
 			end: 11,
 			content: " not very pleasant".to_string(),
-			hash: None
+			hash: None,
 		};
 		let result = change.apply("hello cruel world!");
 		assert_eq!(result, "hello not very pleasant world!");
@@ -106,7 +108,7 @@ mod tests {
 			start: 100,
 			end: 110,
 			content: "a very long string \n which totally matters".to_string(),
-			hash: None
+			hash: None,
 		};
 		let result = change.apply("a short text");
 		assert_eq!(
@@ -121,7 +123,7 @@ mod tests {
 			start: 42,
 			end: 42,
 			content: "".to_string(),
-			hash: None
+			hash: None,
 		};
 		let result = change.apply("some important text");
 		assert_eq!(result, "some important text");

src/api/event.rs

@@ -1,6 +1,7 @@
 use codemp_proto::workspace::workspace_event::Event as WorkspaceEventInner;
 
 #[derive(Debug, Clone)]
+#[cfg_attr(feature = "python", pyo3::pyclass)]
 pub enum Event {
 	FileTreeUpdated(String),
 	UserJoin(String),

src/buffer/controller.rs

@@ -5,7 +5,7 @@
 use std::sync::Arc;
 
 use diamond_types::LocalVersion;
-use tokio::sync::{oneshot, mpsc, watch};
+use tokio::sync::{mpsc, oneshot, watch};
 use tonic::async_trait;
 
 use crate::api::controller::ControllerCallback;
@@ -40,7 +40,9 @@ impl BufferController {
 		let (tx, rx) = oneshot::channel();
 		self.0.content_request.send(tx).await?;
 		let content = rx.await?;
-		self.0.last_update.set(self.0.latest_version.borrow().clone());
+		self.0
+			.last_update
+			.set(self.0.latest_version.borrow().clone());
 		Ok(content)
 	}
 }

src/buffer/worker.rs

@@ -108,7 +108,7 @@ impl ControllerWorker<TextChange> for BufferWorker {
 						let last_ver = oplog.local_version();
 
 						if change.is_delete() {
-							branch.delete_without_content(&mut oplog, 1, change.span());
+							branch.delete_without_content(&mut oplog, 1, change.start as usize..change.end as usize);
 						}
 
 						if change.is_insert() {

src/ffi/python/client.rs

@@ -2,28 +2,39 @@ use crate::workspace::Workspace;
 use crate::Client;
 use pyo3::prelude::*;
 
-// #[pyfunction]
+use super::tokio;
-// pub fn codemp_init<'a>(py: Python<'a>) -> PyResult<Py<Client>> {
-// 	Ok(Py::new(py, Client::default())?)
-// }
 
 #[pymethods]
 impl Client {
 	#[new]
-	fn pyconnect(host: String, username: String, password: String) -> PyResult<Self> {
+	fn __new__(host: String, username: String, password: String) -> crate::Result<Self> {
-		let cli =
+		tokio().block_on(Client::new(host, username, password))
-			pyo3_asyncio::tokio::get_runtime().block_on(Client::new(host, username, password));
-		Ok(cli?)
 	}
 
-	#[pyo3(name = "join_workspace")]
+	// #[pyo3(name = "join_workspace")]
-	fn pyjoin_workspace<'a>(&'a self, py: Python<'a>, workspace: String) -> PyResult<&PyAny> {
+	// async fn pyjoin_workspace(&self, workspace: String) -> JoinHandle<crate::Result<Workspace>> {
-		let rc = self.clone();
+	// 	tracing::info!("attempting to join the workspace {}", workspace);
 
-		pyo3_asyncio::tokio::future_into_py(py, async move {
+	// 	let this = self.clone();
-			let workspace: Workspace = rc.join_workspace(workspace.as_str()).await?;
+	// 	async {
-			Python::with_gil(|py| Py::new(py, workspace))
+	// 		tokio()
-		})
+	// 			.spawn(async move { this.join_workspace(workspace).await })
+	// 			.await
+	// 	}
+	// }
+
+	#[pyo3(name = "join_workspace")]
+	fn pyjoin_workspace(&self, py: Python<'_>, workspace: String) -> PyResult<super::Promise> {
+		tracing::info!("attempting to join the workspace {}", workspace);
+		let this = self.clone();
+		crate::a_sync_allow_threads!(py, this.join_workspace(workspace).await)
+		// let this = self.clone();
+		// Ok(super::Promise(Some(tokio().spawn(async move {
+		// 	Ok(this
+		// 		.join_workspace(workspace)
+		// 		.await
+		// 		.map(|f| Python::with_gil(|py| f.into_py(py)))?)
+		// }))))
 	}
 
 	#[pyo3(name = "leave_workspace")]
@@ -33,11 +44,8 @@ impl Client {
 
 	// join a workspace
 	#[pyo3(name = "get_workspace")]
-	fn pyget_workspace(&self, py: Python<'_>, id: String) -> PyResult<Option<Py<Workspace>>> {
+	fn pyget_workspace(&self, id: String) -> Option<Workspace> {
-		match self.get_workspace(id.as_str()) {
+		self.get_workspace(id.as_str())
-			Some(ws) => Ok(Some(Py::new(py, ws)?)),
-			None => Ok(None),
-		}
 	}
 
 	#[pyo3(name = "active_workspaces")]

src/ffi/python/controllers.rs

@@ -4,62 +4,105 @@ use crate::api::TextChange;
 use crate::buffer::Controller as BufferController;
 use crate::cursor::Controller as CursorController;
 use pyo3::prelude::*;
-use pyo3_asyncio::tokio::future_into_py;
 
-// use super::CodempController;
+use super::Promise;
+use crate::a_sync_allow_threads;
 
+// need to do manually since Controller is a trait implementation
 #[pymethods]
 impl CursorController {
 	#[pyo3(name = "send")]
-	pub fn pysend<'p>(
+	fn pysend(
 		&self,
-		py: Python<'p>,
+		py: Python,
 		path: String,
 		start: (i32, i32),
 		end: (i32, i32),
-	) -> PyResult<&'p PyAny> {
+	) -> PyResult<Promise> {
-		let rc = self.clone();
 		let pos = Cursor {
 			start,
 			end,
 			buffer: path,
 			user: None,
 		};
-		let rc = self.clone();
+		let this = self.clone();
-		future_into_py(py, async move { Ok(rc.send(pos).await?) })
+		a_sync_allow_threads!(py, this.send(pos).await)
 	}
 
 	#[pyo3(name = "try_recv")]
-	pub fn pytry_recv<'p>(&self, py: Python<'p>) -> PyResult<&'p PyAny> {
+	fn pytry_recv(&self, py: Python) -> PyResult<Promise> {
-		//PyResult<Option<Py<Cursor>>>
+		let this = self.clone();
-		let rc = self.clone();
+		a_sync_allow_threads!(py, this.try_recv().await)
-
-		future_into_py(py, async move { Ok(rc.try_recv().await?) })
 	}
 
 	#[pyo3(name = "recv")]
-	pub fn pyrecv<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
+	fn pyrecv(&self, py: Python) -> crate::Result<Option<Cursor>> {
-		let rc = self.clone();
+		py.allow_threads(|| super::tokio().block_on(self.try_recv()))
-
+		// let this = self.clone();
-		future_into_py(py, async move {
+		// a_sync_allow_threads!(py, this.recv().await)
-			let cur_event: Cursor = rc.recv().await?;
-			Python::with_gil(|py| Py::new(py, cur_event))
-		})
 	}
 
 	#[pyo3(name = "poll")]
-	pub fn pypoll<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
+	fn pypoll(&self, py: Python) -> PyResult<Promise> {
-		let rc = self.clone();
+		let this = self.clone();
-
+		a_sync_allow_threads!(py, this.poll().await)
-		future_into_py(py, async move { Ok(rc.poll().await?) })
 	}
 
 	#[pyo3(name = "stop")]
-	pub fn pystop(&self) -> bool {
+	fn pystop(&self) -> bool {
 		self.stop()
 	}
 }
 
+// need to do manually since Controller is a trait implementation
+#[pymethods]
+impl BufferController {
+	#[pyo3(name = "content")]
+	fn pycontent(&self, py: Python) -> PyResult<Promise> {
+		let this = self.clone();
+		a_sync_allow_threads!(py, this.content().await)
+	}
+
+	#[pyo3(name = "send")]
+	fn pysend(&self, py: Python, start: u32, end: u32, txt: String) -> PyResult<Promise> {
+		let op = TextChange {
+			start,
+			end,
+			content: txt,
+			hash: None,
+		};
+		let this = self.clone();
+		a_sync_allow_threads!(py, this.send(op).await)
+	}
+
+	#[pyo3(name = "try_recv")]
+	fn pytry_recv(&self, py: Python) -> crate::Result<Option<TextChange>> {
+		py.allow_threads(|| super::tokio().block_on(self.try_recv()))
+		// let this = self.clone();
+		// a_sync_allow_threads!(py, this.try_recv().await)
+	}
+
+	#[pyo3(name = "recv")]
+	fn pyrecv(&self, py: Python) -> PyResult<Promise> {
+		let this = self.clone();
+		a_sync_allow_threads!(py, this.recv().await)
+	}
+
+	#[pyo3(name = "poll")]
+	fn pypoll(&self, py: Python) -> PyResult<Promise> {
+		let this = self.clone();
+		a_sync_allow_threads!(py, this.poll().await)
+	}
+
+	#[pyo3(name = "stop")]
+	fn pystop(&self) -> bool {
+		self.stop()
+	}
+}
+
+// We have to write this manually since
+// cursor.user has type Option which cannot be translated
+// automatically
 #[pymethods]
 impl Cursor {
 	#[getter(start)]
@@ -78,85 +121,7 @@ impl Cursor {
 	}
 
 	#[getter(user)]
-	fn pyuser(&self) -> String {
+	fn pyuser(&self) -> Option<String> {
-		match self.user {
+		self.user.map(|user| user.to_string())
-			Some(user) => user.to_string(),
-			None => "".to_string(),
-		}
-	}
-}
-
-#[pymethods]
-impl BufferController {
-	#[pyo3(name = "content")]
-	fn pycontent<'p>(&self, py: Python<'p>) -> PyResult<&'p PyAny> {
-		let rc = self.clone();
-		future_into_py(py, async move { Ok(rc.content().await?) })
-	}
-
-	#[pyo3(name = "send")]
-	fn pysend<'p>(&self, py: Python<'p>, start: u32, end: u32, txt: String) -> PyResult<&'p PyAny> {
-		let op = TextChange {
-			start,
-			end,
-			content: txt,
-			hash: None,
-		};
-		let rc = self.clone();
-		future_into_py(py, async move { Ok(rc.send(op).await?) })
-	}
-
-	#[pyo3(name = "try_recv")]
-	fn pytry_recv<'p>(&self, py: Python<'p>) -> PyResult<&'p PyAny> {
-		// match self.try_recv()? {
-		// 	Some(txt_change) => {
-		// 		let evt = txt_change;
-		// 		Ok(evt.into_py(py))
-		// 	}
-		// 	None => Ok(py.None()),
-		// }
-		let rc = self.clone();
-
-		future_into_py(py, async move { Ok(rc.try_recv().await?) })
-	}
-
-	#[pyo3(name = "recv")]
-	fn pyrecv<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
-		let rc = self.clone();
-
-		future_into_py(py, async move {
-			let txt_change: TextChange = rc.recv().await?;
-			Python::with_gil(|py| Py::new(py, txt_change))
-		})
-	}
-
-	#[pyo3(name = "poll")]
-	fn pypoll<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
-		let rc = self.clone();
-
-		pyo3_asyncio::tokio::future_into_py(py, async move { Ok(rc.poll().await?) })
-	}
-}
-
-#[pymethods]
-impl TextChange {
-	#[pyo3(name = "is_deletion")]
-	fn pyis_deletion(&self) -> bool {
-		self.is_delete()
-	}
-
-	#[pyo3(name = "is_addition")]
-	fn pyis_addition(&self) -> bool {
-		self.is_insert()
-	}
-
-	#[pyo3(name = "is_empty")]
-	fn pyis_empty(&self) -> bool {
-		self.is_empty()
-	}
-
-	#[pyo3(name = "apply")]
-	fn pyapply(&self, txt: &str) -> String {
-		self.apply(txt)
 	}
 }

src/ffi/python/mod.rs

@@ -2,17 +2,186 @@ pub mod client;
 pub mod controllers;
 pub mod workspace;
 
-use std::sync::Arc;
-
 use crate::{
 	api::{Cursor, TextChange},
 	buffer::Controller as BufferController,
 	cursor::Controller as CursorController,
 	Client, Workspace,
 };
-use pyo3::exceptions::{PyConnectionError, PyRuntimeError, PySystemError};
+
 use pyo3::prelude::*;
-use tokio::sync::{mpsc, Mutex};
+use pyo3::{
+	exceptions::{PyConnectionError, PyRuntimeError, PySystemError},
+	types::PyFunction,
+};
+
+use std::sync::OnceLock;
+use tokio::sync::{mpsc, oneshot};
+
+// global reference to a current_thread tokio runtime
+pub fn tokio() -> &'static tokio::runtime::Runtime {
+	static RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
+	RT.get_or_init(|| {
+		tokio::runtime::Builder::new_current_thread()
+			.enable_all()
+			.build()
+			.unwrap()
+	})
+}
+
+// #[pyfunction]
+// fn register_event_loop(event_loop: PyObject) {
+// 	static EVENT_LOOP: OnceLock<PyObject> = OnceLock::new();
+// 	EVENT_LOOP.
+// }
+
+// #[pyfunction]
+// fn setup_async(
+// 	event_loop: PyObject,
+// 	call_soon_thread_safe: PyObject, // asyncio.EventLoop.call_soon_threadsafe
+// 	call_coroutine_thread_safe: PyObject, // asyncio.call_coroutine_threadsafe
+// 	create_future: PyObject,         // asyncio.EventLoop.create_future
+// ) {
+// 	let _ = EVENT_LOOP.get_or_init(|| event_loop);
+// 	let _ = CALL_SOON.get_or_init(|| call_soon_thread_safe);
+// 	let _ = CREATE_TASK.get_or_init(|| call_coroutine_thread_safe);
+// 	let _ = CREATE_FUTURE.get_or_init(|| create_future);
+// }
+
+#[pyclass]
+pub struct Promise(Option<tokio::task::JoinHandle<PyResult<PyObject>>>);
+
+#[pymethods]
+impl Promise {
+	#[pyo3(name = "wait")]
+	fn _await(&mut self, py: Python<'_>) -> PyResult<PyObject> {
+		py.allow_threads(move || match self.0.take() {
+			None => Err(PyRuntimeError::new_err(
+				"promise can't be awaited multiple times!",
+			)),
+			Some(x) => match tokio().block_on(x) {
+				Err(e) => Err(PyRuntimeError::new_err(format!(
+					"error awaiting promise: {e}"
+				))),
+				Ok(res) => res,
+			},
+		})
+	}
+
+	fn done(&self, py: Python<'_>) -> PyResult<bool> {
+		py.allow_threads(|| {
+			if let Some(handle) = &self.0 {
+				Ok(handle.is_finished())
+			} else {
+				Err(PyRuntimeError::new_err("promise was already awaited."))
+			}
+		})
+	}
+}
+
+#[macro_export]
+macro_rules! a_sync {
+	($x:expr) => {{
+		Ok($crate::ffi::python::Promise(Some(
+			$crate::ffi::python::tokio()
+				.spawn(async move { Ok($x.map(|f| Python::with_gil(|py| f.into_py(py)))?) }),
+		)))
+	}};
+}
+
+#[macro_export]
+macro_rules! a_sync_allow_threads {
+	($py:ident, $x:expr) => {{
+		$py.allow_threads(move || {
+			Ok($crate::ffi::python::Promise(Some(
+				$crate::ffi::python::tokio()
+					.spawn(async move { Ok($x.map(|f| Python::with_gil(|py| f.into_py(py)))?) }),
+			)))
+		})
+	}};
+}
+
+#[derive(Debug, Clone)]
+struct LoggerProducer(mpsc::UnboundedSender<String>);
+
+impl std::io::Write for LoggerProducer {
+	fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+		let _ = self.0.send(String::from_utf8_lossy(buf).to_string());
+		Ok(buf.len())
+	}
+
+	fn flush(&mut self) -> std::io::Result<()> {
+		Ok(())
+	}
+}
+
+#[pyclass]
+pub struct Driver(Option<oneshot::Sender<()>>);
+#[pymethods]
+impl Driver {
+	fn stop(&mut self) -> PyResult<()> {
+		match self.0.take() {
+			Some(tx) => {
+				let _ = tx.send(());
+				Ok(())
+			}
+			None => Err(PySystemError::new_err("Runtime was already stopped.")),
+		}
+	}
+}
+#[pyfunction]
+fn init(py: Python, logging_cb: Py<PyFunction>, debug: bool) -> PyResult<Driver> {
+	let (tx, mut rx) = mpsc::unbounded_channel();
+	let level = if debug {
+		tracing::Level::DEBUG
+	} else {
+		tracing::Level::INFO
+	};
+
+	let format = tracing_subscriber::fmt::format()
+		.without_time()
+		.with_level(true)
+		.with_target(true)
+		.with_thread_ids(false)
+		.with_thread_names(false)
+		.with_file(false)
+		.with_line_number(false)
+		.with_source_location(false)
+		.compact();
+
+	let log_subscribing = tracing_subscriber::fmt()
+		.with_ansi(false)
+		.event_format(format)
+		.with_max_level(level)
+		.with_writer(std::sync::Mutex::new(LoggerProducer(tx)))
+		.try_init();
+
+	let (rt_stop_tx, mut rt_stop_rx) = oneshot::channel::<()>();
+
+	match log_subscribing {
+		Ok(_) => {
+			// the runtime is driven by the logger awaiting messages from codemp and echoing them back to
+			// a provided logger.
+			py.allow_threads(move || {
+				std::thread::spawn(move || {
+					tokio().block_on(async move {
+						loop {
+							tokio::select! {
+								biased;
+								Some(msg) = rx.recv() => {
+									let _ = Python::with_gil(|py| logging_cb.call1(py, (msg,)));
+								},
+								_ = &mut rt_stop_rx => { break }, // a bit brutal but will do for now.
+							}
+						}
+					})
+				})
+			});
+			Ok(Driver(Some(rt_stop_tx)))
+		}
+		Err(_) => Err(PyRuntimeError::new_err("codemp was already initialised.")),
+	}
+}
 
 impl From<crate::Error> for PyErr {
 	fn from(value: crate::Error) -> Self {
@@ -31,66 +200,16 @@ impl From<crate::Error> for PyErr {
 	}
 }
 
-#[derive(Debug, Clone)]
+impl IntoPy<PyObject> for crate::api::User {
-struct LoggerProducer(mpsc::Sender<String>);
+	fn into_py(self, py: Python<'_>) -> PyObject {
-
+		self.id.to_string().into_py(py)
-impl std::io::Write for LoggerProducer {
-	fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-		let _ = self.0.try_send(String::from_utf8_lossy(buf).to_string()); // ignore: logger disconnected or with full buffer
-		Ok(buf.len())
-	}
-
-	fn flush(&mut self) -> std::io::Result<()> {
-		Ok(())
-	}
-}
-
-#[pyclass]
-struct PyLogger(Arc<Mutex<mpsc::Receiver<String>>>);
-
-#[pymethods]
-impl PyLogger {
-	#[new]
-	fn init_logger(debug: bool) -> PyResult<Self> {
-		let (tx, rx) = mpsc::channel(256);
-		let level = if debug {
-			tracing::Level::DEBUG
-		} else {
-			tracing::Level::INFO
-		};
-
-		let format = tracing_subscriber::fmt::format()
-			.without_time()
-			.with_level(true)
-			.with_target(true)
-			.with_thread_ids(false)
-			.with_thread_names(false)
-			.with_file(false)
-			.with_line_number(false)
-			.with_source_location(false)
-			.compact();
-
-		match tracing_subscriber::fmt()
-			.with_ansi(false)
-			.event_format(format)
-			.with_max_level(level)
-			.with_writer(std::sync::Mutex::new(LoggerProducer(tx)))
-			.try_init()
-		{
-			Ok(_) => Ok(PyLogger(Arc::new(Mutex::new(rx)))),
-			Err(_) => Err(PySystemError::new_err("A logger already exists")),
-		}
-	}
-
-	fn listen<'p>(&'p self, py: Python<'p>) -> PyResult<&'p PyAny> {
-		let rc = self.0.clone();
-		pyo3_asyncio::tokio::future_into_py(py, async move { Ok(rc.lock().await.recv().await) })
 	}
 }
 
 #[pymodule]
-fn codemp(_py: Python, m: &PyModule) -> PyResult<()> {
+fn codemp(m: &Bound<'_, PyModule>) -> PyResult<()> {
-	m.add_class::<PyLogger>()?;
+	m.add_function(wrap_pyfunction!(init, m)?)?;
+	m.add_class::<Driver>()?;
 
 	m.add_class::<TextChange>()?;
 	m.add_class::<BufferController>()?;

src/ffi/python/workspace.rs

@@ -2,30 +2,23 @@ use crate::buffer::Controller as BufferController;
 use crate::cursor::Controller as CursorController;
 use crate::workspace::Workspace;
 use pyo3::prelude::*;
-use pyo3::types::PyString;
+
-use pyo3_asyncio::generic::future_into_py;
+use super::Promise;
+use crate::a_sync_allow_threads;
 
 #[pymethods]
 impl Workspace {
 	// join a workspace
 	#[pyo3(name = "create")]
-	fn pycreate<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&'p PyAny> {
+	fn pycreate(&self, py: Python, path: String) -> PyResult<Promise> {
-		let ws = self.clone();
+		let this = self.clone();
-
+		a_sync_allow_threads!(py, this.create(path.as_str()).await)
-		pyo3_asyncio::tokio::future_into_py(py, async move {
-			ws.create(path.as_str()).await?;
-			Ok(())
-		})
 	}
-	#[pyo3(name = "attach")]
-	fn pyattach<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&PyAny> {
-		let ws = self.clone();
 
-		pyo3_asyncio::tokio::future_into_py(py, async move {
+	#[pyo3(name = "attach")]
-			let buffctl: BufferController = ws.attach(path.as_str()).await?;
+	fn pyattach(&self, py: Python, path: String) -> PyResult<Promise> {
-			Ok(buffctl)
+		let this = self.clone();
-			// Python::with_gil(|py| Py::new(py, buffctl))
+		a_sync_allow_threads!(py, this.attach(path.as_str()).await)
-		})
 	}
 
 	#[pyo3(name = "detach")]
@@ -37,79 +30,50 @@ impl Workspace {
 		}
 	}
 
-	// #[pyo3(name = "event")]
+	#[pyo3(name = "event")]
-	// fn pyevent(&self, py: Python<'_>, path: String) -> PyResult<&PyAny> {
+	fn pyevent(&self, py: Python) -> PyResult<Promise> {
-	// 	let rc = self.clone();
+		let this = self.clone();
-	// 	future_into_py(py, async move { Ok(rc.event().await?) })
+		a_sync_allow_threads!(py, this.event().await)
-	// }
+	}
 
 	#[pyo3(name = "fetch_buffers")]
-	fn pyfetch_buffers<'p>(&'p self, py: Python<'p>) -> PyResult<&PyAny> {
+	fn pyfetch_buffers(&self, py: Python) -> PyResult<Promise> {
-		let ws = self.clone();
+		let this = self.clone();
-
+		a_sync_allow_threads!(py, this.fetch_buffers().await)
-		pyo3_asyncio::tokio::future_into_py(py, async move {
-			ws.fetch_buffers().await?;
-			Ok(())
-		})
 	}
 
 	#[pyo3(name = "fetch_users")]
-	fn pyfetch_users<'p>(&'p self, py: Python<'p>) -> PyResult<&PyAny> {
+	fn pyfetch_users(&self, py: Python) -> PyResult<Promise> {
-		let ws = self.clone();
+		let this = self.clone();
-
+		a_sync_allow_threads!(py, this.fetch_users().await)
-		pyo3_asyncio::tokio::future_into_py(py, async move {
-			ws.fetch_users().await?;
-			Ok(())
-		})
 	}
 
 	#[pyo3(name = "list_buffer_users")]
-	fn pylist_buffer_users<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&PyAny> {
+	fn pylist_buffer_users(&self, py: Python, path: String) -> PyResult<Promise> {
-		let ws = self.clone();
+		// crate::Result<Vec<crate::api::User>>
-
+		let this = self.clone();
-		pyo3_asyncio::tokio::future_into_py(py, async move {
+		a_sync_allow_threads!(py, this.list_buffer_users(path.as_str()).await)
-			let usrlist: Vec<String> = ws
-				.list_buffer_users(path.as_str())
-				.await?
-				.into_iter()
-				.map(|e| e.id.to_string())
-				.collect();
-
-			Ok(usrlist)
-		})
 	}
 
 	#[pyo3(name = "delete")]
-	fn pydelete<'p>(&'p self, py: Python<'p>, path: String) -> PyResult<&PyAny> {
+	fn pydelete(&self, py: Python, path: String) -> PyResult<Promise> {
-		let ws = self.clone();
+		let this = self.clone();
-
+		a_sync_allow_threads!(py, this.delete(path.as_str()).await)
-		pyo3_asyncio::tokio::future_into_py(py, async move {
-			ws.delete(path.as_str()).await?;
-			Ok(())
-		})
 	}
 
 	#[pyo3(name = "id")]
-	fn pyid(&self, py: Python<'_>) -> Py<PyString> {
+	fn pyid(&self) -> String {
-		PyString::new(py, self.id().as_str()).into()
+		self.id()
 	}
 
 	#[pyo3(name = "cursor")]
-	fn pycursor(&self, py: Python<'_>) -> PyResult<Py<CursorController>> {
+	fn pycursor(&self) -> CursorController {
-		Py::new(py, self.cursor())
+		self.cursor()
 	}
 
 	#[pyo3(name = "buffer_by_name")]
-	fn pybuffer_by_name(
+	fn pybuffer_by_name(&self, path: String) -> Option<BufferController> {
-		&self,
+		self.buffer_by_name(path.as_str())
-		py: Python<'_>,
-		path: String,
-	) -> PyResult<Option<Py<BufferController>>> {
-		let Some(bufctl) = self.buffer_by_name(path.as_str()) else {
-			return Ok(None);
-		};
-
-		Ok(Some(Py::new(py, bufctl)?))
 	}
 
 	#[pyo3(name = "buffer_list")]
@@ -118,14 +82,8 @@ impl Workspace {
 	}
 
 	#[pyo3(name = "filetree")]
+	#[pyo3(signature = (filter=None))]
 	fn pyfiletree(&self, filter: Option<&str>) -> Vec<String> {
 		self.filetree(filter)
 	}
 }
-
-// #[pyclass]
-// enum PyEvent {
-// 	FileTreeUpdated,
-// 	UserJoin { name: String },
-// 	UserLeave { name: String },
-// }

src/workspace/worker.rs

@@ -140,7 +140,9 @@ impl Workspace {
 			}
 		});
 	}
+}
 
+impl Workspace {
 	/// create a new buffer in current workspace
 	pub async fn create(&self, path: &str) -> crate::Result<()> {
 		let mut workspace_client = self.0.services.ws();
@@ -357,6 +359,8 @@ impl Drop for WorkspaceInner {
 	}
 }
 
+#[cfg_attr(feature = "python", pyo3::pyclass(eq, eq_int))]
+#[cfg_attr(feature = "python", derive(PartialEq))]
 pub enum DetachResult {
 	NotAttached,
 	Detaching,