fix(createReadStream): Add support to stream from any Readable

Author: mochatek

README.md

@@ -31,13 +31,25 @@ npm install jsonarrayfs
 
 ```ts
 import { createReadStream } from "jsonarrayfs";
-// import fs from "fs";
 
 // Option 1: Create a streamer to read JSON array elements from a file
 const streamer = await createReadStream("./data.json", { encoding: "utf-8" });
 
-// Option 2: Pass an existing readStream
-// const streamer = await createReadStream(fs.createReadStream("./data.json", { encoding: "utf-8" }));
+// Option 2: Create a streamer to read JSON array elements from an existing Readable:
+
+// From a file
+import fs from "fs";
+const readStream = fs.createReadStream("./data.json", { encoding: "utf-8" });
+const streamer = await createReadStream(readStream);
+
+// From an API response
+import { ReadableStream } from "stream/web";
+import { Readable } from "stream";
+const response = await fetch("https://www.example.com/json-data");
+const readableStream = Readable.fromWeb(
+  response.body as unknown as ReadableStream
+);
+const streamer = await createReadStream(readableStream);
 
 // Stream JSON array elements in batches of 100
 for await (const chunk of streamer.stream(100)) {

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "jsonarrayfs",
-  "version": "1.2.0",
+  "version": "1.2.1",
   "description": "Efficiently handle JSON array files in Node.js with minimal memory usage. Perfect for processing large data volumes without worrying about memory limitations.",
   "main": "./dist/index.js",
   "module": "./dist/index.mjs",

package.json

@@ -1,10 +1,11 @@
-import { ReadStream, createReadStream } from "fs";
+import { createReadStream } from "fs";
+import { Readable } from "stream";
 import { once } from "events";
 import type { ElementType, ReadStreamOptions } from "../index.types";
 import { CHARACTER, ERRORS } from "../constants";
 
 class JsonArrayStreamer<T> {
-  private readStream: ReadStream | null;
+  private readStream: Readable | null;
   private rootDetected: boolean;
   private elementDetected: boolean;
   private elementType: ElementType;
@@ -17,12 +18,9 @@ class JsonArrayStreamer<T> {
   private chunkBuffer: string;
   private resultBuffer: T[];
 
-  private constructor(readStream: ReadStream);
+  private constructor(readStream: Readable);
   private constructor(filePath: string, options?: ReadStreamOptions);
-  private constructor(
-    source: ReadStream | string,
-    options?: ReadStreamOptions
-  ) {
+  private constructor(source: Readable | string, options?: ReadStreamOptions) {
     this.readStream = JsonArrayStreamer.getReadStreamWithEncoding(
       source,
       options
@@ -201,7 +199,7 @@ class JsonArrayStreamer<T> {
         }
       }
 
-      this.readStream?.close();
+      this.readStream?.destroy();
       this.readStream = null;
 
       if (this.chunkBuffer.length) {
@@ -215,7 +213,7 @@ class JsonArrayStreamer<T> {
 
       return this.resultBuffer;
     } catch (error) {
-      this.readStream?.close();
+      this.readStream?.destroy();
       this.resetParser();
       this.resultBuffer = [];
       this.readStream = null;
@@ -237,11 +235,11 @@ class JsonArrayStreamer<T> {
   };
 
   private static getReadStreamWithEncoding = (
-    source: ReadStream | string,
+    source: Readable | string,
     options?: ReadStreamOptions
   ) => {
     const readStream =
-      source instanceof ReadStream
+      source instanceof Readable
         ? source
         : createReadStream(
             source,
@@ -256,18 +254,16 @@ class JsonArrayStreamer<T> {
     return readStream;
   };
 
-  public static create<T>(
-    readStream: ReadStream
-  ): Promise<JsonArrayStreamer<T>>;
+  public static create<T>(readStream: Readable): Promise<JsonArrayStreamer<T>>;
   public static create<T>(
     filePath: string,
     options?: ReadStreamOptions
   ): Promise<JsonArrayStreamer<T>>;
   public static async create<T>(
-    source: ReadStream | string,
+    source: Readable | string,
     options?: ReadStreamOptions
   ): Promise<JsonArrayStreamer<T>> {
-    const sourceIsReadableStream = source instanceof ReadStream;
+    const sourceIsReadableStream = source instanceof Readable;
     const instance = sourceIsReadableStream
       ? new JsonArrayStreamer<T>(source)
       : new JsonArrayStreamer<T>(source, options);