feat: Add on-the-fly filtering

Author: mochatek

README.md

@@ -1,6 +1,19 @@
 # jsonarrayfs <img src="https://img.shields.io/npm/dm/jsonarrayfs" />
 
-jsonarrayfs is a Node.js library designed to facilitate efficient handling of JSON array data stored in files. It offers functionality for streaming JSON array elements in batches, allowing for reduced memory consumption when working with large datasets. Additionally, jsonarrayfs provides a method for appending new data to existing JSON array files without loading the entire dataset into memory.
+"jsonarrayfs" is a Node.js library crafted for robust and memory-efficient management of massive JSON array files. It enables seamless handling of JSON arrays without the need to load the entire file into memory, making it perfect for efficiently managing large datasets without overwhelming system resources.
+
+## Key Features
+
+- **Stream Processing**: Read JSON array in manageable chunks (eg: 50k elements at a time) using stream.
+- **On-the-Fly Filtering**: Apply filter to the stream to fetch only relevant data, reducing the data you handle even further.
+- **Direct Appends**: Append new elements directly to the JSON array file, avoiding unnecessary loading, modification and rewriting.
+
+## Benefits
+
+- **Memory Optimization**: Process JSON array files with minimal memory usage, making it ideal for resource-constrained environments.
+- **Handles Large Datasets**: Efficiently manage massive JSON array files without memory limitations.
+- **Improved Performance**: Faster processing times due to efficient streaming, filtering and appending capabilities.
+- **Enhanced Scalability**: Scales seamlessly with growing datasets, ensuring smooth performance.
 
 ## Installation
 
@@ -10,35 +23,54 @@ To install jsonarrayfs, use:
 npm install jsonarrayfs
 ```
 
-## Features
+## Usage
 
-- Stream JSON array elements:
+- Stream Processing:
 
-```js
+```ts
 import { createReadStream } from "jsonarrayfs";
 
 // Create a streamer to read JSON array elements from a file
-const streamer = await createReadStream("./data.json", { encoding: 'utf-8' });
+const streamer = await createReadStream("./data.json", { encoding: "utf-8" });
 
 // Stream JSON array elements in batches of 100
 for await (const chunk of streamer.stream(100)) {
   // Your processing logic here
 }
 ```
 
+- On-the-Fly Filtering:
+
+```ts
+import { createReadStream } from "jsonarrayfs";
+
+const streamer = await createReadStream<{ offer: boolean; price: number }>(
+  "./data.json",
+  { encoding: "utf-8" }
+);
+
+// Add filter to the stream to fetch only relevant elements
+for await (const chunk of streamer.stream(
+  100,
+  (element) => element.price < 500 || element.offer
+)) {
+  // Your processing logic here
+}
+```
+
 - Append data to existing JSON array:
 
-```js
+```ts
 import { appendFile } from "jsonarrayfs";
 
 // Simulate new data to append
 const newData = [
-  { id: 1, name: "JavaScript" },
-  { id: 2, name: "Go" }
+  { id: 1, name: "Earth", price: 1000, offer: true },
+  { id: 2, name: "Moon", price: 500, offer: false },
 ];
 
 // Append new data to the existing JSON array file
-await appendFile("./data.json", 'utf-8', ...newData);
+await appendFile("./data.json", "utf-8", ...newData);
 ```
 
 ## Contributing
@@ -47,4 +79,4 @@ Pull requests are welcome. For major changes, please open an issue first to disc
 
 ## License
 
-[MIT License ](https://github.com/mochatek/jsonarrayfs/blob/main/LICENSE)
+[MIT License ](https://github.com/mochatek/jsonarrayfs/blob/main/LICENSE)

package.json

@@ -1,7 +1,7 @@
 {
   "name": "jsonarrayfs",
   "version": "1.0.1",
-  "description": "A Node.js library to stream JSON array elements from files in custom-sized chunks and effortlessly append data to existing arrays.",
+  "description": "Efficiently handle JSON array files in Node.js with minimal memory usage. Perfect for efficiently processing large data volumes without worrying about memory limitations.",
   "exports": {
     ".": {
       "require": "./dist/index.js",
@@ -21,10 +21,17 @@
   "keywords": [
     "json",
     "stream",
-    "json array",
     "chunk",
     "batch",
-    "append json"
+    "json file handling",
+    "json array",
+    "json stream",
+    "stream filter",
+    "json append",
+    "stream processing",
+    "on the fly filtering",
+    "direct appends",
+    "memory efficient"
   ],
   "author": "Akash S Panickar",
   "license": "MIT",

src/modules/JsonArrayStreamer.ts

@@ -8,7 +8,9 @@ class JsonArrayStreamer<T> {
   private rootDetected: boolean;
   private elementDetected: boolean;
   private elementType: ElementType;
-  private elementParser: ((char: string) => void) | null;
+  private elementParser:
+    | ((char: string, filter?: (element: T) => boolean) => void)
+    | null;
   private elementEnclosureCount: number;
   private isCharInsideQuotes: boolean;
   private isCharEscaped: boolean;
@@ -39,6 +41,16 @@ class JsonArrayStreamer<T> {
     }
   }
 
+  private addToResult(element: T, filter?: (element: T) => boolean) {
+    if (!filter) {
+      this.resultBuffer.push(element);
+    } else {
+      try {
+        if (filter(element)) this.resultBuffer.push(element);
+      } catch (_) {}
+    }
+  }
+
   private resetParser() {
     this.elementDetected = false;
     this.elementType = "others";
@@ -49,13 +61,13 @@ class JsonArrayStreamer<T> {
     this.chunkBuffer = "";
   }
 
-  private stringElementParser(char: string) {
+  private stringElementParser(char: string, filter?: (element: T) => boolean) {
     this.chunkBuffer = `${this.chunkBuffer}${char}`;
 
     if (char === CHARACTER.QUOTE) {
       if (this.isCharInsideQuotes && !this.isCharEscaped) {
         const element: T = JSON.parse(this.chunkBuffer);
-        this.resultBuffer.push(element);
+        this.addToResult(element, filter);
         this.resetParser();
       } else if (this.chunkBuffer === CHARACTER.QUOTE) {
         this.isCharInsideQuotes = true;
@@ -69,17 +81,23 @@ class JsonArrayStreamer<T> {
     }
   }
 
-  private primitiveElementParser(char: string) {
+  private primitiveElementParser(
+    char: string,
+    filter?: (element: T) => boolean
+  ) {
     if ([CHARACTER.COMMA, CHARACTER.BRACKET.CLOSE].includes(char)) {
       const element: T = JSON.parse(this.chunkBuffer);
-      this.resultBuffer.push(element);
+      this.addToResult(element, filter);
       this.resetParser();
     } else {
       this.chunkBuffer = `${this.chunkBuffer}${char}`;
     }
   }
 
-  private containerElementParser(char: string) {
+  private containerElementParser(
+    char: string,
+    filter?: (element: T) => boolean
+  ) {
     const ENCLOSURE =
       this.elementType === "array" ? CHARACTER.BRACKET : CHARACTER.BRACE;
 
@@ -92,7 +110,7 @@ class JsonArrayStreamer<T> {
 
       if (this.elementEnclosureCount === 0) {
         const element: T = JSON.parse(this.chunkBuffer);
-        this.resultBuffer.push(element);
+        this.addToResult(element, filter);
         this.resetParser();
       }
     } else if (this.chunkBuffer.length) {
@@ -108,7 +126,7 @@ class JsonArrayStreamer<T> {
     }
   }
 
-  public async *stream<T>(chunkSize: number) {
+  public async *stream(chunkSize: number, filter?: (element: T) => boolean) {
     for await (const chunk of this.chunkGenerator()) {
       for (let char of chunk) {
         if (!this.rootDetected) {
@@ -142,7 +160,7 @@ class JsonArrayStreamer<T> {
             }
           }
 
-          this.elementParser(char);
+          this.elementParser(char, filter);
 
           if (this.resultBuffer.length === chunkSize) {
             if (!this.readStream?.closed) this.readStream?.pause();
@@ -157,8 +175,8 @@ class JsonArrayStreamer<T> {
     this.readStream = null;
 
     if (this.chunkBuffer.length) {
-      const element = JSON.parse(this.chunkBuffer);
-      this.resultBuffer.push(element);
+      const element: T = JSON.parse(this.chunkBuffer);
+      this.addToResult(element, filter);
       this.resetParser();
     }
     if (this.resultBuffer.length) {