feat: add wasm-bindgen-futures as an optional runtime for AsyncDB, target wasm32-unknown-unknown support #49
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,151 @@ | ||
| use std::collections::HashMap; | ||
| use std::path::Path; | ||
|
|
||
| use async_std::channel::{self, TryRecvError}; | ||
| use wasm_bindgen_futures::spawn_local; | ||
|
|
||
| use crate::asyncdb::{ReceiverExt, Request, Response, CHANNEL_BUFFER_SIZE}; | ||
| use crate::snapshot::Snapshot; | ||
| use crate::{Options, Result, Status, StatusCode, DB}; | ||
|
|
||
| pub(crate) struct Message { | ||
| pub(crate) req: Request, | ||
| pub(crate) resp_channel: channel::Sender<Response>, | ||
| } | ||
|
|
||
| /// `AsyncDB` makes it easy to use LevelDB in a async-std runtime. | ||
| /// The methods follow very closely the main API (see `DB` type). Iteration is not yet implemented. | ||
| #[derive(Clone)] | ||
| pub struct AsyncDB { | ||
| shutdown: channel::Sender<()>, | ||
| send: channel::Sender<Message>, | ||
| } | ||
|
|
||
| impl AsyncDB { | ||
| /// Create a new or open an existing database. | ||
| pub fn new<P: AsRef<Path>>(name: P, opts: Options) -> Result<AsyncDB> { | ||
| let db = DB::open(name, opts)?; | ||
|
|
||
| let (send, recv) = channel::bounded(CHANNEL_BUFFER_SIZE); | ||
| let (shutdown, shutdown_recv) = channel::bounded(1); | ||
|
|
||
| spawn_local(async move { | ||
| AsyncDB::run_server_async(db, recv, shutdown_recv, HashMap::new(), 0).await; | ||
| }); | ||
|
|
||
| Ok(AsyncDB { shutdown, send }) | ||
| } | ||
|
|
||
| pub(crate) async fn process_request(&self, req: Request) -> Result<Response> { | ||
| let (tx, rx) = channel::bounded(1); | ||
|
|
||
| let m = Message { | ||
| req, | ||
| resp_channel: tx, | ||
| }; | ||
| if let Err(e) = self.send.send(m).await { | ||
| return Err(Status { | ||
| code: StatusCode::AsyncError, | ||
| err: e.to_string(), | ||
| }); | ||
| } | ||
| let resp = rx.recv().await; | ||
| match resp { | ||
| Err(e) => Err(Status { | ||
| code: StatusCode::AsyncError, | ||
| err: e.to_string(), | ||
| }), | ||
| Ok(r) => Ok(r), | ||
| } | ||
| } | ||
|
|
||
| pub(crate) async fn run_server_async( | ||
| mut db: DB, | ||
| mut recv: impl ReceiverExt<Message> + Clone + 'static, | ||
| mut shutdown: impl ReceiverExt<()> + Clone + 'static, | ||
| mut snapshots: HashMap<usize, Snapshot>, | ||
| mut snapshot_counter: usize, | ||
| ) { | ||
| if let Some(message) = recv.recv().await { | ||
|
There was a problem hiding this comment. what's the plan/idea behind this? For Why then implement the background thread using asynchronous I/O? If
From what I can see here it look like we're dealing with case 1, that's how What am I missing? There was a problem hiding this comment. Yeah, it's a bit weird to be honest (and even weirder when you try an implement an Env that will work in WASM/the browser). My goal was to get this to work with wasm-bindgen-futures, so that it can run in a WASM environment in the browser. Right now, that would mean in-memory only, although I was starting to poke at what a WASM-based Env would look like to run it e.g. in a browser. The JS runtime is single-threaded, but uses Promises for asynchronous scheduling, so that the browser doesn't block on waiting for something. Though, working through the Env trait, I think it might be more challenging to implement (e.g. how do we sleep in WASM when we don't have time available), so any wasm32-unknown-unknown target would be in-memory only. There was a problem hiding this comment. (and ultimately it would be backed by IndexedDB, which is LevelDB at it's core so it's a bit of a :spider-man-pointing: situation -- maybe it's better to build what I'm building in IndexedDB directly for WASM and use this for the non-WASM Rust stuff) There was a problem hiding this comment. I agree with your last point: unless the IndexedDB API is so restrictive as to be useless (I don't know it myself), it is probably best to use it directly, and instead abstract LevelDB under a common database interface (instead of abstracting storage below LevelDB). It might save you some time and sweat. |
||
| Self::match_message( | ||
| &mut db, | ||
| recv.clone(), | ||
| &mut snapshots, | ||
| &mut snapshot_counter, | ||
| message, | ||
| ); | ||
| } | ||
|
|
||
| spawn_local(async move { | ||
| // check shutdown | ||
| if let Some(()) = shutdown.recv().await { | ||
| return; | ||
| } else { | ||
| AsyncDB::run_server_async(db, recv, shutdown, snapshots, snapshot_counter).await | ||
| }; | ||
| }); | ||
| } | ||
|
|
||
| pub(crate) async fn stop_server_async(&self) { | ||
| self.shutdown.close(); | ||
| } | ||
| } | ||
|
|
||
| pub(crate) fn send_response_result(ch: channel::Sender<Response>, result: Result<()>) { | ||
| if let Err(e) = result { | ||
| ch.try_send(Response::Error(e)).ok(); | ||
| } else { | ||
| ch.try_send(Response::OK).ok(); | ||
| } | ||
| } | ||
|
|
||
| pub(crate) fn send_response(ch: channel::Sender<Response>, res: Response) { | ||
| ch.send_blocking(res).ok(); | ||
| } | ||
|
|
||
| impl ReceiverExt<Message> for channel::Receiver<Message> { | ||
| fn blocking_recv(&mut self) -> Option<Message> { | ||
| self.recv_blocking().ok() | ||
| } | ||
|
|
||
| fn close(&mut self) { | ||
| channel::Receiver::close(self); | ||
| } | ||
|
|
||
| async fn recv(&mut self) -> Option<Message> { | ||
| channel::Receiver::recv(&self).await.ok() | ||
| } | ||
| } | ||
|
|
||
| impl ReceiverExt<()> for channel::Receiver<()> { | ||
| fn blocking_recv(&mut self) -> Option<()> { | ||
| self.recv_blocking().ok() | ||
| } | ||
|
|
||
| fn close(&mut self) { | ||
| channel::Receiver::close(self); | ||
| } | ||
|
|
||
| async fn recv(&mut self) -> Option<()> { | ||
| match channel::Receiver::try_recv(&self) { | ||
| Ok(_) => Some(()), | ||
| Err(TryRecvError::Empty) => None, | ||
| Err(TryRecvError::Closed) => Some(()), | ||
| } | ||
| } | ||
| } | ||
|
|
||
| #[cfg(test)] | ||
| pub mod tests { | ||
| use crate::{in_memory, AsyncDB}; | ||
| use wasm_bindgen_test::wasm_bindgen_test; | ||
|
|
||
| #[wasm_bindgen_test] | ||
| async fn test_asyncdb() { | ||
| let db = AsyncDB::new("test.db", in_memory()).unwrap(); | ||
| db.put(b"key".to_vec(), b"value".to_vec()).await.unwrap(); | ||
| let val = db.get(b"key".to_vec()).await.unwrap(); | ||
| assert_eq!(val, Some(b"value".to_vec())); | ||
| db.stop_server_async().await; | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
can you refactor
run_server()to usematch_message()?