diff --git a/.gitignore b/.gitignore index d002fce..5dd689d 100644 --- a/.gitignore +++ b/.gitignore @@ -201,4 +201,7 @@ test.mjs manifest.json # JetBrains -.idea \ No newline at end of file +.idea + + +TESTFILE diff --git a/.test2/TESTFILE b/.test2/TESTFILE new file mode 100644 index 0000000..e85fb16 --- /dev/null +++ b/.test2/TESTFILE @@ -0,0 +1 @@ +g'day what's up my koala bros \ No newline at end of file diff --git a/Cargo.toml b/Cargo.toml index d95a22b..a3142c3 100644 --- a/Cargo.toml +++ b/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" diff --git a/__test__/utils.spec.mjs b/__test__/utils.spec.mjs new file mode 100644 index 0000000..7522629 --- /dev/null +++ b/__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 }); +}); diff --git a/index.d.ts b/index.d.ts index 09351cd..6b181bf 100644 --- a/index.d.ts +++ b/index.d.ts @@ -3,12 +3,14 @@ /* auto-generated by NAPI-RS */ -export declare function hasBackendForPath(path: string): boolean -export declare function listFiles(path: string): Array -export declare function callAltThreadFunc(callback: (...args: any[]) => any): void -export declare function generateManifest(dir: string, progress: (...args: any[]) => any, log: (...args: any[]) => any, callback: (...args: any[]) => any): void -export declare function generateRootCa(): Array -export declare function generateClientCertificate(clientId: string, clientName: string, rootCa: string, rootCaPrivate: string): Array -export declare function verifyClientCertificate(clientCert: string, rootCa: string): boolean -export declare function signNonce(privateKey: string, nonce: string): string -export declare function verifyNonce(publicCert: string, nonce: string, signature: string): boolean +function hasBackendForPath(path: string): boolean +function listFiles(path: string): Array +function readFile(path: string, subPath: string): ReadableStream | null +function callAltThreadFunc(tsfn: ((err: Error | null, ) => any)): void +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 +function generateRootCa(): Array +function generateClientCertificate(clientId: string, clientName: string, rootCa: string, rootCaPrivate: string): Array +function verifyClientCertificate(clientCert: string, rootCa: string): boolean +function signNonce(privateKey: string, nonce: string): string +function verifyNonce(publicCert: string, nonce: string, signature: string): boolean +undefinedundefined diff --git a/index.js b/index.js index 28697a4..c7cc0b4 100644 --- a/index.js +++ b/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 diff --git a/package.json b/package.json index 2fc1cbe..a6aaaf5 100644 --- a/package.json +++ b/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" }, diff --git a/src/file_utils.rs b/src/file_utils.rs index bef2ce5..bd4d126 100644 --- a/src/file_utils.rs +++ b/src/file_utils.rs @@ -2,12 +2,19 @@ use std::os::unix::fs::PermissionsExt; use std::{ fs::{self, metadata, File}, - io::BufReader, + io::{self, BufReader, ErrorKind, 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, path: &Path) { if metadata(path).unwrap().is_dir() { @@ -30,7 +37,7 @@ pub struct VersionFile { pub trait VersionBackend: 'static { fn list_files(&self, path: &Path) -> Vec; - fn reader(&self, file: &VersionFile) -> BufReader; + fn reader(&self, file: &VersionFile) -> Option; } pub struct PathVersionBackend { @@ -70,10 +77,10 @@ impl VersionBackend for PathVersionBackend { results } - fn reader(&self, file: &VersionFile) -> BufReader { - let file = File::open(self.base_dir.join(file.relative_filename.clone())).unwrap(); - let reader = BufReader::with_capacity(CHUNK_SIZE, file); - return reader; + fn reader(&self, file: &VersionFile) -> Option { + let file = File::open(self.base_dir.join(file.relative_filename.clone())).ok()?; + + return Some(file); } } @@ -85,7 +92,7 @@ impl VersionBackend for ArchiveVersionBackend { todo!() } - fn reader(&self, file: &VersionFile) -> BufReader { + fn reader(&self, file: &VersionFile) -> Option { todo!() } } @@ -120,4 +127,39 @@ pub fn list_files(path: String) -> Vec { let backend = create_backend_for_path(path).unwrap(); let files = backend.list_files(path); files.into_iter().map(|e| e.relative_filename).collect() -} \ No newline at end of file +} + +#[napi] +pub fn read_file( + path: String, + sub_path: String, + env: &Env, +) -> Option>> { + 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 + }; + // Use `?` operator for cleaner error propagation from `Option` + let reader = backend.reader(&version_file)?; + + // Convert std::fs::File to tokio::fs::File for async operations + let reader = tokio::fs::File::from_std(reader); + + // Create a FramedRead stream with BytesCodec for chunking + + let stream = FramedRead::new(reader, BytesCodec::new()) + // Use StreamExt::map to transform each Result item + .map(|result_item| { + result_item + // Apply Result::map to transform Ok(BytesMut) to Ok(Vec) + .map(|bytes| bytes.to_vec()) + // Apply Result::map_err to transform Err(std::io::Error) to Err(napi::Error) + .map_err(|e| napi::Error::from(e)) // napi::Error implements From + }); + // Create the napi-rs ReadableStream from the tokio_stream::Stream + // The unwrap() here means if stream creation fails, it will panic. + // For a production system, consider returning Result> and handling this. + Some(ReadableStream::create_with_stream_bytes(env, stream).unwrap()) +} diff --git a/src/manifest.rs b/src/manifest.rs index eaf00f9..c5136c6 100644 --- a/src/manifest.rs +++ b/src/manifest.rs @@ -1,23 +1,22 @@ use std::{ - collections::HashMap, - fs::File, - io::{BufRead, BufReader}, - path::Path, - thread, + collections::HashMap, fs::File, io::{BufRead, BufReader}, path::Path, rc::Rc, sync::Arc, thread }; #[cfg(unix)] use std::os::unix::fs::PermissionsExt; use napi::{ - threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode}, - Error, JsFunction, + bindgen_prelude::Function, + 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> { - let tsfn: ThreadsafeFunction = callback - .create_threadsafe_function(0, |ctx| { - ctx.env.create_uint32(ctx.value + 1).map(|v| vec![v]) - })?; - let tsfn = tsfn.clone(); +pub fn call_alt_thread_func(tsfn: Arc>) -> Result<(), String> { + let tsfn_cloned = 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, - log: JsFunction, - callback: JsFunction, -) -> Result<(), Error> { - let progress_sfn: ThreadsafeFunction = progress - .create_threadsafe_function(0, |ctx| ctx.env.create_int32(ctx.value).map(|v| vec![v])) - .unwrap(); - let log_sfn: ThreadsafeFunction = log - .create_threadsafe_function(0, |ctx| { - ctx.env.create_string_from_std(ctx.value).map(|v| vec![v]) - }) - .unwrap(); - let callback_sfn: ThreadsafeFunction = callback - .create_threadsafe_function(0, |ctx| { - ctx.env.create_string_from_std(ctx.value).map(|v| vec![v]) - }) - .unwrap(); - + progress_sfn: ThreadsafeFunction, + log_sfn: ThreadsafeFunction, + callback_sfn: ThreadsafeFunction, +) -> Result<(), String> { 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, - &version_file.relative_filename + chunk_index, &version_file.relative_filename ); log_sfn.call(Ok(log_str), ThreadsafeFunctionCallMode::Blocking); diff --git a/yarn.lock b/yarn.lock index 071dd0f..7916d6d 100644 --- a/yarn.lock +++ b/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: