Merge pull request #12 from palcarazm/fix/backoff

fix: backoff pressure

Author: palcarazm

package.json

@@ -1,6 +1,6 @@
 {
   "name": "batchjs-data",
-  "version": "1.0.0",
+  "version": "1.0.1",
   "author": {
     "name": "Pablo Alcaraz Martínez",
     "url": "https://github.com/palcarazm/"
@@ -79,7 +79,7 @@
     "sqlite3": "^5.1.7"
   },
   "dependencies": {
-    "batchjs": "^1.1.0"
+    "batchjs": "^1.1.3"
   },
   "commitlint": {
     "extends": "@commitlint/config-conventional"

src/common/classes/AbstractBatchEntityReaderStream.ts

@@ -38,24 +38,26 @@ export abstract class AbstractBatchEntityReaderStream<T> extends ObjectReadable
      * If an error occurs while reading data, it emits an error event to the stream.
      *
      * @param {number} [size] - The size parameter for controlling the read operation.
-     * @returns {Promise<void>} A promise that resolves when the data has been read and pushed to the consumer stream.
      */
-    async _read(size: number): Promise<void> {
+    _read(size: number): void {
         if (this.reading) return;
+
         this.reading = true;
-        try {
-            const entities: T[] = await this.fetch(Math.min(size, this.batchSize));
-            if (entities.length === 0) {
-                this.push(null);
-            }else{
-                this.buffer.push(...entities);
-                await this._flush();
-            }
-        } catch (error) {
-            this.emit("error", error as Error);
-        }finally{
-            this.reading = false;
-        }
+
+        this.fetch(Math.min(size, this.batchSize))
+            .then((entities) => {
+                if (entities.length === 0) {
+                    this.push(null);
+                } else {
+                    this.buffer.push(...entities);
+                    this._flush().finally(() => {
+                        this.reading = false;
+                    });
+                }
+            })
+            .catch((error) => {
+                this.emit("error", error);
+            });
     }
 
     /**
@@ -65,12 +67,12 @@ export abstract class AbstractBatchEntityReaderStream<T> extends ObjectReadable
      * @private
      * @returns {Promise<void>} A promise that resolves when the buffer is flushed.
      */
-    private async _flush():Promise<void>{
+    private _flush():Promise<void>{
         while (this.buffer.length > 0) {
             const chunk = this.buffer.shift() as T;
             if (!this.push(chunk)) {
-                this.buffer.unshift(chunk);
-                await new Promise<void>((resolve) => this.once("drain", resolve));
+                this.once("drain", () => this._flush());
+                return Promise.resolve();
             }
         }
         return Promise.resolve();

src/common/classes/AbstractBatchEntityWriterStream.ts

@@ -35,19 +35,17 @@ export abstract class AbstractBatchEntityWriterStream<T> extends ObjectWritable
      * @param {T} chunk - The data chunk to write to the stream.
      * @param {BufferEncoding} encoding - The encoding of the data.
      * @param {WriteCallback} callback - The callback function to be executed after writing the data.
-     * @return {Promise<void>} This function does not return anything.
      */
-    async _write(chunk: T, encoding: BufferEncoding, callback: WriteCallback): Promise<void> {
-        try {
+    _write(chunk: T, encoding: BufferEncoding, callback: WriteCallback): void {
             this.buffer.push(chunk);
             if (this.buffer.length >= this.batchSize) {
-                await this._flush();
+                  this._flush()
+                      .then(() => callback())
+                      .catch((error) => callback(error));
+            } else {
+                callback();
             }
-            callback();
-        } catch (error) {
-            callback(error as Error);
-        }
-    }
+       }
 
     /**
      * Finalizes the stream by pushing remaining data batches, handling errors,

test/common/classes/AbstractBatchEntityReaderStream.test.ts

@@ -63,7 +63,10 @@ describe("AbstractBatchEntityReaderStream", () => {
             result.push(chunk);
 
             if (result.length === 2) {
-                spy.mockImplementationOnce(() => false);
+                spy.mockImplementationOnce((data) => {
+                    reader.emit('data',data);
+                    return false;
+                });
                 setTimeout(()=>{
                     reader.emit("drain");
                 },50);
@@ -72,7 +75,7 @@ describe("AbstractBatchEntityReaderStream", () => {
 
         reader.on("end", () => {
             expect(result).toEqual(data);
-            expect(spy).toHaveBeenCalledTimes(data.length + 2);
+            expect(spy).toHaveBeenCalledTimes(data.length + 1);
             done();
         });