3.1.4

changelog.md

@@ -0,0 +1,60 @@
+# Changelog
+
+## 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.
+
+- Adjusted the delay time from 1000 ms to 10000 ms before stopping the server to ensure tests complete smoothly.
+
+## 2024-09-06 - 3.0.32 - fix(virtualstream)
+Fix keep-alive loop handling and test cleanup
+
+- Prevent unnecessary keep-alive loop from starting on the responding side
+- Add logging for keep-alive loop initiation in VirtualStream
+- Temporarily comment out stream close and tap forceful stop in test to avoid abrupt termination
+
+## 2024-09-06 - 3.0.31 - fix(core)
+Updated dependencies and added close method to VirtualStream
+
+- Updated dependencies in package.json for better compatibility
+- Added close method to VirtualStream class in ts/classes.virtualstream.ts for more graceful stream termination
+
+## 2024-05-31 - 3.0.28 - Error Handling
+Enhancement to error handling mechanisms.
+
+- Logs now include the method to which an error was given.
+
+## 2023-08-04 - 3.0.0 - Core
+Introduced a breaking change.
+
+- Major update to core functionalities.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@api.global/typedrequest",
-  "version": "3.0.23",
+  "version": "3.1.4",
   "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",
@@ -14,24 +14,25 @@
     "buildDocs": "tsdoc"
   },
   "devDependencies": {
-    "@api.global/typedserver": "^3.0.29",
-    "@git.zone/tsbuild": "^2.1.72",
+    "@api.global/typedserver": "^3.0.51",
+    "@git.zone/tsbuild": "^2.1.84",
     "@git.zone/tsbundle": "^2.0.15",
-    "@git.zone/tsrun": "^1.2.44",
+    "@git.zone/tsrun": "^1.2.49",
     "@git.zone/tstest": "^1.0.90",
     "@push.rocks/smartenv": "^5.0.12",
-    "@push.rocks/tapbundle": "^5.0.23",
-    "@types/node": "^20.12.8"
+    "@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/smartpromise": "^4.0.3",
+    "@push.rocks/smartguard": "^3.1.0",
+    "@push.rocks/smartpromise": "^4.0.4",
     "@push.rocks/webrequest": "^3.0.37",
-    "@push.rocks/webstream": "^1.0.8"
+    "@push.rocks/webstream": "^1.0.10"
   },
   "files": [
     "ts/**/*",

test/test.browser.ts

@@ -44,4 +44,4 @@ tap.skip.test('test', async (tools) => {
   await tools.delayFor(5000);
 })
 
-tap.start();
+export default tap.start();

test/test.ts

@@ -98,12 +98,14 @@ 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 newRespondingVS.close();
 });
 
 tap.test('should end the server', async (toolsArg) => {
-  await toolsArg.delayFor(1000);
+  await toolsArg.delayFor(10000);
   await testServer.stop();
-  setTimeout(() => process.exit(0), 100);
+  await tap.stopForcefully();
 });
 
-tap.start();
+export default tap.start();

ts/00_commitinfo_data.ts

@@ -1,8 +1,8 @@
 /**
- * autocreated commitinfo by @pushrocks/commitinfo
+ * autocreated commitinfo by @push.rocks/commitinfo
  */
 export const commitinfo = {
   name: '@api.global/typedrequest',
-  version: '3.0.23',
+  version: '3.1.4',
   description: 'A TypeScript library for making typed requests towards APIs, including facilities for handling requests, routing, and virtual stream handling.'
 }

ts/classes.typedhandler.ts

@@ -1,8 +1,10 @@
 import * as plugins from './plugins.js';
-import { TypedResponseError } from './typedrequest.classes.typedresponseerror.js';
+import { TypedResponseError } from './classes.typedresponseerror.js';
+import { TypedTools } from './classes.typedtools.js';
 
 export type THandlerFunction<T extends plugins.typedRequestInterfaces.ITypedRequest> = (
-  requestArg: T['request']
+  requestArg: T['request'],
+  typedToolsArg?: TypedTools
 ) => Promise<T['response']>;
 
 /**
@@ -28,7 +30,8 @@ export class TypedHandler<T extends plugins.typedRequestInterfaces.ITypedRequest
       );
     }
     let typedResponseError: TypedResponseError;
-    const response = await this.handlerFunction(typedRequestArg.request).catch((e) => {
+    const typedtoolsInstance = new TypedTools();
+    const response = await this.handlerFunction(typedRequestArg.request, typedtoolsInstance).catch((e) => {
       if (e instanceof TypedResponseError) {
         typedResponseError = e;
       } else {

ts/classes.typedrequest.ts

@@ -1,8 +1,8 @@
 import * as plugins from './plugins.js';
-import { VirtualStream } from './typedrequest.classes.virtualstream.js';
-import { TypedResponseError } from './typedrequest.classes.typedresponseerror.js';
-import { TypedRouter } from './typedrequest.classes.typedrouter.js';
-import { TypedTarget } from './typedrequest.classes.typedtarget.js';
+import { VirtualStream } from './classes.virtualstream.js';
+import { TypedResponseError } from './classes.typedresponseerror.js';
+import { TypedRouter } from './classes.typedrouter.js';
+import { TypedTarget } from './classes.typedtarget.js';
 
 const webrequestInstance = new plugins.webrequest.WebRequest();
 
@@ -79,7 +79,7 @@ export class TypedRequest<T extends plugins.typedRequestInterfaces.ITypedRequest
     }
     if (payloadReceiving.error) {
       console.error(
-        `Got an error ${payloadReceiving.error.text} with data ${JSON.stringify(
+        `method: >>${this.method}<< got an ERROR: "${payloadReceiving.error.text}" with data ${JSON.stringify(
           payloadReceiving.error.data,
           null,
           2

ts/classes.typedrouter.ts

@@ -1,8 +1,8 @@
 import * as plugins from './plugins.js';
-import { VirtualStream } from './typedrequest.classes.virtualstream.js';
+import { VirtualStream } from './classes.virtualstream.js';
 
-import { TypedHandler } from './typedrequest.classes.typedhandler.js';
-import { TypedRequest } from './typedrequest.classes.typedrequest.js';
+import { TypedHandler } from './classes.typedhandler.js';
+import { TypedRequest } from './classes.typedrequest.js';
 
 /**
  * A typed router decides on which typed handler to call based on the method

ts/classes.typedtarget.ts

@@ -1,4 +1,4 @@
-import { TypedRouter } from './typedrequest.classes.typedrouter.js';
+import { TypedRouter } from './classes.typedrouter.js';
 import * as plugins from './plugins.js';
 
 export type IPostMethod = (

ts/classes.typedtools.ts

@@ -0,0 +1,13 @@
+import { TypedResponseError } from './classes.typedresponseerror.js';
+import * as plugins from './plugins.js';
+
+export class TypedTools {
+  public async passGuards<T = any>(guardsArg: plugins.smartguard.Guard<T>[], dataArg: T) {
+    const guardSet = new plugins.smartguard.GuardSet<T>(guardsArg);
+    const guardResult = await guardSet.allGuardsPass(dataArg);
+    if (!guardResult) {
+      const failedHint = await guardSet.getFailedHint(dataArg);
+      throw new TypedResponseError(`guard failed: ${failedHint}`, { failedHint });
+    }
+  }
+}

ts/classes.virtualstream.ts

@@ -1,5 +1,5 @@
 import * as plugins from './plugins.js';
-import type { TypedRouter } from './typedrequest.classes.typedrouter.js';
+import { TypedRouter } from './classes.typedrouter.js';
 
 export interface ICommFunctions {
   sendMethod?: (
@@ -79,7 +79,11 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
   }
 
   public static decodePayloadFromNetwork(objectPayload: any, commFunctions: ICommFunctions): any {
-    if (plugins.smartbuffer.isBufferLike(objectPayload)) {
+    
+    if (
+      plugins.smartbuffer.isBufferLike(objectPayload)
+      || objectPayload instanceof TypedRouter
+    ) {
       return objectPayload;
     }
     if (objectPayload !== null && typeof objectPayload === 'object') {
@@ -158,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(() => {
@@ -174,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();
@@ -200,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;
@@ -231,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(),
       };
     }
 
@@ -243,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(),
       };
     }
 
@@ -258,7 +269,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
           cycle: 'response',
           mainPurpose: 'chunk',
           next: this.sendBackpressuredArray.data.length > 1,
-          backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
+          backpressure: !this.receiveBackpressuredArray.checkSpaceAvailable(),
           chunkData: this.sendBackpressuredArray.shift(),
         };
       } else {
@@ -268,6 +279,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;
@@ -291,10 +303,14 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
    */
   private async startKeepAliveLoop() {
     // initially wait for a second
+    if (this.side === 'responding') {
+      return;
+    }
     await plugins.smartdelay.delayFor(0);
+    console.log(`starting keepalive loop on side ${this.side}`);
     let counter = 0;
     keepAliveLoop: while (this.keepAlive) {
-      const triggerResult = await this.triggerKeepAlive();
+      await this.triggerKeepAlive();
       await plugins.smartdelay.delayFor(1000);
     }
     await plugins.smartdelay.delayFor(1000);
@@ -312,7 +328,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(() => {
@@ -355,7 +371,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) {
@@ -365,12 +381,19 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
       }
       streamIsDone = done;
     }
+    if (closeAfterReading) {
+      await this.close();
+    }
   }
 
   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());
     }
   }
+
+  public async close() {
+    this.keepAlive = false;
+  }
 }

ts/index.ts

@@ -1,6 +1,6 @@
-export * from './typedrequest.classes.typedrequest.js';
-export * from './typedrequest.classes.typedhandler.js';
-export * from './typedrequest.classes.typedrouter.js';
-export * from './typedrequest.classes.typedresponseerror.js';
-export * from './typedrequest.classes.typedtarget.js';
-export * from './typedrequest.classes.virtualstream.js';
+export * from './classes.typedrequest.js';
+export * from './classes.typedhandler.js';
+export * from './classes.typedrouter.js';
+export * from './classes.typedresponseerror.js';
+export * from './classes.typedtarget.js';
+export * from './classes.virtualstream.js';

ts/plugins.ts

@@ -8,8 +8,8 @@ import * as isounique from '@push.rocks/isounique';
 import * as lik from '@push.rocks/lik';
 import * as smartbuffer from '@push.rocks/smartbuffer';
 import * as smartdelay from '@push.rocks/smartdelay';
+import * as smartguard from '@push.rocks/smartguard';
 import * as smartpromise from '@push.rocks/smartpromise';
 import * as webrequest from '@push.rocks/webrequest';
 
-
-export { isounique, lik, smartbuffer, smartdelay, smartpromise, webrequest };
+export { isounique, lik, smartbuffer, smartdelay, smartguard, smartpromise, webrequest };