3.0.18

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@api.global/typedrequest",
-  "version": "3.0.8",
+  "version": "3.0.18",
   "private": false,
   "description": "make typed requests towards apis",
   "main": "dist_ts/index.js",
@@ -14,23 +14,24 @@
     "buildDocs": "tsdoc"
   },
   "devDependencies": {
-    "@api.global/typedserver": "^3.0.24",
+    "@api.global/typedserver": "^3.0.26",
     "@git.zone/tsbuild": "^2.1.72",
     "@git.zone/tsbundle": "^2.0.15",
     "@git.zone/tsrun": "^1.2.44",
     "@git.zone/tstest": "^1.0.86",
     "@push.rocks/smartenv": "^5.0.12",
     "@push.rocks/tapbundle": "^5.0.15",
-    "@types/node": "^20.11.20"
+    "@types/node": "^20.11.24"
   },
   "dependencies": {
-    "@api.global/typedrequest-interfaces": "^3.0.17",
+    "@api.global/typedrequest-interfaces": "^3.0.18",
     "@push.rocks/isounique": "^1.0.5",
-    "@push.rocks/lik": "^6.0.13",
-    "@push.rocks/smartbuffer": "^1.0.6",
+    "@push.rocks/lik": "^6.0.14",
+    "@push.rocks/smartbuffer": "^1.0.7",
     "@push.rocks/smartdelay": "^3.0.5",
     "@push.rocks/smartpromise": "^4.0.3",
-    "@push.rocks/webrequest": "^3.0.34"
+    "@push.rocks/webrequest": "^3.0.34",
+    "@push.rocks/webstream": "^1.0.8"
   },
   "files": [
     "ts/**/*",

test/test.browser.ts

@@ -30,10 +30,18 @@ tap.test('should define a testHandler', async () => {
 });
 
 tap.test('should fire a request', async () => {
-  const typedRequest = new typedrequest.TypedRequest<ITestReqRes>(
-    'http://localhost:3000/testroute',
-    'hi'
-  );
+  let response = await fetch('http://localhost:3000/typedrequest', {
+    "method": "POST",
+    "headers": {
+      "Content-Type": "application/json"
+    },
+    "body": "{\"correlation\":{\"id\":\"uni_aefe56c1a0f61a3e91082209\",\"phase\":\"request\"},\"method\":\"hi\",\"request\":{\"name\":\"yes\"},\"response\":null}"
+  })
+  console.log(await response.text());
 });
 
+tap.test('test', async (tools) => {
+  await tools.delayFor(5000);
+})
+
 tap.start();

test/test.ts

@@ -98,11 +98,12 @@ tap.test('should allow VirtualStreams', async () => {
   const data = await generatedRequestingVS.fetchData();
   const decodedData = data.toString();
   expect(data.toString()).toEqual('hello');
-})
+});
 
 tap.test('should end the server', async (toolsArg) => {
-  await toolsArg.delayFor(5000);
+  await toolsArg.delayFor(1000);
   await testServer.stop();
+  setTimeout(() => process.exit(0), 100);
 });
 
 tap.start();

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@api.global/typedrequest',
-  version: '3.0.8',
+  version: '3.0.18',
   description: 'make typed requests towards apis'
 }

ts/plugins.ts

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

ts/typedrequest.classes.typedrouter.ts

@@ -120,6 +120,7 @@ export class TypedRouter {
         typedRequestArg.correlation.phase = 'response';
 
         // encode again before handing back
+        typedRequestArg.localData = null;
         typedRequestArg = VirtualStream.encodePayloadForNetwork(typedRequestArg, {
           typedrouter: this,
         });

ts/typedrequest.classes.virtualstream.ts

@@ -23,6 +23,9 @@ export class VirtualStream<T = ArrayBufferLike> implements plugins.typedRequestI
     originalPayload?: any,
     path = []
   ): any {
+    if (!objectPayload) {
+      return objectPayload;
+    }
     if (plugins.smartbuffer.isBufferLike(objectPayload)) {
       return objectPayload;
     }
@@ -49,26 +52,24 @@ export class VirtualStream<T = ArrayBufferLike> implements plugins.typedRequestI
         };
       }
     } else if (Array.isArray(objectPayload)) {
-      const returnArray = [];
-      for (const item of objectPayload) {
-        returnArray.push(
-          VirtualStream.encodePayloadForNetwork(
-            item,
-            commFunctions,
-            originalPayload || objectPayload,
-            path
-          )
-        );
-      }
-      return returnArray;
-    } else if (objectPayload !== null && typeof objectPayload === 'object') {
-      return Object.keys(objectPayload).reduce((acc, key) => {
-        path.push(key);
-        acc[key] = VirtualStream.encodePayloadForNetwork(
-          objectPayload[key],
+      // For arrays, we recurse over each item.
+      return objectPayload.map((item, index) => 
+        VirtualStream.encodePayloadForNetwork(
+          item,
           commFunctions,
           originalPayload || objectPayload,
-          path
+          path.concat(String(index)) // Convert index to string and concatenate to path
+        )
+      );
+    } else if (objectPayload !== null && typeof objectPayload === 'object') {
+      // For objects, we recurse over each key-value pair.
+      return Object.entries(objectPayload).reduce((acc, [key, value]) => {
+        const newPath = path.concat(key); // Concatenate the new key to the path
+        acc[key] = VirtualStream.encodePayloadForNetwork(
+          value,
+          commFunctions,
+          originalPayload || objectPayload,
+          newPath
         );
         return acc;
       }, {});
@@ -143,6 +144,10 @@ export class VirtualStream<T = ArrayBufferLike> implements plugins.typedRequestI
    */
   private async workOnQueue() {
     if(this.side === 'requesting') {
+      let thisSideIsBackpressured = !this.receiveBackpressuredArray.checkSpaceAvailable();
+      let otherSideHasNext = false;
+      let otherSideIsBackpressured = false;
+      
       // helper functions
       const getFeedback = async () => {
         const streamTr = await this.sendMethod({
@@ -165,12 +170,9 @@ export class VirtualStream<T = ArrayBufferLike> implements plugins.typedRequestI
           otherSideHasNext = streamTr.response.next;
         }
       }
-
+      await getFeedback();
 
       // do work loop
-      let thisSideIsBackpressured = !this.receiveBackpressuredArray.checkSpaceAvailable();
-      let otherSideHasNext = false;
-      let otherSideIsBackpressured = false;
       while (this.sendBackpressuredArray.data.length > 0 || otherSideHasNext) {
         let dataArg: typeof this.sendBackpressuredArray.data[0];
         if (this.sendBackpressuredArray.data.length > 0) {
@@ -243,7 +245,6 @@ export class VirtualStream<T = ArrayBufferLike> implements plugins.typedRequestI
         next: this.sendBackpressuredArray.data.length > 0,
         backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
       };
-      streamTrArg.request = null;
     }
 
     // chunk handling
@@ -350,10 +351,26 @@ export class VirtualStream<T = ArrayBufferLike> implements plugins.typedRequestI
     return dataPackage;
   }
 
-  public pipeWebStream(webStream: any) {
-    // lets do the piping
-    webStream.on('data', (data: any) => {});
-    webStream.on('end', () => {});
-    webStream.on('error', (error: any) => {});
+  /**
+   * reads from a Readable and sends it to the other side
+   * @param readableStreamArg
+   */
+  public async readFromWebstream(readableStreamArg: ReadableStream<T>) {
+    const reader = readableStreamArg.getReader();
+    let streamIsDone = false;
+    while(!streamIsDone) {
+      const { value, done } = await reader.read();
+      if(value) {
+        await this.sendData(value);
+      }
+      streamIsDone = done;
+    }
+  }
+
+  public async writeToWebstream(writableStreamArg: WritableStream<T>) {
+    const writer = writableStreamArg.getWriter();
+    while(this.keepAlive) {
+      await writer.write(await this.fetchData());
+    }
   }
 }