fix(createReadStream): Default batch size and change method name

Author: mochatek

README.md

@@ -52,7 +52,7 @@ const readableStream = Readable.fromWeb(
 const streamer = await createReadStream(readableStream);
 
 // Stream JSON array elements in batches of 100
-for await (const chunk of streamer.stream(100)) {
+for await (const elements of streamer.batch(100)) {
   // Your processing logic here
 }
 ```
@@ -67,8 +67,8 @@ const streamer = await createReadStream<{ offer: boolean; price: number }>(
   { encoding: "utf-8" }
 );
 
-// Add filter to the stream to fetch only relevant elements
-for await (const chunk of streamer.stream(
+// Add filter to the batch to fetch only relevant elements
+for await (const elements of streamer.batch(
   100,
   (element) => element.price < 500 || element.offer
 )) {

package-lock.json

@@ -138,7 +138,7 @@ class JsonArrayStreamer<T> {
     }
   }
 
-  public async *stream(chunkSize: number, filter?: (element: T) => boolean) {
+  public async *batch(batchSize?: number, filter?: (element: T) => boolean) {
     try {
       characterStream: for await (const chunk of this.chunkGenerator()) {
         for (let char of chunk) {
@@ -190,9 +190,13 @@ class JsonArrayStreamer<T> {
 
             this.elementParser(char, filter);
 
-            if (this.resultBuffer.length === chunkSize) {
+            const targetSize = Math.max(
+              1,
+              batchSize || this.resultBuffer.length
+            );
+            if (this.resultBuffer.length === targetSize) {
               if (!this.readStream?.closed) this.readStream?.pause();
-              yield this.resultBuffer.splice(0, chunkSize);
+              yield this.resultBuffer.splice(0, targetSize);
               if (!this.readStream?.closed) this.readStream?.resume();
             }
           }