v3.1.11

changelog.md

@@ -1,5 +1,53 @@
 # Changelog
 
+## 2025-12-03 - 3.1.11 - fix(virtualstream)
+Expose transport localData to handlers via TypedTools; improve VirtualStream payload encode/decode to preserve built-ins and handle nested arrays/objects
+
+- TypedHandler: pass typedRequest.localData into a TypedTools instance so handlers can access transport-layer context (e.g. websocket peer).
+- TypedTools: add a public localData property to store transport-specific context available to handlers.
+- VirtualStream.decodePayloadFromNetwork: preserve built-in objects (Set, Map, Date, RegExp, Error, Promise or thenable) to avoid incorrect transformation.
+- VirtualStream.encodePayloadForNetwork / decodePayloadFromNetwork: added proper recursion for arrays and objects to correctly handle nested payloads and virtual streams, with path tracking to support deduplication logic.
+
+## 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
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@api.global/typedrequest",
-  "version": "3.1.3",
+  "version": "3.1.11",
   "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.3',
+  version: '3.1.11',
   description: 'A TypeScript library for making typed requests towards APIs, including facilities for handling requests, routing, and virtual stream handling.'
 }

ts/classes.typedhandler.ts

@@ -31,6 +31,10 @@ export class TypedHandler<T extends plugins.typedRequestInterfaces.ITypedRequest
     }
     let typedResponseError: TypedResponseError;
     const typedtoolsInstance = new TypedTools();
+    // Pass localData from the request to TypedTools so handlers can access transport-layer context
+    if (typedRequestArg.localData) {
+      typedtoolsInstance.localData = typedRequestArg.localData;
+    }
     const response = await this.handlerFunction(typedRequestArg.request, typedtoolsInstance).catch((e) => {
       if (e instanceof TypedResponseError) {
         typedResponseError = e;

ts/classes.typedtools.ts

@@ -2,6 +2,12 @@ import { TypedResponseError } from './classes.typedresponseerror.js';
 import * as plugins from './plugins.js';
 
 export class TypedTools {
+  /**
+   * Local data passed from the transport layer.
+   * This can contain connection-specific context like the WebSocket peer.
+   */
+  public localData: Record<string, any> = {};
+
   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);

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
@@ -79,7 +82,7 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
   }
 
   public static decodePayloadFromNetwork(objectPayload: any, commFunctions: ICommFunctions): any {
-    
+
     if (
       plugins.smartbuffer.isBufferLike(objectPayload)
       || objectPayload instanceof TypedRouter
@@ -87,6 +90,18 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
       return objectPayload;
     }
     if (objectPayload !== null && typeof objectPayload === 'object') {
+      // Preserve built-in objects that shouldn't be transformed
+      if (
+        objectPayload instanceof Set ||
+        objectPayload instanceof Map ||
+        objectPayload instanceof Date ||
+        objectPayload instanceof RegExp ||
+        objectPayload instanceof Error ||
+        objectPayload instanceof Promise ||
+        typeof objectPayload.then === 'function'
+      ) {
+        return objectPayload;
+      }
       if (objectPayload._isVirtualStream) {
         const virtualStream = new VirtualStream();
         virtualStream.streamId = objectPayload.streamId;
@@ -143,10 +158,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 +202,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 +241,8 @@ export class VirtualStream<T = Uint8Array> implements plugins.typedRequestInterf
       }
 
     }
+    this.workingDeferred.resolve();
+    this.workingDeferred = null;
   }
 
   /**
@@ -268,8 +293,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 {
@@ -279,6 +304,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;
@@ -381,18 +407,32 @@ 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) {
+        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;
   }
 }