3.1.2

changelog.md

@@ -1,5 +1,18 @@
 # Changelog
 
+## 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
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@api.global/typedrequest",
-  "version": "3.1.0",
+  "version": "3.1.2",
   "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",

ts/00_commitinfo_data.ts

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

ts/classes.virtualstream.ts

@@ -162,7 +162,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 +178,12 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
 
       // do work loop
       while (this.sendBackpressuredArray.data.length > 0 || otherSideHasNext) {
+        if (otherSideIsBackpressured) {
+          while (otherSideIsBackpressured) {
+            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,7 +210,8 @@ 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;
@@ -235,7 +242,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 +254,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(),
       };
     }
 
@@ -380,7 +387,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
 
   public async writeToWebstream(writableStreamArg: WritableStream<T>) {
     const writer = writableStreamArg.getWriter();
-    while(this.keepAlive) {
+    while(this.keepAlive || this.receiveBackpressuredArray.checkHasItems()) {
       await writer.write(await this.fetchData());
     }
   }