Better http channel (#7)

* add flag to http endpoint to await request handled

* add default reponse code

* implement feedback (rename res to resolve)

* bump alpha version

Author: ajuvercr

channels/http.ttl

@@ -25,6 +25,18 @@
     sh:maxCount 1;
     sh:datatype xsd:boolean;
     sh:description "Stream raw bytes if true";
+  ], [
+    sh:path :waitHandled;
+    sh:name "waitHandled";
+    sh:maxCount 1;
+    sh:datatype xsd:boolean;
+    sh:description "Wait for handlers to be have handled the incoming message";
+  ], [
+    sh:path :responseCode;
+    sh:name "responseCode";
+    sh:maxCount 1;
+    sh:datatype xsd:integer;
+    sh:description "Specify the expected response code (default 200)";
   ].
 
 [ ] a sh:NodeShape;

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@ajuvercr/js-runner",
-  "version": "0.1.16",
+  "version": "0.1.18-alpha.0",
   "type": "module",
   "exports": {
     "import": "./dist/index.js",

src/connectors/http.ts

@@ -15,15 +15,18 @@ import {
   WriterConstructor,
 } from "../connectors";
 
-function streamToString(stream: Readable, binary: boolean): Promise<string | Buffer> {
+function streamToString(
+  stream: Readable,
+  binary: boolean,
+): Promise<string | Buffer> {
   const datas = <Buffer[]>[];
   return new Promise((res) => {
     stream.on("data", (data) => {
       datas.push(data);
     });
     stream.on("end", () => {
       const streamData = Buffer.concat(datas);
-      res(binary ? streamData : streamData.toString())
+      res(binary ? streamData : streamData.toString());
     });
   });
 }
@@ -32,6 +35,8 @@ export interface HttpReaderConfig extends Config {
   endpoint: string;
   port: number;
   binary: boolean;
+  waitHandled?: boolean;
+  responseCode?: number;
 }
 
 export const startHttpStreamReader: ReaderConstructor<HttpReaderConfig> = (
@@ -42,11 +47,12 @@ export const startHttpStreamReader: ReaderConstructor<HttpReaderConfig> = (
   const stream = new SimpleStream<string | Buffer>(
     () =>
       new Promise((res) => {
-        const cb = (): void => res();
         if (server !== undefined) {
-          server.close(cb);
+          server.close(() => {
+            res();
+          });
         } else {
-          cb();
+          res();
         }
       }),
   );
@@ -57,20 +63,23 @@ export const startHttpStreamReader: ReaderConstructor<HttpReaderConfig> = (
   ) {
     try {
       const content = await streamToString(req, config.binary);
-      stream.push(content).catch((error) => {
+
+      const promise = stream.push(content).catch((error) => {
         throw error;
       });
+      if (config.waitHandled) {
+        await promise;
+      }
     } catch (error: unknown) {
       console.error("Failed", error);
     }
 
-    res.writeHead(200);
+    res.writeHead(config.responseCode || 200);
     res.end("OK");
   };
 
   server = createServer(requestListener);
   const init = () => {
-    console.log("HTTP init!");
     return new Promise<void>((res) => {
       const cb = (): void => res(undefined);
       if (server) {
@@ -94,23 +103,25 @@ export const startHttpStreamWriter: WriterConstructor<HttpWriterConfig> = (
   const requestConfig = <https.RequestOptions>new URL(config.endpoint);
 
   const push = async (item: string | Buffer): Promise<void> => {
-    await new Promise((res) => {
+    await new Promise(async (resolve) => {
       const options = {
         hostname: requestConfig.hostname,
         path: requestConfig.path,
         method: config.method,
         port: requestConfig.port,
       };
+
       const cb = (response: IncomingMessage): void => {
         response.on("data", () => {});
         response.on("end", () => {
-          res(null);
+          resolve(null);
         });
       };
 
       const req = http.request(options, cb);
-      req.write(item, () => res(null));
-      req.end();
+      await new Promise((res) => req.write(item, res));
+      await new Promise((res) => req.end(res));
+      // res(null);
     });
   };
 

test/connectors/http.test.ts

@@ -19,7 +19,7 @@ describe("connector-http", () => {
     const factory = new conn.ChannelFactory();
     const reader = factory.createReader(readerConfig);
     const writer = factory.createWriter(writerConfig);
-    
+
     reader.data((data) => {
       items.push(data);
     });
@@ -43,6 +43,7 @@ describe("connector-http", () => {
       endpoint: "localhost",
       port: 8081,
       binary: true,
+      waitHandled: false,
       ty: conn.Conn.HttpReaderChannel,
     };
     const writerConfig: HttpWriterConfig = {
@@ -54,8 +55,9 @@ describe("connector-http", () => {
     const factory = new conn.ChannelFactory();
     const reader = factory.createReader(readerConfig);
     const writer = factory.createWriter(writerConfig);
-    
+
     reader.data((data) => {
+      console.log("This reader works");
       expect(Buffer.isBuffer(data)).toBeTruthy();
       items.push(data.toString());
     });
@@ -73,6 +75,76 @@ describe("connector-http", () => {
 
     await Promise.all([reader.end(), writer.end()]);
   });
+
+  test("Should write -> HTTP -> read (Buffer) and await response", async () => {
+    const readerConfig: HttpReaderConfig = {
+      endpoint: "localhost",
+      port: 8082,
+      binary: true,
+      waitHandled: true,
+      ty: conn.Conn.HttpReaderChannel,
+    };
+    const writerConfig: HttpWriterConfig = {
+      endpoint: "http://localhost:8082",
+      method: "POST",
+      ty: conn.Conn.HttpWriterChannel,
+    };
+
+    const factory = new conn.ChannelFactory();
+    const reader = factory.createReader(readerConfig);
+    const writer = factory.createWriter(writerConfig);
+
+    reader.data(async (data) => {
+      expect(Buffer.isBuffer(data)).toBeTruthy();
+      items.push(data.toString());
+      await sleep(1500);
+    });
+
+    await factory.init();
+
+    const items: unknown[] = [];
+
+    const start = new Date().getTime();
+    await writer.push(Buffer.from("test1", "utf8"));
+    const end = new Date().getTime();
+    await sleep(200);
+
+    expect(end - start > 1000).toBeTruthy();
+    expect(items).toEqual(["test1"]);
+
+    await Promise.all([reader.end(), writer.end()]);
+  });
+
+  test("http channel uses correct response code", async () => {
+    const readerConfig: HttpReaderConfig = {
+      endpoint: "localhost",
+      port: 8083,
+      binary: false,
+      responseCode: 202,
+      ty: conn.Conn.HttpReaderChannel,
+    };
+
+    const factory = new conn.ChannelFactory();
+    const reader = factory.createReader(readerConfig);
+
+    reader.data((data) => {
+      items.push(data);
+    });
+
+    await factory.init();
+
+    const items: unknown[] = [];
+
+    const resp = await fetch("http://localhost:8083", {
+      body: "test1",
+      method: "PUT",
+    });
+
+    expect(items).toEqual(["test1"]);
+    expect(resp.status).toEqual(202);
+
+    await Promise.all([reader.end()]);
+  });
 });
 
 function sleep(x: number): Promise<unknown> {