Update async-sqlite

kjvalencik · kjvalencik · commit 105045c3bd99 · 2025-11-24T17:01:21.000-05:00
diff --git a/examples/async-sqlite/README.md b/examples/async-sqlite/README.md
@@ -5,7 +5,7 @@ The Async SQLite example implements a simple database in Rust with a JavaScript
 ## Usage
 
 ```js
-const Database = require(".");
+const { Database } = require(".");
 
 (async () => {
     const db = new Database();
@@ -31,25 +31,18 @@ Since SQLite is naturally single threaded, our application does not benefit from
 
 Once the database thread is spawned, the JavaScript main thread needs a way to communicate with it. A [multi-producer, single-consumer (mpsc)][mpsc] channel is created. The receiving end is owned by the database thread and the sender is held by JavaScript.
 
-#### `JsBox`
+#### `#[neon::export(class)]`
 
-Rust data cannot be directly held by JavaScript. The [`JsBox`][jsbox] provides a mechanism for allowing JavaScript to hold a reference to Rust data and later access it again from Rust.
+The Rust sender side of the channel is held in a JavaScript [class][class]. It can be referenced later in methods.  
 
 #### Rust and Neon Channels
 
 The mpsc channel provides a way for the JavaScript main thread to communicate with the database thread, but it is one-way. In order to complete the callback, the database thread must be able to communicate with the JavaScript main thread. [`neon::event::Channel`][channel] provides a channel for sending these events back.
 
-#### `Root`
-
-The last issue to solve is sending a reference to the JavaScript callback to the database thread and back again before finally calling it. [Handles][handle] to JavaScript values are not `Send`; they cannot escape the scope that created them. The reason they cannot be passed to other threads is because when control is returned back to the JavaScript engine, the garbage collector may determine they are no longer used and free the value.
-
-A [`Root`][root] is a special handle to a JavaScript value that prevents the value from being freed as long as the `Root` has not been dropped. By placing the callback in a `Root`, it can be safely sent across threads and finally accessed and called when back on the JavaScript main thread.
-
 ### JavaScript
 
 [thread]: https://doc.rust-lang.org/std/thread/
 [mpsc]: https://doc.rust-lang.org/std/sync/mpsc/index.html
-[jsbox]: https://docs.rs/neon/latest/neon/types/struct.JsBox.html
+[class]: https://docs.rs/neon/latest/neon/attr.class.html
 [channel]: https://docs.rs/neon/latest/neon/event/struct.Channel.html
 [handle]: https://docs.rs/neon/latest/neon/handle/struct.Handle.html
-[root]: https://docs.rs/neon/latest/neon/handle/struct.Root.html
diff --git a/examples/async-sqlite/index.js b/examples/async-sqlite/index.js
diff --git a/examples/async-sqlite/package.json b/examples/async-sqlite/package.json
@@ -2,7 +2,7 @@
   "name": "async-sqlite",
   "version": "0.1.0",
   "description": "Neon Async SQLite",
-  "main": "index.js",
+  "main": "index.node",
   "scripts": {
     "build": "cargo-cp-artifact -nc index.node -- cargo build --message-format=json-render-diagnostics",
     "install": "npm run build",
diff --git a/examples/async-sqlite/src/lib.rs b/examples/async-sqlite/src/lib.rs
@@ -1,44 +1,70 @@
-use std::sync::mpsc;
-use std::thread;
+use std::{ops::ControlFlow, sync::mpsc, thread};
+
+use neon::{
+    prelude::*,
+    types::Deferred,
+    types::extract::{Error, TryIntoJs},
+};
 
-use neon::{prelude::*, types::Deferred};
 use rusqlite::Connection;
 
-type DbCallback = Box<dyn FnOnce(&mut Connection, &Channel, Deferred) + Send>;
+type DbCallback =
+    Box<dyn FnOnce(&mut Connection, &Channel, Deferred) -> ControlFlow<(), ()> + Send>;
 
 // Wraps a SQLite connection a channel, allowing concurrent access
 struct Database {
-    tx: mpsc::Sender<DbMessage>,
-}
-
-// Messages sent on the database channel
-enum DbMessage {
-    // Promise to resolve and callback to be executed
-    // Deferred is threaded through the message instead of moved to the closure so that it
-    // can be manually rejected.
-    Callback(Deferred, DbCallback),
-    // Indicates that the thread should be stopped and connection closed
-    Close,
+    tx: mpsc::Sender<(Deferred, DbCallback)>,
 }
 
-// Clean-up when Database is garbage collected, could go here
-// but, it's not needed,
-impl Finalize for Database {}
-
 // Internal implementation
 impl Database {
-    // Creates a new instance of `Database`
-    //
-    // 1. Creates a connection and a channel
-    // 2. Spawns a thread and moves the channel receiver and connection to it
-    // 3. On a separate thread, read closures off the channel and execute with access
-    //    to the connection.
-    fn new<'a, C>(cx: &mut C) -> rusqlite::Result<Self>
+    fn exec<'cx, O, V, F>(&self, cx: &mut Cx<'cx>, f: F) -> JsResult<'cx, JsPromise>
     where
-        C: Context<'a>,
+        F: FnOnce(&mut Connection) -> O + Send + 'static,
+        for<'a> O: TryIntoJs<'a, Value = V> + Send + 'static,
+        V: Value,
     {
+        self.send(cx, |conn| ControlFlow::Continue(f(conn)))
+    }
+
+    fn send<'cx, O, V, F>(&self, cx: &mut Cx<'cx>, f: F) -> JsResult<'cx, JsPromise>
+    where
+        F: FnOnce(&mut Connection) -> ControlFlow<(), O> + Send + 'static,
+        for<'a> O: TryIntoJs<'a, Value = V> + Send + 'static,
+        V: Value,
+    {
+        let (d, promise) = cx.promise();
+        let Err(mpsc::SendError((d, _))) = self.tx.send((
+            d,
+            Box::new(move |conn, ch, d| {
+                let output = f(conn).continue_value();
+                let should_break = output.is_none();
+                let res = d.try_settle_with(ch, move |mut cx| output.try_into_js(&mut cx));
+
+                // If we can no longer settle promises, we can stop the worker thread
+                if res.is_err() || should_break {
+                    ControlFlow::Break(())
+                } else {
+                    ControlFlow::Continue(())
+                }
+            }),
+        )) else {
+            return Ok(promise);
+        };
+
+        let err = cx.error("Database is closed")?;
+        d.reject(cx, err);
+
+        Ok(promise)
+    }
+}
+
+#[neon::export(class)]
+// JavaScript class
+impl Database {
+    fn new(cx: &mut Cx) -> Result<Self, Error> {
         // Channel for sending callbacks to execute on the sqlite connection thread
-        let (tx, rx) = mpsc::channel::<DbMessage>();
+        let (tx, rx) = mpsc::channel();
 
         // Open a connection sqlite, this will be moved to the thread
         let mut conn = Connection::open_in_memory()?;
@@ -48,6 +74,7 @@ impl Database {
         // The JavaScript process will not exit as long as this channel has not been
         // dropped.
         let channel = cx.channel();
+        let db = Self { tx };
 
         // Create a table in the in-memory database
         // In production code, this would likely be handled somewhere else
@@ -69,155 +96,48 @@ impl Database {
             // When the instance of `Database` is dropped, the channel will be closed
             // and `rx.recv()` will return an `Err`, ending the loop and terminating
             // the thread.
-            while let Ok(message) = rx.recv() {
-                match message {
-                    DbMessage::Callback(deferred, f) => {
-                        // The connection and channel are owned by the thread, but _lent_ to
-                        // the callback. The callback has exclusive access to the connection
-                        // for the duration of the callback.
-                        f(&mut conn, &channel, deferred);
-                    }
-                    // Immediately close the connection, even if there are pending messages
-                    DbMessage::Close => break,
+            while let Ok((d, callback)) = rx.recv() {
+                // Returning `true` means to _stop_
+                if callback(&mut conn, &channel, d).is_break() {
+                    break;
                 }
             }
         });
 
-        Ok(Self { tx })
-    }
-
-    // Idiomatic rust would take an owned `self` to prevent use after close
-    // However, it's not possible to prevent JavaScript from continuing to hold a closed database
-    fn close(&self) -> Result<(), mpsc::SendError<DbMessage>> {
-        self.tx.send(DbMessage::Close)
-    }
-
-    fn send(
-        &self,
-        deferred: Deferred,
-        callback: impl FnOnce(&mut Connection, &Channel, Deferred) + Send + 'static,
-    ) -> Result<(), mpsc::SendError<DbMessage>> {
-        self.tx
-            .send(DbMessage::Callback(deferred, Box::new(callback)))
-    }
-}
-
-// Methods exposed to JavaScript
-// The `JsBox` boxed `Database` is expected as the `this` value on all methods except `js_new`
-impl Database {
-    // Create a new instance of `Database` and place it inside a `JsBox`
-    // JavaScript can hold a reference to a `JsBox`, but the contents are opaque
-    fn js_new(mut cx: FunctionContext) -> JsResult<JsBox<Database>> {
-        let db = Database::new(&mut cx).or_else(|err| cx.throw_error(err.to_string()))?;
-
-        Ok(cx.boxed(db))
-    }
-
-    // Manually close a database connection
-    // After calling `close`, all other methods will fail
-    // It is not necessary to call `close` since the database will be closed when the wrapping
-    // `JsBox` is garbage collected. However, calling `close` allows the process to exit
-    // immediately instead of waiting on garbage collection. This is useful in tests.
-    fn js_close(mut cx: FunctionContext) -> JsResult<JsUndefined> {
-        // Get the `this` value as a `JsBox<Database>`
-        cx.this::<JsBox<Database>>()?
-            .close()
-            .or_else(|err| cx.throw_error(err.to_string()))?;
-
-        Ok(cx.undefined())
+        Ok(db)
     }
 
     // Inserts a `name` into the database
     // Accepts a `name` and returns a `Promise`
-    fn js_insert(mut cx: FunctionContext) -> JsResult<JsPromise> {
-        // Get the first argument as a `JsString` and convert to a Rust `String`
-        let name = cx.argument::<JsString>(0)?.value(&mut cx);
-
-        // Get the `this` value as a `JsBox<Database>`
-        let db = cx.this::<JsBox<Database>>()?;
-        let (deferred, promise) = cx.promise();
-
-        db.send(deferred, move |conn, channel, deferred| {
-            let result = conn
-                .execute(
-                    "INSERT INTO person (name) VALUES (?)",
-                    rusqlite::params![name],
-                )
-                .map(|_| conn.last_insert_rowid());
-
-            deferred.settle_with(channel, move |mut cx| {
-                let id = result.or_else(|err| cx.throw_error(err.to_string()))?;
-
-                Ok(cx.number(id as f64))
-            });
+    fn insert<'cx>(&self, cx: &mut Cx<'cx>, name: String) -> JsResult<'cx, JsPromise> {
+        self.exec(cx, move |conn| -> Result<_, Error> {
+            conn.execute(
+                "INSERT INTO person (name) VALUES (?)",
+                rusqlite::params![name],
+            )?;
+
+            Ok(conn.last_insert_rowid() as f64)
         })
-        .into_rejection(&mut cx)?;
-
-        Ok(promise)
     }
 
     // Get a `name` by `id` value
     // Accepts an `id` and callback as parameters
-    fn js_get_by_id(mut cx: FunctionContext) -> JsResult<JsPromise> {
-        // Get the first argument as a `JsNumber` and convert to an `f64`
-        let id = cx.argument::<JsNumber>(0)?.value(&mut cx);
-
-        // Get the `this` value as a `JsBox<Database>`
-        let db = cx.this::<JsBox<Database>>()?;
-        let (deferred, promise) = cx.promise();
-
-        db.send(deferred, move |conn, channel, deferred| {
-            let result: Result<String, _> = conn
-                .prepare("SELECT name FROM person WHERE id = ?")
-                .and_then(|mut stmt| stmt.query_row(rusqlite::params![id], |row| row.get(0)));
-
-            deferred.settle_with(channel, move |mut cx| -> JsResult<JsValue> {
-                // If the row was not found, return `undefined` as a success instead
-                // of throwing an exception
-                if matches!(result, Err(rusqlite::Error::QueryReturnedNoRows)) {
-                    return Ok(cx.undefined().upcast());
-                }
-
-                let name = result.or_else(|err| cx.throw_error(err.to_string()))?;
-
-                Ok(cx.string(name).upcast())
-            });
-        })
-        .into_rejection(&mut cx)?;
-
-        Ok(promise)
-    }
-}
-
-trait SendResultExt {
-    // Sending a query closure to execute may fail if the channel has been closed.
-    // This method converts the failure into a promise rejection.
-    fn into_rejection<'a, C: Context<'a>>(self, cx: &mut C) -> NeonResult<()>;
-}
-
-impl SendResultExt for Result<(), mpsc::SendError<DbMessage>> {
-    fn into_rejection<'a, C: Context<'a>>(self, cx: &mut C) -> NeonResult<()> {
-        self.or_else(|err| {
-            let msg = err.to_string();
-
-            match err.0 {
-                DbMessage::Callback(deferred, _) => {
-                    let err = cx.error(msg)?;
-                    deferred.reject(cx, err);
-                    Ok(())
-                }
-                DbMessage::Close => cx.throw_error("Expected DbMessage::Callback"),
+    fn by_id<'cx>(&self, cx: &mut Cx<'cx>, id: f64) -> JsResult<'cx, JsPromise> {
+        self.exec(cx, move |conn| -> Result<Option<String>, Error> {
+            match conn
+                .prepare("SELECT name FROM person WHERE id = ?")?
+                .query_row(rusqlite::params![id], |row| row.get(0))
+            {
+                Ok(name) => Ok(Some(name)),
+                Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
+                Err(err) => Err(Error::from(err)),
             }
         })
     }
-}
-
-#[neon::main]
-fn main(mut cx: ModuleContext) -> NeonResult<()> {
-    cx.export_function("databaseNew", Database::js_new)?;
-    cx.export_function("databaseClose", Database::js_close)?;
-    cx.export_function("databaseInsert", Database::js_insert)?;
-    cx.export_function("databaseGetById", Database::js_get_by_id)?;
 
-    Ok(())
+    // Idiomatic rust would take an owned `self` to prevent use after close
+    // However, it's not possible to prevent JavaScript from continuing to hold a closed database
+    fn close<'cx>(&self, cx: &mut Cx<'cx>) -> JsResult<'cx, JsPromise> {
+        self.send(cx, |_| ControlFlow::<(), ()>::Break(()))
+    }
 }
diff --git a/examples/async-sqlite/test/db.test.js b/examples/async-sqlite/test/db.test.js
@@ -2,7 +2,7 @@
 
 const assert = require("assert");
 
-const Database = require("..");
+const { Database } = require("..");
 
 describe("SQLite Database", () => {
     it("should insert and return name", async () => {
@@ -27,8 +27,7 @@ describe("SQLite Database", () => {
     it("should reject calls to a closed database", async () => {
         const db = new Database();
 
-        db.close();
-
-        await assert.rejects(() => db.byId(5), /closed channel/);
+        await db.close();
+        await assert.rejects(() => db.byId(5), /closed/);
     });
 });
