sdk/nodejs: add body() function with default limit of 64kb for readin…

…g body, and catch errors and return them as json
lithdew · Jun 15, 2020 · fb5b21c · fb5b21c
1 parent 4f34ace
commit fb5b21c
Show file tree

Hide file tree

Showing 2 changed files with 64 additions and 3 deletions.
diff --git a/nodejs/src/flatend.ts b/nodejs/src/flatend.ts
@@ -55,6 +55,7 @@ export class Context extends Duplex {
     }
 
     json(data: any) {
+        this.header('content-type', 'application/json');
         this.send(JSON.stringify(data));
     }
 
@@ -84,6 +85,50 @@ export class Context extends Duplex {
         callback();
     }
 
+    // @ts-ignore
+    async body(opts: { limit: number } = {limit: 65536}): Promise<Buffer> {
+        return await (new Promise((resolve, reject) => {
+            let buffer: Buffer[] = [];
+            let received: number = 0;
+            let complete: boolean = false;
+
+            const done = (err?: Error) => {
+                if (complete) return;
+
+                complete = true;
+                if (!err) {
+                    resolve(Buffer.concat(buffer));
+                } else {
+                    reject(err);
+                }
+            }
+
+            const onData = (data: Buffer) => {
+                if (complete) return;
+                received += data.byteLength;
+                if (received > opts.limit) {
+                    done(new Error(`request entity too large`));
+                    return;
+                }
+                buffer.push(data);
+            };
+
+            const onEnd = (err: Error) => done(err);
+
+            const onClose = () => {
+                this.removeListener('data', onData);
+                this.removeListener('error', onEnd);
+                this.removeListener('end', onEnd);
+                this.removeListener('close', onClose);
+            }
+
+            this.on('data', onData);
+            this.on('error', onEnd);
+            this.on('end', onEnd);
+            this.on('close', onClose);
+        }));
+    }
+
     _read(size: number) {
         return;
     }
@@ -207,6 +252,8 @@ export class Node {
         body = body.slice(1);
 
         switch (opcode) {
+            case Opcode.ServiceResponse:
+                break;
             case Opcode.Handshake: {
                 const packet = HandshakePacket.decode(body)[0];
                 if (packet.id && packet.signature) {
@@ -233,7 +280,14 @@ export class Node {
                 provider.streams.set(packet.id, ctx);
 
                 const handler = this.#handlers.get(service)!;
-                handler(ctx);
+
+                (async () => {
+                    try {
+                        await handler(ctx);
+                    } catch (err) {
+                        ctx.json({error: err.message});
+                    }
+                })()
 
                 return;
             }

diff --git a/nodejs/src/index.ts b/nodejs/src/index.ts
@@ -3,9 +3,16 @@ import * as fs from "fs";
 
 async function main() {
     const node = new Node();
-    node.register("get_todos", (ctx: Context) => ctx.json(ctx.headers));
+    node.register("get_todos", (ctx: Context) => {
+        ctx.json(ctx.headers)
+    });
+
     node.register("all_todos", (ctx: Context) => fs.createReadStream("tsconfig.json").pipe(ctx));
-    node.register('pipe', (ctx: Context) => ctx.pipe(ctx));
+
+    node.register('pipe', async (ctx: Context) => {
+        const body = await ctx.body({limit: 65536});
+        ctx.json(JSON.parse(body.toString("utf8")));
+    });
     await node.dial("0.0.0.0:9000");
 }