3.0.11

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@api.global/typedrequest",
-  "version": "3.0.7",
+  "version": "3.0.11",
   "private": false,
   "description": "make typed requests towards apis",
   "main": "dist_ts/index.js",
@@ -14,22 +14,24 @@
     "buildDocs": "tsdoc"
   },
   "devDependencies": {
-    "@api.global/typedserver": "^3.0.23",
+    "@api.global/typedserver": "^3.0.25",
     "@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.14",
+    "@api.global/typedrequest-interfaces": "^3.0.18",
     "@push.rocks/isounique": "^1.0.5",
-    "@push.rocks/lik": "^6.0.13",
+    "@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/skip.ts

@@ -2,6 +2,7 @@ import { expect, tap } from '@push.rocks/tapbundle';
 import * as typedserver from '@api.global/typedserver';
 
 import * as typedrequest from '../ts/index.js';
+import * as typedrequestInterfaces from '@api.global/typedrequest-interfaces';
 
 let testServer: typedserver.servertools.Server;
 let testTypedRouter: typedrequest.TypedRouter;
@@ -21,10 +22,10 @@ interface ITestReqRes {
 interface ITestStream {
   method: 'handleStream';
   request: {
-    requestStream: any;
+    requestStream: typedrequestInterfaces.IVirtualStream;
   };
   response: {
-    responseStream: any;
+    responseStream: typedrequestInterfaces.IVirtualStream;
   };
 }
 
@@ -72,11 +73,16 @@ tap.test('should fire a request', async () => {
 });
 
 tap.test('should allow VirtualStreams', async () => {
+  const newRequestingVS = new typedrequest.VirtualStream();
+  const newRespondingVS = new typedrequest.VirtualStream();
+  let generatedRequestingVS: typedrequestInterfaces.IVirtualStream;
+  let generatedRespondingVS: typedrequestInterfaces.IVirtualStream;
   testTypedRouter.addTypedHandler(new typedrequest.TypedHandler<ITestStream>('handleStream', async (reqArg) => {
     console.log('hey there');
     console.log(reqArg.requestStream);
+    generatedRequestingVS = reqArg.requestStream;
     return {
-      responseStream: new typedrequest.VirtualStream(),
+      responseStream: newRespondingVS,
     };
   }));
   const typedRequest = new typedrequest.TypedRequest<ITestStream>(
@@ -84,10 +90,15 @@ tap.test('should allow VirtualStreams', async () => {
     'handleStream'
   );
   const response = await typedRequest.fire({
-    requestStream: new typedrequest.VirtualStream(),
+    requestStream: newRequestingVS,
   });
   console.log(response.responseStream);
-})
+
+  newRequestingVS.sendData(Buffer.from('hello'));
+  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);

test/test.browser.ts

@@ -31,9 +31,15 @@ tap.test('should define a testHandler', async () => {
 
 tap.test('should fire a request', async () => {
   const typedRequest = new typedrequest.TypedRequest<ITestReqRes>(
-    'http://localhost:3000/testroute',
+    'http://localhost:3000/typedrequest',
     'hi'
   );
+  const result = await typedRequest.fire({
+    name: 'yes',
+  }).catch(err => {
+    console.log(err);
+  });
+  console.log(result);
 });
 
 tap.start();

ts/00_commitinfo_data.ts

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

ts/plugins.ts

@@ -6,8 +6,10 @@ export { typedRequestInterfaces };
 // pushrocks scope
 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 smartpromise from '@push.rocks/smartpromise';
 import * as webrequest from '@push.rocks/webrequest';
+import * as webstream from '@push.rocks/webstream';
 
-export { isounique, lik, smartdelay, smartpromise, webrequest };
+export { isounique, lik, smartbuffer, smartdelay, smartpromise, webrequest, webstream };

ts/typedrequest.classes.virtualstream.ts

@@ -15,7 +15,7 @@ export interface ICommFunctions {
  * 3. It has a Readable and Writable side.
  * 4. The Writable side is Readable on the other side and vice versa.
  */
-export class VirtualStream<T = ArrayBufferLike> {
+export class VirtualStream<T = ArrayBufferLike> implements plugins.typedRequestInterfaces.IVirtualStream<T> {
   // STATIC
   public static encodePayloadForNetwork(
     objectPayload: any,
@@ -23,6 +23,12 @@ export class VirtualStream<T = ArrayBufferLike> {
     originalPayload?: any,
     path = []
   ): any {
+    if (!objectPayload) {
+      return objectPayload;
+    }
+    if (plugins.smartbuffer.isBufferLike(objectPayload)) {
+      return objectPayload;
+    }
     if (objectPayload instanceof VirtualStream) {
       if (!objectPayload.side && commFunctions.sendMethod) {
         objectPayload.side = 'requesting';
@@ -34,6 +40,7 @@ export class VirtualStream<T = ArrayBufferLike> {
         commFunctions.typedrouter.registeredVirtualStreams.add(objectPayload);
       }
       if (!originalPayload.response || path.includes('response')) {
+        objectPayload.startKeepAliveLoop();
         return {
           _isVirtualStream: true,
           streamId: objectPayload.streamId,
@@ -74,6 +81,9 @@ export class VirtualStream<T = ArrayBufferLike> {
   }
 
   public static decodePayloadFromNetwork(objectPayload: any, commFunctions: ICommFunctions): any {
+    if (plugins.smartbuffer.isBufferLike(objectPayload)) {
+      return objectPayload;
+    }
     if (objectPayload !== null && typeof objectPayload === 'object') {
       if (objectPayload._isVirtualStream) {
         const virtualStream = new VirtualStream();
@@ -87,6 +97,7 @@ export class VirtualStream<T = ArrayBufferLike> {
           virtualStream.typedrouter = commFunctions.typedrouter;
           commFunctions.typedrouter.registeredVirtualStreams.add(virtualStream);
         }
+        virtualStream.startKeepAliveLoop();
         return virtualStream;
       } else if (Array.isArray(objectPayload)) {
         const returnArray = [];
@@ -116,7 +127,7 @@ export class VirtualStream<T = ArrayBufferLike> {
 
   // wether to keep the stream alive
   private keepAlive = true;
-  private lastKeepAliveEvent = Date.now();
+  private lastKeepAliveEvent: number;
 
   // backpressured arrays
   private sendBackpressuredArray =
@@ -128,15 +139,17 @@ export class VirtualStream<T = ArrayBufferLike> {
       16
     );
 
-  constructor() {
-    this.startKeepAliveLoop();
-  }
+  constructor() {}
 
   /**
    * takes care of sending 
    */
   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({
@@ -159,29 +172,31 @@ export class VirtualStream<T = ArrayBufferLike> {
           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) {
-        const dataArg = this.sendBackpressuredArray.shift();
-        const streamTr = await this.sendMethod({
+        let dataArg: typeof this.sendBackpressuredArray.data[0];
+        if (this.sendBackpressuredArray.data.length > 0) {
+          dataArg = this.sendBackpressuredArray.shift();
+        }
+        let streamTr: plugins.typedRequestInterfaces.IStreamRequest;
+        streamTr = await this.sendMethod({
           method: '##VirtualStream##',
           request: {
             streamId: this.streamId,
             cycleId: plugins.isounique.uni(),
             cycle: 'request',
-            mainPurpose: 'chunk',
+            mainPurpose: dataArg ? 'chunk' : 'read',
             backpressure: thisSideIsBackpressured,
             next: this.sendBackpressuredArray.data.length > 0,
-            chunkData: dataArg,
+            ...dataArg ? { chunkData: dataArg } : {},
           },
           response: null,
         }).catch(() => {
           console.log('stream ended immaturely');
           this.keepAlive = false;
+          return null;
         });
 
         if (streamTr && streamTr.response && streamTr.response.chunkData) {
@@ -232,7 +247,6 @@ export class VirtualStream<T = ArrayBufferLike> {
         next: this.sendBackpressuredArray.data.length > 0,
         backpressure: this.receiveBackpressuredArray.checkSpaceAvailable(),
       };
-      streamTrArg.request = null;
     }
 
     // chunk handling
@@ -264,6 +278,7 @@ export class VirtualStream<T = ArrayBufferLike> {
     return streamTrArg;
   }
 
+  // lifecycle methods
   /**
    * closes the virtual stream
    */
@@ -328,6 +343,7 @@ export class VirtualStream<T = ArrayBufferLike> {
     this.workOnQueue();
     await this.sendBackpressuredArray.waitForSpace();
   }
+
   public async fetchData(): Promise<T> {
     if (this.receiveBackpressuredArray.hasSpace) {
       // do something maybe?
@@ -337,10 +353,26 @@ export class VirtualStream<T = ArrayBufferLike> {
     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());
+    }
   }
 }