v4.0.0

changelog.md

@@ -0,0 +1,134 @@
+# Changelog
+
+## 2025-12-04 - 4.0.0 - BREAKING CHANGE(typedrouter)
+Introduce options object for TypedRouter.routeAndAddResponse (localRequest, skipHooks); add defaultRouteOptions and make hook calls respect skipHooks; bump package version to 3.2.3
+
+- Changed TypedRouter.routeAndAddResponse signature to accept an options object ({ localRequest?: boolean; skipHooks?: boolean }) instead of a boolean second argument — this is a breaking API change.
+- Added TypedRouter.defaultRouteOptions with defaults { localRequest: false, skipHooks: false }.
+- Routing now respects options.localRequest and options.skipHooks; hook calls (onIncomingRequest, onOutgoingResponse, onIncomingResponse) are skipped when skipHooks is true to avoid hook recursion or duplicate handling (useful for broadcast-received messages).
+- Bumped package.json version to 3.2.3.
+
+## 2025-12-04 - 3.2.2 - fix(typedrequest)
+Add skipHooks flag to TypedRequest to optionally suppress global hooks for internal requests
+
+- Introduce public skipHooks boolean on TypedRequest (default false) with documentation comment explaining it should be used for internal/logging requests to prevent infinite loops.
+- Guard calls to global hooks (onOutgoingRequest and onIncomingResponse) in TypedRequest.fire() so hooks are not invoked when skipHooks is true.
+
+## 2025-12-04 - 3.2.1 - fix(typedrouter)
+Use globalThis-backed globalHooks for TypedRouter to enable cross-bundle sharing; fix merging and clearing of global hooks.
+
+- Replace static globalHooks field with getter/setter that stores hooks on globalThis so hooks are shared across bundles.
+- Fix setGlobalHooks to merge new hooks with existing ones (avoiding accidental overwrite).
+- Update clearGlobalHooks to clear the globalThis storage used for hooks.
+
+## 2025-12-04 - 3.2.0 - feat(typedrouter)
+Add request/response hooks and monitoring to TypedRouter; emit hooks from TypedRequest; improve VirtualStream encoding/decoding; re-export hook types
+
+- Introduce ITypedRequestLogEntry and ITypedRouterHooks interfaces to represent structured traffic log entries and hook callbacks.
+- Add static globalHooks on TypedRouter with helper APIs TypedRouter.setGlobalHooks and TypedRouter.clearGlobalHooks for global traffic monitoring.
+- Add instance-level hooks on TypedRouter (setHooks) and a unified callHook() that invokes both global and instance hooks safely (errors are caught and logged).
+- TypedRequest now emits onOutgoingRequest before sending and onIncomingResponse after receiving, including timestamps, duration and payload/error details.
+- TypedRouter now emits lifecycle hooks while routing: onIncomingRequest when a request arrives, onOutgoingResponse for responses (both success and handler-missing error cases), and onIncomingResponse when responses arrive to be fulfilled.
+- VirtualStream.encodePayloadForNetwork and decodePayloadFromNetwork were enhanced to recurse into arrays and nested objects (preserving special built-ins) to correctly handle embedded virtual streams.
+- Re-export ITypedRequestLogEntry and ITypedRouterHooks from the package index for external consumption.
+
+## 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
+
+- 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.29",
+  "version": "4.0.0",
   "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,23 +14,23 @@
     "buildDocs": "tsdoc"
   },
   "devDependencies": {
-    "@api.global/typedserver": "^3.0.50",
-    "@git.zone/tsbuild": "^2.1.80",
+    "@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.13"
+    "@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.0.2",
-    "@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.10"
   },

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();
 });
 
 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.29',
+  version: '4.0.0',
   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.typedrequest.ts

@@ -1,11 +1,25 @@
 import * as plugins from './plugins.js';
 import { VirtualStream } from './classes.virtualstream.js';
 import { TypedResponseError } from './classes.typedresponseerror.js';
-import { TypedRouter } from './classes.typedrouter.js';
+import { TypedRouter, type ITypedRequestLogEntry } from './classes.typedrouter.js';
 import { TypedTarget } from './classes.typedtarget.js';
 
 const webrequestInstance = new plugins.webrequest.WebRequest();
 
+/**
+ * Helper to call global hooks from TypedRequest
+ */
+function callGlobalHook(
+  hookName: keyof typeof TypedRouter.globalHooks,
+  entry: ITypedRequestLogEntry
+): void {
+  try {
+    TypedRouter.globalHooks[hookName]?.(entry);
+  } catch (err) {
+    console.error(`TypedRequest hook error (${hookName}):`, err);
+  }
+}
+
 export class TypedRequest<T extends plugins.typedRequestInterfaces.ITypedRequest> {
   /**
    * in case we post against a url endpoint
@@ -19,6 +33,12 @@ export class TypedRequest<T extends plugins.typedRequestInterfaces.ITypedRequest
 
   public method: string;
 
+  /**
+   * When true, hooks will not be called for this request.
+   * Use this for internal/logging requests to prevent infinite loops.
+   */
+  public skipHooks: boolean = false;
+
   /**
    * @param postEndPointArg
    * @param methodArg
@@ -36,6 +56,8 @@ export class TypedRequest<T extends plugins.typedRequestInterfaces.ITypedRequest
    * fires the request
    */
   public async fire(fireArg: T['request'], useCacheArg: boolean = false): Promise<T['response']> {
+    const requestStartTime = Date.now();
+
     let payloadSending: plugins.typedRequestInterfaces.ITypedRequest = {
       method: this.method,
       request: fireArg,
@@ -53,6 +75,18 @@ export class TypedRequest<T extends plugins.typedRequestInterfaces.ITypedRequest
       }
     });
 
+    // Hook: outgoing request (skip if this is an internal request)
+    if (!this.skipHooks) {
+      callGlobalHook('onOutgoingRequest', {
+        correlationId: payloadSending.correlation.id,
+        method: this.method,
+        direction: 'outgoing',
+        phase: 'request',
+        timestamp: requestStartTime,
+        payload: fireArg,
+      });
+    }
+
     let payloadReceiving: plugins.typedRequestInterfaces.ITypedRequest;
     payloadReceiving = await this.postTrObject(payloadSending, useCacheArg);
 
@@ -62,6 +96,21 @@ export class TypedRequest<T extends plugins.typedRequestInterfaces.ITypedRequest
         return this.postTrObject(payloadArg) as Promise<plugins.typedRequestInterfaces.IStreamRequest>;
       }
     });
+
+    // Hook: incoming response (skip if this is an internal request)
+    if (!this.skipHooks) {
+      callGlobalHook('onIncomingResponse', {
+        correlationId: payloadSending.correlation.id,
+        method: this.method,
+        direction: 'incoming',
+        phase: 'response',
+        timestamp: Date.now(),
+        durationMs: Date.now() - requestStartTime,
+        payload: payloadReceiving?.response,
+        error: payloadReceiving?.error?.text,
+      });
+    }
+
     return payloadReceiving.response;
   }
 

ts/classes.typedrouter.ts

@@ -4,12 +4,91 @@ import { VirtualStream } from './classes.virtualstream.js';
 import { TypedHandler } from './classes.typedhandler.js';
 import { TypedRequest } from './classes.typedrequest.js';
 
+/**
+ * Log entry for TypedRequest traffic monitoring
+ */
+export interface ITypedRequestLogEntry {
+  correlationId: string;
+  method: string;
+  direction: 'outgoing' | 'incoming';
+  phase: 'request' | 'response';
+  timestamp: number;
+  durationMs?: number;
+  payload: any;
+  error?: string;
+}
+
+/**
+ * Hooks for intercepting TypedRequest traffic
+ */
+export interface ITypedRouterHooks {
+  onOutgoingRequest?: (entry: ITypedRequestLogEntry) => void;
+  onIncomingResponse?: (entry: ITypedRequestLogEntry) => void;
+  onIncomingRequest?: (entry: ITypedRequestLogEntry) => void;
+  onOutgoingResponse?: (entry: ITypedRequestLogEntry) => void;
+}
+
 /**
  * A typed router decides on which typed handler to call based on the method
  * specified in the typed request
  * This is thought for reusing the same url endpoint for different methods
  */
 export class TypedRouter {
+  // Use globalThis for cross-bundle hook sharing
+  public static get globalHooks(): ITypedRouterHooks {
+    if (!(globalThis as any).__typedRouterGlobalHooks) {
+      (globalThis as any).__typedRouterGlobalHooks = {};
+    }
+    return (globalThis as any).__typedRouterGlobalHooks;
+  }
+
+  public static set globalHooks(value: ITypedRouterHooks) {
+    (globalThis as any).__typedRouterGlobalHooks = value;
+  }
+
+  /**
+   * Set global hooks for monitoring all TypedRequest traffic
+   * Hooks are shared across all bundles via globalThis
+   */
+  public static setGlobalHooks(hooks: ITypedRouterHooks): void {
+    const current = TypedRouter.globalHooks;
+    TypedRouter.globalHooks = { ...current, ...hooks };
+  }
+
+  /**
+   * Clear all global hooks
+   */
+  public static clearGlobalHooks(): void {
+    (globalThis as any).__typedRouterGlobalHooks = {};
+  }
+
+  // Instance-level hooks (for per-router monitoring)
+  public hooks: ITypedRouterHooks = {};
+
+  /**
+   * Set instance-level hooks for monitoring traffic through this router
+   */
+  public setHooks(hooks: ITypedRouterHooks): void {
+    this.hooks = { ...this.hooks, ...hooks };
+  }
+
+  /**
+   * Helper to call both global and instance hooks
+   */
+  private callHook(
+    hookName: keyof ITypedRouterHooks,
+    entry: ITypedRequestLogEntry
+  ): void {
+    try {
+      // Call global hooks
+      TypedRouter.globalHooks[hookName]?.(entry);
+      // Call instance hooks
+      this.hooks[hookName]?.(entry);
+    } catch (err) {
+      console.error(`TypedRouter hook error (${hookName}):`, err);
+    }
+  }
+
   public routerMap = new plugins.lik.ObjectMap<TypedRouter>();
   public handlerMap = new plugins.lik.ObjectMap<
     TypedHandler<any & plugins.typedRequestInterfaces.ITypedRequest>
@@ -83,14 +162,26 @@ export class TypedRouter {
     return typedHandler;
   }
 
+  /**
+   * Options for routeAndAddResponse
+   */
+  public static defaultRouteOptions = {
+    localRequest: false,
+    skipHooks: false,
+  };
+
   /**
    * if typedrequest object has correlation.phase === 'request' -> routes a typed request object to a handler
    * if typedrequest object has correlation.phase === 'response' -> routes a typed request object to request fire event
    * @param typedRequestArg
+   * @param optionsArg - Options object with:
+   *   - localRequest: treat as local request (default: false)
+   *   - skipHooks: skip calling hooks for this routing (default: false, use for broadcast-received messages)
    */
   public async routeAndAddResponse<
     T extends plugins.typedRequestInterfaces.ITypedRequest = plugins.typedRequestInterfaces.ITypedRequest
-  >(typedRequestArg: T, localRequestArg = false): Promise<T> {
+  >(typedRequestArg: T, optionsArg: { localRequest?: boolean; skipHooks?: boolean } = {}): Promise<T> {
+    const options = { ...TypedRouter.defaultRouteOptions, ...optionsArg };
     // decoding first
     typedRequestArg = VirtualStream.decodePayloadFromNetwork(typedRequestArg, {
       typedrouter: this,
@@ -108,7 +199,21 @@ export class TypedRouter {
     }
 
     // lets do normal routing
-    if (typedRequestArg?.correlation?.phase === 'request' || localRequestArg) {
+    if (typedRequestArg?.correlation?.phase === 'request' || options.localRequest) {
+      const requestStartTime = Date.now();
+
+      // Hook: incoming request (skip if routing broadcast-received messages)
+      if (!options.skipHooks) {
+        this.callHook('onIncomingRequest', {
+          correlationId: typedRequestArg.correlation?.id || 'unknown',
+          method: typedRequestArg.method,
+          direction: 'incoming',
+          phase: 'request',
+          timestamp: requestStartTime,
+          payload: typedRequestArg.request,
+        });
+      }
+
       const typedHandler = this.getTypedHandlerForMethod(typedRequestArg.method);
 
       if (!typedHandler) {
@@ -124,6 +229,21 @@ export class TypedRouter {
         typedRequestArg = VirtualStream.encodePayloadForNetwork(typedRequestArg, {
           typedrouter: this,
         });
+
+        // Hook: outgoing response (error - no handler)
+        if (!options.skipHooks) {
+          this.callHook('onOutgoingResponse', {
+            correlationId: typedRequestArg.correlation?.id || 'unknown',
+            method: typedRequestArg.method,
+            direction: 'outgoing',
+            phase: 'response',
+            timestamp: Date.now(),
+            durationMs: Date.now() - requestStartTime,
+            payload: typedRequestArg.response,
+            error: typedRequestArg.error?.text,
+          });
+        }
+
         return typedRequestArg;
       }
 
@@ -133,8 +253,36 @@ export class TypedRouter {
       typedRequestArg = VirtualStream.encodePayloadForNetwork(typedRequestArg, {
         typedrouter: this,
       });
+
+      // Hook: outgoing response (success)
+      if (!options.skipHooks) {
+        this.callHook('onOutgoingResponse', {
+          correlationId: typedRequestArg.correlation?.id || 'unknown',
+          method: typedRequestArg.method,
+          direction: 'outgoing',
+          phase: 'response',
+          timestamp: Date.now(),
+          durationMs: Date.now() - requestStartTime,
+          payload: typedRequestArg.response,
+          error: typedRequestArg.error?.text,
+        });
+      }
+
       return typedRequestArg;
     } else if (typedRequestArg?.correlation?.phase === 'response') {
+      // Hook: incoming response
+      if (!options.skipHooks) {
+        this.callHook('onIncomingResponse', {
+          correlationId: typedRequestArg.correlation?.id || 'unknown',
+          method: typedRequestArg.method,
+          direction: 'incoming',
+          phase: 'response',
+          timestamp: Date.now(),
+          payload: typedRequestArg.response,
+          error: typedRequestArg.error?.text,
+        });
+      }
+
       this.fireEventInterestMap
         .findInterest(typedRequestArg.correlation.id)
         ?.fullfillInterest(typedRequestArg);

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;
@@ -162,7 +184,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 +200,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 +233,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 +267,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 +279,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 +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 {
@@ -272,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;
@@ -295,10 +328,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);
@@ -316,7 +353,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(() => {
@@ -359,7 +396,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) {
@@ -369,12 +406,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);
     }
   }
+
+  /**
+   * 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;
+  }
 }

ts/index.ts

@@ -3,4 +3,7 @@ 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';
+export * from './classes.virtualstream.js';
+
+// Re-export hook interfaces from typedrouter
+export type { ITypedRequestLogEntry, ITypedRouterHooks } from './classes.typedrouter.js';