Make the client API stream response data instead of waiting

Author: pimterry

src/api/api-model.ts

@@ -1,4 +1,5 @@
 import * as _ from 'lodash';
+import * as stream from 'stream';
 import * as os from 'os';
 
 import { generateSPKIFingerprint } from 'mockttp';
@@ -195,10 +196,10 @@ export class ApiModel {
         return { success: !interceptor.isActive(proxyPort) };
     }
 
-    async sendRequest(
+    sendRequest(
         requestDefinition: Client.RequestDefinition,
         requestOptions: Client.RequestOptions
-    ): Promise<Client.ResponseDefinition> {
+    ): stream.Readable {
         return Client.sendRequest(requestDefinition, requestOptions);
     }
 

src/api/rest-api.ts

@@ -88,20 +88,47 @@ export function exposeRestAPI(
         if (!request) throw new StatusError(400, "No request definition provided");
         if (!options) throw new StatusError(400, "No request options provided");
 
-        const result = await apiModel.sendRequest({
+        const resultStream = apiModel.sendRequest({
             ...request,
             // Body buffers are serialized as base64 (for both requests & responses)
             rawBody: Buffer.from(request.rawBody ?? '', 'base64')
         }, {
             ...options
         });
 
-        res.send({
-            type: 'response', // Later we may send ND-JSON with chunked data & other types
-            response: {
-                ...result,
-                rawBody: result.rawBody?.toString('base64') ?? ''
+        res.writeHead(200, {
+            'content-type': 'application/x-ndjson'
+        });
+
+        resultStream.on('data', (evt: Client.ResponseStreamEvents) => {
+            if (evt.type === 'response-body-part') {
+                res.write(JSON.stringify({
+                    ...evt,
+                    data: evt.data.toString('base64')
+                }));
+            } else {
+                res.write(JSON.stringify(evt));
             }
+            res.write('\n');
+        });
+
+        resultStream.on('end', () => {
+            res.write(JSON.stringify({
+                'type': 'response-end'
+            }) + '\n');
+            res.end();
+        });
+
+        resultStream.on('error', (error: ErrorLike) => {
+            res.write(JSON.stringify({
+                type: 'error',
+                error: {
+                    code: error.code,
+                    message: error.message,
+                    stack: error.stack
+                }
+            }) + '\n');
+            res.end();
         });
     }));
 }

src/client/client.ts

@@ -1,3 +1,4 @@
+import * as stream from 'stream';
 import * as net from 'net';
 import * as http from 'http';
 import * as https from 'https';
@@ -24,17 +25,27 @@ export interface RequestDefinition {
 export interface RequestOptions {
 }
 
-export interface ResponseDefinition {
+export type ResponseStreamEvents =
+    | ResponseHead
+    | ResponseBodyPart;
+// Other notable events: errors (via 'error' event) and clean closure (via 'end').
+
+export interface ResponseHead {
+    type: 'response-head';
     statusCode: number;
     statusMessage?: string;
     headers: RawHeaders;
-    rawBody?: Buffer;
 }
 
-export async function sendRequest(
+export interface ResponseBodyPart {
+    type: 'response-body-part';
+    data: Buffer;
+}
+
+export function sendRequest(
     requestDefn: RequestDefinition,
     options: RequestOptions
-): Promise<ResponseDefinition> {
+): stream.Readable {
     const url = new URL(requestDefn.url);
 
     const request = (url.protocol === 'https' ? https : http).request(requestDefn.url, {
@@ -60,19 +71,34 @@ export async function sendRequest(
         request.end();
     }
 
-    const response = await new Promise<http.IncomingMessage>((resolve, reject) => {
+    const resultsStream = new stream.Readable({
+        objectMode: true,
+        read() {} // Can't pull data - we manually fill this with .push() instead.
+    });
+
+    new Promise<http.IncomingMessage>((resolve, reject) => {
         request.on('error', reject);
         request.on('response', resolve);
+    }).then((response) => {
+        resultsStream.push({
+            type: 'response-head',
+            statusCode: response.statusCode!,
+            statusMessage: response.statusMessage,
+            headers: pairFlatRawHeaders(response.rawHeaders)
+        });
+
+        response.on('data', (data) => resultsStream.push({
+            type: 'response-body-part',
+            data
+        }));
+
+        response.on('end', () => resultsStream.push(null));
+    }).catch((error) => {
+        resultsStream.destroy(error);
+        request.destroy();
     });
 
-    const body = await streamToBuffer(response);
-
-    return {
-        statusCode: response.statusCode!,
-        statusMessage: response.statusMessage,
-        headers: pairFlatRawHeaders(response.rawHeaders),
-        rawBody: body
-    }
+    return resultsStream;
 }
 
 /**

src/util/stream.ts

@@ -1,10 +1,15 @@
 import * as stream from 'stream';
 
-export function streamToBuffer(input: stream.Readable) {
-    return new Promise<Buffer>((resolve, reject) => {
-        const chunks: Uint8Array[] = [];
+export function streamToArray<T extends unknown>(input: stream.Readable) {
+    return new Promise<T[]>((resolve, reject) => {
+        const chunks: T[] = [];
         input.on('data', (d) => chunks.push(d));
-        input.on('end', () => resolve(Buffer.concat(chunks)));
+        input.on('end', () => resolve(chunks));
         input.on('error', reject);
     });
+};
+
+export async function streamToBuffer(input: stream.Readable) {
+    const chunks = await streamToArray<Uint8Array>(input);
+    return Buffer.concat(chunks);
 };

test/client/send-request.spec.ts

@@ -1,7 +1,8 @@
 import { expect } from 'chai';
 import * as mockttp from 'mockttp';
 
-import { sendRequest } from '../../src/client/client';
+import { ResponseStreamEvents, sendRequest } from '../../src/client/client';
+import { streamToArray } from '../../src/util/stream';
 
 describe("The HTTP client API", () => {
 
@@ -29,7 +30,7 @@ describe("The HTTP client API", () => {
             };
         });
 
-        const response = await sendRequest({
+        const responseStream = sendRequest({
             url: mockServer.urlFor('/path?qwe=asd'),
             method: 'POST',
             headers: [
@@ -40,11 +41,19 @@ describe("The HTTP client API", () => {
             rawBody: Buffer.from('Request body')
         }, {});
 
-        expect(response.statusCode).to.equal(200);
-        expect(response.statusMessage).to.equal('Custom status message');
-        expect(response.headers).to.deep.equal([
-            ['custom-HEADER', 'custom-VALUE']
-        ]);
-        expect(response.rawBody!.toString()).to.equal('Mock response body');
+        const responseParts = await streamToArray<any>(responseStream);
+
+        expect(responseParts.length).to.equal(2);
+        expect(responseParts[0]).to.deep.equal({
+            type: 'response-head',
+            statusCode: 200,
+            statusMessage: 'Custom status message',
+            headers: [
+                ['custom-HEADER', 'custom-VALUE']
+            ]
+        });
+
+        expect(responseParts[1].type).equal('response-body-part');
+        expect(responseParts[1].data.toString()).to.equal('Mock response body');
     })
 });