3.0.6

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@api.global/typedrequest",
-  "version": "3.0.1",
+  "version": "3.0.6",
   "private": false,
   "description": "make typed requests towards apis",
   "main": "dist_ts/index.js",
@@ -14,22 +14,22 @@
     "buildDocs": "tsdoc"
   },
   "devDependencies": {
-    "@api.global/typedserver": "^3.0.1",
-    "@gitzone/tsbuild": "^2.1.66",
-    "@gitzone/tsbundle": "^2.0.8",
-    "@gitzone/tsrun": "^1.2.44",
-    "@gitzone/tstest": "^1.0.77",
-    "@push.rocks/smartenv": "^5.0.3",
-    "@push.rocks/tapbundle": "^5.0.12",
-    "@types/node": "^20.4.7"
+    "@api.global/typedserver": "^3.0.23",
+    "@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.19"
   },
   "dependencies": {
-    "@api.global/typedrequest-interfaces": "^3.0.1",
+    "@api.global/typedrequest-interfaces": "^3.0.6",
     "@push.rocks/isounique": "^1.0.5",
-    "@push.rocks/lik": "^6.0.3",
+    "@push.rocks/lik": "^6.0.12",
     "@push.rocks/smartdelay": "^3.0.5",
     "@push.rocks/smartpromise": "^4.0.3",
-    "@push.rocks/webrequest": "^3.0.32"
+    "@push.rocks/webrequest": "^3.0.34"
   },
   "files": [
     "ts/**/*",

test/test.ts

@@ -4,6 +4,7 @@ import * as typedserver from '@api.global/typedserver';
 import * as typedrequest from '../ts/index.js';
 
 let testServer: typedserver.servertools.Server;
+let testTypedRouter: typedrequest.TypedRouter;
 let testTypedHandler: typedrequest.TypedHandler<ITestReqRes>;
 
 // lets define an interface
@@ -17,6 +18,16 @@ interface ITestReqRes {
   };
 }
 
+interface ITestStream {
+  method: 'handleStream';
+  request: {
+    requestStream: any;
+  };
+  response: {
+    responseStream: any;
+  };
+}
+
 tap.test('should create a typedHandler', async () => {
   // lets use the interface in a TypedHandler
   testTypedHandler = new typedrequest.TypedHandler<ITestReqRes>('hi', async (reqArg) => {
@@ -35,7 +46,7 @@ tap.test('should spawn a server to test with', async () => {
 });
 
 tap.test('should define a testHandler', async () => {
-  const testTypedRouter = new typedrequest.TypedRouter(); // typed routers can broker typedrequests between handlers
+  testTypedRouter = new typedrequest.TypedRouter(); // typed routers can broker typedrequests between handlers
   testTypedRouter.addTypedHandler(testTypedHandler);
   testServer.addRoute(
     '/testroute',
@@ -60,6 +71,24 @@ tap.test('should fire a request', async () => {
   expect(response.surname).toEqual('wow');
 });
 
+tap.test('should allow VirtualStreams', async () => {
+  testTypedRouter.addTypedHandler(new typedrequest.TypedHandler<ITestStream>('handleStream', async (reqArg) => {
+    console.log('hey there');
+    console.log(reqArg.requestStream);
+    return {
+      responseStream: new typedrequest.VirtualStream(),
+    };
+  }));
+  const typedRequest = new typedrequest.TypedRequest<ITestStream>(
+    'http://localhost:3000/testroute',
+    'handleStream'
+  );
+  const response = await typedRequest.fire({
+    requestStream: new typedrequest.VirtualStream(),
+  });
+  console.log(response.responseStream);
+})
+
 tap.test('should end the server', async () => {
   await testServer.stop();
 });

ts/00_commitinfo_data.ts

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

ts/index.ts

@@ -3,3 +3,4 @@ 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';

ts/typedrequest.classes.typedhandler.ts

@@ -1,4 +1,4 @@
-import * as plugins from './typedrequest.plugins.js';
+import * as plugins from './plugins.js';
 import { TypedResponseError } from './typedrequest.classes.typedresponseerror.js';
 
 export type THandlerFunction<T extends plugins.typedRequestInterfaces.ITypedRequest> = (

ts/typedrequest.classes.typedrequest.ts

@@ -1,4 +1,5 @@
-import * as plugins from './typedrequest.plugins.js';
+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';
@@ -36,7 +37,7 @@ export class TypedRequest<T extends plugins.typedRequestInterfaces.ITypedRequest
    * fires the request
    */
   public async fire(fireArg: T['request'], useCacheArg: boolean = false): Promise<T['response']> {
-    const payload: plugins.typedRequestInterfaces.ITypedRequest = {
+    let payloadSending: plugins.typedRequestInterfaces.ITypedRequest = {
       method: this.method,
       request: fireArg,
       response: null,
@@ -46,30 +47,58 @@ export class TypedRequest<T extends plugins.typedRequestInterfaces.ITypedRequest
       },
     };
 
-    let responseBody: plugins.typedRequestInterfaces.ITypedRequest;
+    // lets preprocess the payload
+    payloadSending = VirtualStream.encodePayloadForNetwork(payloadSending, {
+      sendMethod: (payloadArg: plugins.typedRequestInterfaces.ITypedRequest) => {
+        return this.postTrObject(payloadArg);
+      }
+    });
+
+    let payloadReceiving: plugins.typedRequestInterfaces.ITypedRequest;
+    payloadReceiving = await this.postTrObject(payloadSending, useCacheArg);
+
+    // lets preprocess the response
+    payloadReceiving = VirtualStream.decodePayloadFromNetwork(payloadReceiving, {
+      sendMethod: (payloadArg: plugins.typedRequestInterfaces.ITypedRequest) => {
+        return this.postTrObject(payloadArg);
+      }
+    });
+    return payloadReceiving.response;
+  }
+
+  private async postTrObject(payloadSendingArg: plugins.typedRequestInterfaces.ITypedRequest, useCacheArg: boolean = false) {
+    let payloadReceiving: plugins.typedRequestInterfaces.ITypedRequest;
     if (this.urlEndPoint) {
-      const response = await webrequestInstance.postJson(this.urlEndPoint, payload, useCacheArg);
-      responseBody = response;
+      const response = await webrequestInstance.postJson(
+        this.urlEndPoint,
+        payloadSendingArg,
+        useCacheArg
+      );
+      payloadReceiving = response;
     } else {
-      responseBody = await this.typedTarget.post(payload);
+      payloadReceiving = await this.typedTarget.post(payloadSendingArg);
     }
-    if (responseBody.error) {
+    if (payloadReceiving.error) {
       console.error(
-        `Got an error ${responseBody.error.text} with data ${JSON.stringify(
-          responseBody.error.data
+        `Got an error ${payloadReceiving.error.text} with data ${JSON.stringify(
+          payloadReceiving.error.data,
+          null,
+          2
         )}`
       );
-      if (!responseBody.retry) {
-        throw new TypedResponseError(responseBody.error.text, responseBody.error.data);
+      if (!payloadReceiving.retry) {
+        throw new TypedResponseError(payloadReceiving.error.text, payloadReceiving.error.data);
       }
       return null;
     }
-    if (responseBody.retry) {
-      console.log(`server requested retry for the following reason: ${responseBody.retry.reason}`);
-      await plugins.smartdelay.delayFor(responseBody.retry.waitForMs);
+    if (payloadReceiving.retry) {
+      console.log(
+        `server requested retry for the following reason: ${payloadReceiving.retry.reason}`
+      );
+      await plugins.smartdelay.delayFor(payloadReceiving.retry.waitForMs);
       // tslint:disable-next-line: no-return-await
-      return await this.fire(fireArg);
+      payloadReceiving =  await this.postTrObject(payloadSendingArg, useCacheArg);
     }
-    return responseBody.response;
+    return payloadReceiving;
   }
 }

ts/typedrequest.classes.typedresponseerror.ts

@@ -1,4 +1,4 @@
-import * as plugins from './typedrequest.plugins.js';
+import * as plugins from './plugins.js';
 
 export class TypedResponseError {
   public errorText: string;

ts/typedrequest.classes.typedrouter.ts

@@ -1,4 +1,5 @@
-import * as plugins from './typedrequest.plugins.js';
+import * as plugins from './plugins.js';
+import { VirtualStream } from './typedrequest.classes.virtualstream.js';
 
 import { TypedHandler } from './typedrequest.classes.typedhandler.js';
 import { TypedRequest } from './typedrequest.classes.typedrequest.js';
@@ -13,6 +14,7 @@ export class TypedRouter {
   public handlerMap = new plugins.lik.ObjectMap<
     TypedHandler<any & plugins.typedRequestInterfaces.ITypedRequest>
   >();
+  public registeredVirtualStreams = new plugins.lik.ObjectMap<VirtualStream>();
 
   public fireEventInterestMap = new plugins.lik.InterestMap<
     string,
@@ -89,6 +91,22 @@ export class TypedRouter {
   public async routeAndAddResponse<
     T extends plugins.typedRequestInterfaces.ITypedRequest = plugins.typedRequestInterfaces.ITypedRequest
   >(typedRequestArg: T, localRequestArg = false): Promise<T> {
+    // decoding first
+    typedRequestArg = VirtualStream.decodePayloadFromNetwork(typedRequestArg, {
+      typedrouter: this,
+    });
+
+    // localdata second
+    typedRequestArg.localData = typedRequestArg.localData || {};
+    typedRequestArg.localData.firstTypedrouter = this;
+
+    // lets do stream processing
+    if (typedRequestArg.method === '##VirtualStream##') {
+      const result = await this.handleStreamTypedRequest(typedRequestArg as plugins.typedRequestInterfaces.IStreamRequest);
+      return result as T;
+    }
+
+    // lets do normal routing
     if (typedRequestArg?.correlation?.phase === 'request' || localRequestArg) {
       const typedHandler = this.getTypedHandlerForMethod(typedRequestArg.method);
 
@@ -99,10 +117,20 @@ export class TypedRouter {
           data: {},
         };
         typedRequestArg.correlation.phase = 'response';
+
+        // encode again before handing back
+        typedRequestArg = VirtualStream.encodePayloadForNetwork(typedRequestArg, {
+          typedrouter: this,
+        });
         return typedRequestArg;
       }
 
       typedRequestArg = await typedHandler.addResponse(typedRequestArg);
+      typedRequestArg.localData = null;
+      // encode again before handing back
+      typedRequestArg = VirtualStream.encodePayloadForNetwork(typedRequestArg, {
+        typedrouter: this,
+      });
       return typedRequestArg;
     } else if (typedRequestArg?.correlation?.phase === 'response') {
       this.fireEventInterestMap
@@ -115,4 +143,16 @@ export class TypedRouter {
       return null;
     }
   }
+
+  /**
+   * handle streaming
+   * @param streamTrArg 
+   */
+  public async handleStreamTypedRequest(streamTrArg: plugins.typedRequestInterfaces.IStreamRequest) {
+    const relevantVirtualStream = await this.registeredVirtualStreams.find(async virtualStreamArg => {
+      return virtualStreamArg.streamId === streamTrArg.request.streamId;
+    });
+    const result = await relevantVirtualStream.handleStreamTr(streamTrArg);
+    return result;
+  }
 }

ts/typedrequest.classes.typedtarget.ts

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

ts/typedrequest.classes.virtualstream.ts

@@ -0,0 +1,100 @@
+import * as plugins from './plugins.js';
+import type { TypedRouter } from './typedrequest.classes.typedrouter.js';
+
+export interface ICommFunctions {
+  sendMethod?: (
+    sendPayload: plugins.typedRequestInterfaces.ITypedRequest
+  ) => Promise<plugins.typedRequestInterfaces.ITypedRequest>;
+  typedrouter?: TypedRouter;
+}
+
+export class VirtualStream {
+  // STATIC
+  public static encodePayloadForNetwork(
+    objectPayload: any,
+    commFunctions: ICommFunctions
+  ): any {
+    if (objectPayload instanceof VirtualStream) {
+      if (!objectPayload.side && commFunctions.sendMethod) {
+        objectPayload.side = 'requesting';
+        objectPayload.sendMethod = commFunctions.sendMethod;
+      }
+      if (!objectPayload.side && commFunctions.typedrouter) {
+        objectPayload.side = 'responding';
+        objectPayload.typedrouter = commFunctions.typedrouter;
+      }
+      return {
+        _isVirtualStream: true,
+        streamId: objectPayload.streamId,
+      };
+    } else if (Array.isArray(objectPayload)) {
+      const returnArray = [];
+      for (const item of objectPayload) {
+        returnArray.push(VirtualStream.encodePayloadForNetwork(item, commFunctions));
+      }
+      return returnArray;
+    } else if (objectPayload !== null && typeof objectPayload === 'object') {
+      return Object.keys(objectPayload).reduce((acc, key) => {
+        acc[key] = VirtualStream.encodePayloadForNetwork(objectPayload[key], commFunctions);
+        return acc;
+      }, {});
+    } else {
+      return objectPayload;
+    }
+  }
+
+  public static decodePayloadFromNetwork(objectPayload: any, commFunctions: ICommFunctions): any {
+    if (objectPayload !== null && typeof objectPayload === 'object') {
+      if (objectPayload._isVirtualStream) {
+        const virtualStream = new VirtualStream();
+        virtualStream.streamId = objectPayload.streamId;
+        if (!virtualStream.side && commFunctions.sendMethod) {
+          virtualStream.side = 'requesting';
+          virtualStream.sendMethod = commFunctions.sendMethod;
+        }
+        if (!virtualStream.side && commFunctions.typedrouter) {
+          virtualStream.side = 'responding';
+          virtualStream.typedrouter = commFunctions.typedrouter;
+        }
+        return virtualStream;
+      } else if (Array.isArray(objectPayload)) {
+        const returnArray = [];
+        for (const item of objectPayload) {
+          returnArray.push(VirtualStream.decodePayloadFromNetwork(item, commFunctions));
+        }
+        return returnArray;
+      } else {
+        return Object.keys(objectPayload).reduce((acc, key) => {
+          acc[key] = VirtualStream.decodePayloadFromNetwork(objectPayload[key], commFunctions);
+          return acc;
+        }, {});
+      }
+    } else {
+      return objectPayload;
+    }
+  }
+
+  // INSTANCE
+
+  public side: 'requesting' | 'responding';
+  public streamId: string = plugins.isounique.uni();
+  public typedrouter: TypedRouter;
+  public sendMethod: (
+    sendPayloadArg: plugins.typedRequestInterfaces.ITypedRequest
+  ) => Promise<plugins.typedRequestInterfaces.ITypedRequest>;
+
+  constructor() {}
+
+  public async handleStreamTr(streamTrArg: plugins.typedRequestInterfaces.IStreamRequest) {
+    return streamTrArg;
+  }
+
+  /**
+   * closes the virtual stream
+   */
+  close() {
+    if (this.typedrouter) {
+      this.typedrouter.registeredVirtualStreams.remove(this);
+    }
+  }
+}

tsconfig.json

@@ -3,8 +3,12 @@
     "experimentalDecorators": true,
     "useDefineForClassFields": false,
     "target": "ES2022",
-    "module": "ES2022",
-    "moduleResolution": "nodenext",
-    "esModuleInterop": true
-  }
+    "module": "NodeNext",
+    "moduleResolution": "NodeNext",
+    "esModuleInterop": true,
+    "verbatimModuleSyntax": true
+  },
+  "exclude": [
+    "dist_*/**/*.d.ts"
+  ]
 }