INTEGRATION [PR#1858 > development/8.1] ARSN-183 type check stream

index.ts

@@ -28,6 +28,7 @@ export * as constants from './lib/constants';
 export * as https from './lib/https';
 export * as metrics from './lib/metrics';
 export * as network from './lib/network';
+export * as stream from './lib/stream';
 
 export const db = require('./lib/db');
 export const errorUtils = require('./lib/errorUtils');
@@ -189,10 +190,6 @@ export const pensieve = {
     credentialUtils: require('./lib/executables/pensieveCreds/utils'),
 };
 
-export const stream = {
-    readJSONStreamObject: require('./lib/stream/readJSONStreamObject'),
-};
-
 export const patches = {
     locationConstraints: require('./lib/patches/locationConstraints'),
 };

lib/stream/index.ts

@@ -0,0 +1 @@
+export { default as readJSONStreamObject } from './readJSONStreamObject';

lib/stream/readJSONStreamObject.ts

@@ -0,0 +1,42 @@
+import errors from '../errors';
+import * as stream from 'stream';
+import joi from 'joi';
+
+/**
+ * read a JSON object from a stream returned as a javascript object,
+ * handle errors.
+ *
+ * @param s - Readable stream
+ * @param [joiSchema] - optional validation schema for the JSON object
+ * @return a Promise resolved with the parsed JSON object as a result
+ */
+export default async function readJSONStreamObject<Data = any>(
+    s: stream.Readable,
+    joiSchema?: joi.Schema<Data>
+): Promise<Data> {
+    return new Promise((resolve, reject) => {
+        const contentsChunks: any = [];
+        s.on('data', (chunk) => contentsChunks.push(chunk));
+        s.on('end', () => {
+            const contents = contentsChunks.join('');
+            try {
+                const parsedContents = JSON.parse(contents);
+                if (joiSchema) {
+                    const { error, value } = joiSchema.validate(parsedContents);
+                    if (error) {
+                        throw error;
+                    }
+                    return resolve(value);
+                }
+                return resolve(parsedContents);
+            } catch (err: any) {
+                return reject(
+                    errors.InvalidArgument.customizeDescription(
+                        `invalid input: ${err.message}`
+                    )
+                );
+            }
+        });
+        s.once('error', reject);
+    });
+}

tests/unit/stream/readJSONStreamObject.spec.js

@@ -1,7 +1,7 @@
 const assert = require('assert');
 const stream = require('stream');
 const joi = require('joi');
-const readJSONStreamObject = require('../../../lib/stream/readJSONStreamObject');
+const readJSONStreamObject = require('../../../lib/stream/readJSONStreamObject').default;
 
 class ReqStream extends stream.Readable {
     constructor(contents) {