3.1.8

changelog.md

@@ -1,5 +1,29 @@
 # Changelog
 
+## 2024-10-16 - 3.1.8 - fix(VirtualStream)
+Fix stream closing behavior to correctly handle closing bits
+
+- Introduced a 'closingBit' constant to properly signal the end of stream data.
+- Updated the 'readFromWebstream' function to send a closing bit upon completion if 'closeAfterReading' is true.
+- Modified the 'close' method to optionally send a closing bit when terminating the stream.
+
+## 2024-10-16 - 3.1.7 - fix(VirtualStream)
+Fix issue in VirtualStream to handle null values during data writing.
+
+- Ensured writableStream closes gracefully when null values are encountered.
+- Added a null check before writing data to the writableStream to prevent errors.
+
+## 2024-10-16 - 3.1.6 - fix(VirtualStream)
+Fix backpressure handling in VirtualStream workOnQueue method
+
+- Resolved an issue in the workOnQueue method of VirtualStream where concurrent execution was not properly managed.
+- Introduced a workingDeferred promise to ensure proper queue handling and resolve potential race conditions.
+
+## 2024-10-16 - 3.1.5 - fix(virtualstream)
+Add console log for debugging backpressure feedback loop
+
+- Inserted a console log message to provide insight when waiting due to backpressure in the workOnQueue method.
+
 ## 2024-10-16 - 3.1.4 - fix(VirtualStream)
 Corrected the logic for backpressure handling in response
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@api.global/typedrequest",
-  "version": "3.1.4",
+  "version": "3.1.8",
   "private": false,
   "description": "A TypeScript library for making typed requests towards APIs, including facilities for handling requests, routing, and virtual stream handling.",
   "main": "dist_ts/index.js",

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@api.global/typedrequest',
-  version: '3.1.4',
+  version: '3.1.8',
   description: 'A TypeScript library for making typed requests towards APIs, including facilities for handling requests, routing, and virtual stream handling.'
 }

ts/classes.virtualstream.ts

@@ -1,6 +1,9 @@
 import * as plugins from './plugins.js';
 import { TypedRouter } from './classes.typedrouter.js';
 
+
+const closingBit: any = '#############CLOSING BIT#############';
+
 export interface ICommFunctions {
   sendMethod?: (
     sendPayload: plugins.typedRequestInterfaces.IStreamRequest
@@ -143,10 +146,17 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
 
   constructor() {}
 
+  workingDeferred: plugins.smartpromise.Deferred<void>;
+
   /**
    * takes care of sending 
    */
   private async workOnQueue() {
+    if (this.workingDeferred) {
+      return this.workingDeferred.promise;
+    } else {
+      this.workingDeferred = plugins.smartpromise.defer();
+    }
     if(this.side === 'requesting') {
       let thisSideIsBackpressured = !this.receiveBackpressuredArray.checkSpaceAvailable();
       let otherSideHasNext = false;
@@ -180,6 +190,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
       while (this.sendBackpressuredArray.data.length > 0 || otherSideHasNext) {
         if (otherSideIsBackpressured) {
           while (otherSideIsBackpressured) {
+            console.log('waiting for feedback because of backpressure...');
             await plugins.smartdelay.delayFor(50);
             await getFeedback();
           }
@@ -218,6 +229,8 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
       }
 
     }
+    this.workingDeferred.resolve();
+    this.workingDeferred = null;
   }
 
   /**
@@ -268,7 +281,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
           cycleId: streamTrArg.request.cycleId,
           cycle: 'response',
           mainPurpose: 'chunk',
-          next: this.sendBackpressuredArray.data.length > 1,
+          next: this.sendBackpressuredArray.data.length > 1, // 1 and not 0 because we call shift a few lines down
           backpressure: !this.receiveBackpressuredArray.checkSpaceAvailable(),
           chunkData: this.sendBackpressuredArray.shift(),
         };
@@ -382,18 +395,31 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
       streamIsDone = done;
     }
     if (closeAfterReading) {
-      await this.close();
+      await this.close(true);
     }
   }
 
   public async writeToWebstream(writableStreamArg: WritableStream<T>) {
     const writer = writableStreamArg.getWriter();
     while(this.keepAlive || this.receiveBackpressuredArray.checkHasItems()) {
-      await writer.write(await this.fetchData());
+      const value = await this.fetchData();
+      if (value === closingBit) {
+        writableStreamArg.close();
+        break;
+      }
+      await writer.write(value);
     }
   }
 
-  public async close() {
+  /**
+   * closes the stream
+   * if sendClosingBitArg is true, the stream will send a closing bit
+   * @param sendClosingBitArg 
+   */
+  public async close(sendClosingBitArg = false) {
+    if (sendClosingBitArg) {
+      this.sendData(closingBit);
+    }
     this.keepAlive = false;
   }
 }