inprogress: handoff to quexeky

.test2/TESTFILE

@@ -0,0 +1 @@
+g'day what's up my koala bros

Cargo.toml

@@ -9,11 +9,12 @@ crate-type = ["cdylib"]
 
 [dependencies]
 # Default enable napi4 feature, see https://nodejs.org/api/n-api.html#node-api-version-matrix
-napi = { version = "2.12.2", default-features = false, features = [
+napi = { version = "3.0.0-alpha.33", default-features = false, features = [
     "napi4",
     "async",
+    "web_stream",
 ] }
-napi-derive = "2.12.2"
+napi-derive = "3.0.0-alpha.33"
 hex = "0.4.3"
 serde_json = "1.0.128"
 md5 = "0.7.0"
@@ -21,6 +22,8 @@ time-macros = "0.2.22"
 time = "0.3.41"
 webpki = "0.22.4"
 ring = "0.17.14"
+tokio = { version = "1.45.1", features = ["fs"] }
+tokio-util = { version = "0.7.15", features = ["codec"] }
 
 [dependencies.x509-parser]
 version = "0.17.0"

__test__/utils.spec.mjs

@@ -0,0 +1,43 @@
+import test from "ava";
+import fs from "node:fs";
+import path from "path";
+
+import droplet from "../index.js";
+
+test("check alt thread util", async (t) => {
+  let endtime1, endtime2;
+
+  droplet.callAltThreadFunc(async () => {
+    await new Promise((r) => setTimeout(r, 1000));
+    endtime1 = Date.now();
+  });
+
+  await new Promise((r) => setTimeout(r, 5000));
+  endtime2 = Date.now();
+
+  const difference = endtime2 - endtime1;
+  if (difference > 4100 || difference < 3900) {
+    t.fail("timing is not close enough");
+  }
+
+  t.pass();
+});
+
+test("read file", async (t) => {
+  const dirName = "./.test2";
+  if (fs.existsSync(dirName)) fs.rmSync(dirName, { recursive: true });
+  fs.mkdirSync(dirName, { recursive: true });
+
+  fs.writeFileSync("./.test2/TESTFILE", "g'day what's up my koala bros");
+
+  const stream = droplet.readFile("./.test2", "TESTFILE");
+  console.log(stream);
+
+  for await (const chunk of stream) {
+    // Do something with each 'chunk'
+    console.log(chunk);
+  }
+
+  t.pass();
+  fs.rmSync(dirName, { recursive: true });
+});

index.d.ts

@@ -3,12 +3,14 @@
 
 /* auto-generated by NAPI-RS */
 
-export declare function hasBackendForPath(path: string): boolean
+function hasBackendForPath(path: string): boolean
-export declare function listFiles(path: string): Array<string>
+function listFiles(path: string): Array<string>
-export declare function callAltThreadFunc(callback: (...args: any[]) => any): void
+function readFile(path: string, subPath: string): ReadableStream<Array<number>> | null
-export declare function generateManifest(dir: string, progress: (...args: any[]) => any, log: (...args: any[]) => any, callback: (...args: any[]) => any): void
+function callAltThreadFunc(tsfn: ((err: Error | null, ) => any)): void
-export declare function generateRootCa(): Array<string>
+function generateManifest(dir: string, progressSfn: ((err: Error | null, arg: number) => any), logSfn: ((err: Error | null, arg: string) => any), callbackSfn: ((err: Error | null, arg: string) => any)): void
-export declare function generateClientCertificate(clientId: string, clientName: string, rootCa: string, rootCaPrivate: string): Array<string>
+function generateRootCa(): Array<string>
-export declare function verifyClientCertificate(clientCert: string, rootCa: string): boolean
+function generateClientCertificate(clientId: string, clientName: string, rootCa: string, rootCaPrivate: string): Array<string>
-export declare function signNonce(privateKey: string, nonce: string): string
+function verifyClientCertificate(clientCert: string, rootCa: string): boolean
-export declare function verifyNonce(publicCert: string, nonce: string, signature: string): boolean
+function signNonce(privateKey: string, nonce: string): string
+function verifyNonce(publicCert: string, nonce: string, signature: string): boolean
+undefinedundefined

index.js

@@ -310,10 +310,11 @@ if (!nativeBinding) {
   throw new Error(`Failed to load native binding`)
 }
 
-const { hasBackendForPath, listFiles, callAltThreadFunc, generateManifest, generateRootCa, generateClientCertificate, verifyClientCertificate, signNonce, verifyNonce } = nativeBinding
+const { hasBackendForPath, listFiles, readFile, callAltThreadFunc, generateManifest, generateRootCa, generateClientCertificate, verifyClientCertificate, signNonce, verifyNonce,  } = nativeBinding
 
 module.exports.hasBackendForPath = hasBackendForPath
 module.exports.listFiles = listFiles
+module.exports.readFile = readFile
 module.exports.callAltThreadFunc = callAltThreadFunc
 module.exports.generateManifest = generateManifest
 module.exports.generateRootCa = generateRootCa
@@ -321,3 +322,4 @@ module.exports.generateClientCertificate = generateClientCertificate
 module.exports.verifyClientCertificate = verifyClientCertificate
 module.exports.signNonce = signNonce
 module.exports.verifyNonce = verifyNonce
+module.exports.undefined = undefined

package.json

@@ -21,7 +21,7 @@
   },
   "license": "MIT",
   "devDependencies": {
-    "@napi-rs/cli": "^2.18.4",
+    "@napi-rs/cli": "2.18.4",
     "@types/node": "^22.13.10",
     "ava": "^6.2.0"
   },

src/file_utils.rs

@@ -2,12 +2,13 @@
 use std::os::unix::fs::PermissionsExt;
 use std::{
   fs::{self, metadata, File},
-  io::BufReader,
+  io::{BufReader, Read},
   path::{Path, PathBuf},
+  task::Poll,
 };
 
-const CHUNK_SIZE: usize = 1024 * 1024 * 64;
+use napi::{bindgen_prelude::*, tokio_stream::{Stream, StreamExt}};
-
+use tokio_util::{bytes::BytesMut, codec::{BytesCodec, FramedRead}};
 
 fn _list_files(vec: &mut Vec<PathBuf>, path: &Path) {
   if metadata(path).unwrap().is_dir() {
@@ -30,7 +31,7 @@ pub struct VersionFile {
 
 pub trait VersionBackend: 'static {
   fn list_files(&self, path: &Path) -> Vec<VersionFile>;
-  fn reader(&self, file: &VersionFile) -> BufReader<File>;
+  fn reader(&self, file: &VersionFile) -> Option<File>;
 }
 
 pub struct PathVersionBackend {
@@ -70,10 +71,10 @@ impl VersionBackend for PathVersionBackend {
     results
   }
 
-  fn reader(&self, file: &VersionFile) -> BufReader<File> {
+  fn reader(&self, file: &VersionFile) -> Option<File> {
-    let file = File::open(self.base_dir.join(file.relative_filename.clone())).unwrap();
+    let file = File::open(self.base_dir.join(file.relative_filename.clone())).ok()?;
-    let reader = BufReader::with_capacity(CHUNK_SIZE, file);
+
-    return reader;
+    return Some(file);
   }
 }
 
@@ -85,7 +86,7 @@ impl VersionBackend for ArchiveVersionBackend {
     todo!()
   }
 
-  fn reader(&self, file: &VersionFile) -> BufReader<File> {
+  fn reader(&self, file: &VersionFile) -> Option<File> {
     todo!()
   }
 }
@@ -121,3 +122,27 @@ pub fn list_files(path: String) -> Vec<String> {
   let files = backend.list_files(path);
   files.into_iter().map(|e| e.relative_filename).collect()
 }
+
+#[napi]
+pub fn read_file(
+  path: String,
+  sub_path: String,
+  env: &Env,
+) -> Option<ReadableStream<'static, Vec<u8>>> {
+  let path = Path::new(&path);
+  let backend = create_backend_for_path(path).unwrap();
+  let version_file = VersionFile {
+    relative_filename: sub_path,
+    permission: 0, // Shouldn't matter
+  };
+  let reader = backend.reader(&version_file)?;
+  let reader = tokio::fs::File::from_std(reader);
+  let stream = FramedRead::new(reader, BytesCodec::new()).map(|e| {
+    if let Ok(bytes) = e {
+      Ok(bytes.to_vec())
+    } else {
+      Err(napi::Error::from_reason(e.unwrap_err().to_string()))
+    }
+  });
+  Some(ReadableStream::create_with_stream_bytes(env, stream).unwrap())
+}

src/manifest.rs

@@ -1,23 +1,22 @@
 use std::{
-  collections::HashMap,
+  collections::HashMap, fs::File, io::{BufRead, BufReader}, path::Path, rc::Rc, sync::Arc, thread
-  fs::File,
-  io::{BufRead, BufReader},
-  path::Path,
-  thread,
 };
 
 #[cfg(unix)]
 use std::os::unix::fs::PermissionsExt;
 
 use napi::{
-  threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode},
+  bindgen_prelude::Function,
-  Error, JsFunction,
+  threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode},
+  Env, Error, Result,
 };
 use serde_json::json;
 use uuid::Uuid;
 
 use crate::file_utils::create_backend_for_path;
 
+const CHUNK_SIZE: usize = 1024 * 1024 * 64;
+
 #[derive(serde::Serialize)]
 struct ChunkData {
   permissions: u32,
@@ -27,14 +26,10 @@ struct ChunkData {
 }
 
 #[napi]
-pub fn call_alt_thread_func(callback: JsFunction) -> Result<(), Error> {
+pub fn call_alt_thread_func(tsfn: Arc<ThreadsafeFunction<()>>) -> Result<(), String> {
-  let tsfn: ThreadsafeFunction<u32, ErrorStrategy::CalleeHandled> = callback
+  let tsfn_cloned = tsfn.clone();
-    .create_threadsafe_function(0, |ctx| {
-      ctx.env.create_uint32(ctx.value + 1).map(|v| vec![v])
-    })?;
-  let tsfn = tsfn.clone();
   thread::spawn(move || {
-    tsfn.call(Ok(0), ThreadsafeFunctionCallMode::NonBlocking);
+    tsfn_cloned.call(Ok(()), ThreadsafeFunctionCallMode::Blocking);
   });
   Ok(())
 }
@@ -42,24 +37,10 @@ pub fn call_alt_thread_func(callback: JsFunction) -> Result<(), Error> {
 #[napi]
 pub fn generate_manifest(
   dir: String,
-  progress: JsFunction,
+  progress_sfn: ThreadsafeFunction<i32>,
-  log: JsFunction,
+  log_sfn: ThreadsafeFunction<String>,
-  callback: JsFunction,
+  callback_sfn: ThreadsafeFunction<String>,
-) -> Result<(), Error> {
+) -> Result<(), String> {
-  let progress_sfn: ThreadsafeFunction<i32, ErrorStrategy::CalleeHandled> = progress
-    .create_threadsafe_function(0, |ctx| ctx.env.create_int32(ctx.value).map(|v| vec![v]))
-    .unwrap();
-  let log_sfn: ThreadsafeFunction<String, ErrorStrategy::CalleeHandled> = log
-    .create_threadsafe_function(0, |ctx| {
-      ctx.env.create_string_from_std(ctx.value).map(|v| vec![v])
-    })
-    .unwrap();
-  let callback_sfn: ThreadsafeFunction<String, ErrorStrategy::CalleeHandled> = callback
-    .create_threadsafe_function(0, |ctx| {
-      ctx.env.create_string_from_std(ctx.value).map(|v| vec![v])
-    })
-    .unwrap();
-
   thread::spawn(move || {
     let base_dir = Path::new(&dir);
     let backend = create_backend_for_path(base_dir).unwrap();
@@ -72,7 +53,8 @@ pub fn generate_manifest(
     let mut i: i32 = 0;
 
     for version_file in files {
-      let mut reader = backend.reader(&version_file);
+      let mut raw_reader= backend.reader(&version_file).unwrap();
+      let mut reader = BufReader::with_capacity(CHUNK_SIZE, raw_reader);
 
       let mut chunk_data = ChunkData {
         permissions: version_file.permission,
@@ -101,8 +83,7 @@ pub fn generate_manifest(
 
         let log_str = format!(
           "Processed chunk {} for {}",
-          chunk_index,
+          chunk_index, &version_file.relative_filename
-          &version_file.relative_filename
         );
         log_sfn.call(Ok(log_str), ThreadsafeFunctionCallMode::Blocking);
 

yarn.lock

@@ -9,7 +9,7 @@ __metadata:
   version: 0.0.0-use.local
   resolution: "@drop-oss/droplet@workspace:."
   dependencies:
-    "@napi-rs/cli": "npm:^2.18.4"
+    "@napi-rs/cli": "npm:2.18.4"
     "@types/node": "npm:^22.13.10"
     ava: "npm:^6.2.0"
   languageName: unknown
@@ -55,7 +55,7 @@ __metadata:
   languageName: node
   linkType: hard
 
-"@napi-rs/cli@npm:^2.18.4":
+"@napi-rs/cli@npm:2.18.4":
   version: 2.18.4
   resolution: "@napi-rs/cli@npm:2.18.4"
   bin: