3.1.10

changelog.md

@@ -1,5 +1,70 @@
 # Changelog
 
+## 2024-10-16 - 3.1.10 - fix(VirtualStream)
+Fix stream closure logic in `writeToWebstream` method
+
+- Added `writer.releaseLock()` call before closing WritableStream when `closingBit` is received in `writeToWebstream` method.
+
+## 2024-10-16 - 3.1.9 - fix(VirtualStream)
+Ensure writable streams are correctly closed asynchronously to prevent potential sync issues.
+
+- Updated VirtualStream to use 'await' when closing writable streams, ensuring proper asynchronous handling.
+
+## 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
+
+- Fixed backpressure flag assignment in the response handling logic of VirtualStream.
+- Ensured correct negation logic for checking receive backpressure status.
+
+## 2024-10-14 - 3.1.3 - fix(VirtualStream)
+Fix keepAlive flag handling in VirtualStream and added stream closure in tests
+
+- Ensure that the keepAlive status is correctly maintained in the keepAlive trigger method.
+- Added closure of VirtualStreams in the test suite for proper resource cleanup.
+
+## 2024-10-14 - 3.1.2 - fix(core)
+Fix incorrect backpressure logic in VirtualStream class
+
+- Corrected the logic for determining backpressure status by checking the available space in the receiveBackpressuredArray.
+- Introduced a looping mechanism to wait when the other side is backpressured before sending more data.
+
+## 2024-10-14 - 3.1.1 - fix(virtualstream)
+Fix handling of virtual streams for proper shutdown
+
+- Ensured that writeToWebstream method checks for remaining items in receiveBackpressuredArray before closing.
+- Corrected package.json dependency for @push.rocks/tapbundle.
+- Updated @types/node to version 22.7.5.
+
+## 2024-10-11 - 3.1.0 - feat(virtualstream)
+Enhance VirtualStream with optional closure when reading from webstream
+
+- Added an optional parameter `closeAfterReading` to the `readFromWebstream` method.
+- The stream will close automatically after reading if `closeAfterReading` is set to true.
+
 ## 2024-10-11 - 3.0.33 - fix(test)
 Increase delay duration before stopping the server in test suite.
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@api.global/typedrequest",
-  "version": "3.0.33",
+  "version": "3.1.10",
   "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",
@@ -20,13 +20,13 @@
     "@git.zone/tsrun": "^1.2.49",
     "@git.zone/tstest": "^1.0.90",
     "@push.rocks/smartenv": "^5.0.12",
-    "@push.rocks/tapbundle": "^5.0.24",
-    "@types/node": "^22.5.4"
+    "@push.rocks/tapbundle": "^5.3.0",
+    "@types/node": "^22.7.5"
   },
   "dependencies": {
     "@api.global/typedrequest-interfaces": "^3.0.19",
     "@push.rocks/isounique": "^1.0.5",
-    "@push.rocks/lik": "^6.0.15",
+    "@push.rocks/lik": "^6.1.0",
     "@push.rocks/smartbuffer": "^3.0.4",
     "@push.rocks/smartdelay": "^3.0.5",
     "@push.rocks/smartguard": "^3.1.0",

test/test.ts

@@ -98,7 +98,8 @@ tap.test('should allow VirtualStreams', async () => {
   const data = await generatedRequestingVS.fetchData();
   const decodedData = new TextDecoder().decode(data);
   expect(decodedData).toEqual('hello');
-  // await newRequestingVS.close();
+  await newRequestingVS.close();
+  await newRespondingVS.close();
 });
 
 tap.test('should end the server', async (toolsArg) => {

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@api.global/typedrequest',
-  version: '3.0.33',
+  version: '3.1.10',
   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;
@@ -162,7 +172,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
             cycle: 'request',
             mainPurpose: 'feedback',
             next: this.sendBackpressuredArray.data.length > 0,
-            backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
+            backpressure: !this.receiveBackpressuredArray.checkSpaceAvailable(),
           },
           response: null,
         }).catch(() => {
@@ -178,6 +188,13 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
 
       // do work loop
       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();
+          }
+        }
         let dataArg: typeof this.sendBackpressuredArray.data[0];
         if (this.sendBackpressuredArray.data.length > 0) {
           dataArg = this.sendBackpressuredArray.shift();
@@ -204,13 +221,16 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
         if (streamTr && streamTr.response && streamTr.response.chunkData) {
           this.receiveBackpressuredArray.push(streamTr.response.chunkData);
         }
-        thisSideIsBackpressured = this.receiveBackpressuredArray.checkSpaceAvailable();
+        otherSideIsBackpressured = streamTr && streamTr.response && streamTr.response.backpressure;
+        thisSideIsBackpressured = !this.receiveBackpressuredArray.checkSpaceAvailable();
 
         // lets care about looping
         otherSideHasNext = streamTr && streamTr.response && streamTr.response.next;
       }
 
     }
+    this.workingDeferred.resolve();
+    this.workingDeferred = null;
   }
 
   /**
@@ -235,7 +255,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
         mainPurpose: 'keepAlive',
         keepAlive: this.keepAlive,
         next: this.sendBackpressuredArray.data.length > 0,
-        backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
+        backpressure: !this.receiveBackpressuredArray.checkSpaceAvailable(),
       };
     }
 
@@ -247,7 +267,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
         cycle: 'response',
         mainPurpose: 'feedback',
         next: this.sendBackpressuredArray.data.length > 0,
-        backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
+        backpressure: !this.receiveBackpressuredArray.checkSpaceAvailable(),
       };
     }
 
@@ -261,8 +281,8 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
           cycleId: streamTrArg.request.cycleId,
           cycle: 'response',
           mainPurpose: 'chunk',
-          next: this.sendBackpressuredArray.data.length > 1,
-          backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
+          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(),
         };
       } else {
@@ -272,6 +292,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
           cycle: 'response',
           mainPurpose: 'feedback',
           next: this.sendBackpressuredArray.data.length > 0,
+          backpressure: !this.receiveBackpressuredArray.checkSpaceAvailable(),
         };
       }
       streamTrArg.request = null;
@@ -320,7 +341,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
           cycleId: plugins.isounique.uni(),
           cycle: 'request',
           mainPurpose: 'keepAlive',
-          keepAlive: true,
+          keepAlive: this.keepAlive,
         },
         response: null,
       }).catch(() => {
@@ -363,7 +384,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
    * reads from a Readable and sends it to the other side
    * @param readableStreamArg
    */
-  public async readFromWebstream(readableStreamArg: ReadableStream<T>) {
+  public async readFromWebstream(readableStreamArg: ReadableStream<T>, closeAfterReading = true) {
     const reader = readableStreamArg.getReader();
     let streamIsDone = false;
     while(!streamIsDone) {
@@ -373,16 +394,33 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
       }
       streamIsDone = done;
     }
+    if (closeAfterReading) {
+      await this.close(true);
+    }
   }
 
   public async writeToWebstream(writableStreamArg: WritableStream<T>) {
     const writer = writableStreamArg.getWriter();
-    while(this.keepAlive) {
-      await writer.write(await this.fetchData());
+    while(this.keepAlive || this.receiveBackpressuredArray.checkHasItems()) {
+      const value = await this.fetchData();
+      if (value === closingBit) {
+        writer.releaseLock();
+        await 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;
   }
 }