refactor streaming logic for simplified datasource

Author: tysonrm

src/adapters/controllers/get-models.js

@@ -20,7 +20,7 @@ export default function getModelsFactory (listModels) {
         httpRequest.stream = true
         return
       }
-      
+
       const { content, contentType } = getContent(httpRequest, models)
 
       return {

src/adapters/datasources/datasource-mongodb.js

@@ -1,7 +1,7 @@
 'use strict'
 
-import CircuitBreaker from '../../domain/circuit-breaker'
-import DataSource from '../../domain/datasource'
+const CircuitBreaker = require('../../domain/circuit-breaker').default
+const DataSource = require('../../domain/datasource').default
 
 const HIGHWATERMARK = 50
 
@@ -55,6 +55,23 @@ export class DataSourceMongoDb extends DataSource {
     }
   }
 
+  async connectionPool () {
+    return new Promise((resolve, reject) => {
+      if (this.db) return resolve(this.db)
+      MongoClient.connect(
+        this.url,
+        {
+          ...this.mongoOpts,
+          poolSize: dsOptions.numConns || 2
+        },
+        (err, database) => {
+          if (err) return reject(err)
+          resolve((this.db = database.db(this.namespace)))
+        }
+      )
+    })
+  }
+
   async connection () {
     try {
       while (connections.length < (dsOptions.numConns || 1)) {
@@ -69,7 +86,7 @@ export class DataSourceMongoDb extends DataSource {
           }
         }
         const breaker = CircuitBreaker(
-          'mongo.conn',
+          'mongodb.connect',
           this.connect(client),
           thresholds
         )
@@ -186,14 +203,12 @@ export class DataSourceMongoDb extends DataSource {
   /**
    *
    * @param {Object} filter Supposed to be a valid Mongo Filter
-   * @param {Object} options Options to sort limit aggregate etc...
-   * @param {Object} options.sort a valid Mongo sort object
-   * @param {Number} options.limit a valid Mongo limit
-   * @param {Object} options.aggregate a valid Mongo aggregate object
+   * @param {Object} sort a valid Mongo sort object
+   * @param {Number} limit a valid Mongo limit
+   * @param {Object} aggregate a valid Mongo aggregate object
    *
-   * @returns
+   * @returns {Promise<import('mongodb').AbstractCursor>}
    */
-
   async mongoFind ({ filter, sort, limit, aggregate, skip } = {}) {
     console.log({ fn: this.mongoFind.name, filter })
     let cursor = (await this.collection()).find(filter)
@@ -204,112 +219,23 @@ export class DataSourceMongoDb extends DataSource {
     return cursor
   }
 
-  /**
-   * Pipes to writable and streams list. List can be filtered. Stream
-   * is serialized by default. Stream can be modified by transform.
-   *
-   * @param  {{
-   *  filter:*
-   *  transform:Transform
-   *  serialize:boolean
-   * }} param0
-   * @returns
-   */
-  streamList ({ writable, serialize, transform, options }) {
-    try {
-      let first = true
-
-      const serializer = new Transform({
-        writableObjectMode: true,
-
-        // start of array
-        construct (callback) {
-          this.push('[')
-          callback()
-        },
-
-        // each chunk is a record
-        transform (chunk, _encoding, next) {
-          // comma-separate
-          if (first) first = false
-          else this.push(',')
-
-          // serialize record
-          this.push(JSON.stringify(chunk))
-          next()
-        },
-
-        // end of array
-        flush (callback) {
-          this.push(']')
-          callback()
-        }
-      })
-
-      return new Promise(async (resolve, reject) => {
-        const readable = (await this.mongoFind(options)).stream()
-
-        readable.on('error', reject)
-        readable.on('end', resolve)
-
-        // optionally transform db stream then pipe to output
-        if (transform && serialize)
-          readable
-            .pipe(transform)
-            .pipe(serializer)
-            .pipe(writable)
-        else if (transform) readable.pipe(transform).pipe(writable)
-        else if (serialize) readable.pipe(serializer).pipe(writable)
-        else readable.pipe(writable)
-      })
-    } catch (error) {}
-  }
-
   processOptions (param) {
     const { options = {}, query = {} } = param
     return { ...options, ...processQuery(query) }
   }
 
   /**
-   * Returns the set of objects satisfying the `filter` if specified;
-   * otherwise returns all objects. If a `writable`stream is provided and `cached`
-   * is false, the list is streamed. Otherwise the list is returned in
-   * an array. A custom transform can be specified to modify the streamed
-   * results. Using {@link createWriteStream} updates can be streamed back
-   * to the db. With streams, we can support queries of very large tables,
-   * with minimal memory overhead on the node server.
    *
    * @override
-   * @param {{key1:string, keyN:string}} filter - e.g. http query
-   * @param {{
-   *  writable: WritableStream,
-   *  cached: boolean,
-   *  serialize: boolean,
-   *  transform: Transform
-   * }} params
-   *    - details
-   *    - `serialize` seriailize input to writable
-   *    - `cached` list cache only
-   *    - `transform` transform stream before writing
-   *    - `writable` writable stream for output
+   * @param {import('../../domain/datasource').listOptions} param
    */
-  async list (param = {}) {
-    const {
-      writable = null,
-      transform = null,
-      serialize = false,
-      query = {}
-    } = param
-
+  async list (param) {
     try {
-      if (query.__cached) return super.listSync(query)
-      if (query.__count) return this.count()
-
       const options = this.processOptions(param)
       console.log({ options })
 
-      if (writable) {
-        return this.streamList({ writable, serialize, transform, options })
+      if (param.streamRequested) {
+        return (await this.mongoFind(options)).stream()
       }
 
       return (await this.mongoFind(options)).toArray()
@@ -318,6 +244,10 @@ export class DataSourceMongoDb extends DataSource {
     }
   }
 
+  /**
+   *
+   * @override
+   */
   async count () {
     return {
       total: await this.countDb(),

src/domain/datasource-factory.js

@@ -1,6 +1,6 @@
 'use strict'
 
-import { Transform } from 'stream'
+import { Readable, Transform, Writable } from 'node:stream'
 import { isMainThread } from 'worker_threads'
 /** @typedef {import('.').Model} Model */
 
@@ -94,39 +94,99 @@ const DsCoreExtensions = superclass =>
       }
     }
 
-    transform () {
+    hydrate () {
       const ctx = this
 
       return new Transform({
         objectMode: true,
 
-        transform (chunk, _encoding, next) {
+        transform (chunk, encoding, next) {
           this.push(ModelFactory.loadModel(broker, ctx, chunk, ctx.name))
           next()
         }
       })
     }
 
+    serialize () {
+      let first = true
+
+      return new Transform({
+        objectMode: true,
+
+        // start of array
+        construct (callback) {
+          this.push('[')
+          callback()
+        },
+
+        // each chunk is a record
+        transform (chunk, encoding, next) {
+          // comma-separate
+          if (first) first = false
+          else this.push(',')
+
+          // serialize record
+          this.push(JSON.stringify(chunk))
+          next()
+        },
+
+        // end of array
+        flush (callback) {
+          this.push(']')
+          callback()
+        }
+      })
+    }
+
     /**
+     *
+     * @param {Model[]} list
+     * @param {import('./datasource').listOptions} options
+     * @returns {Array<Readable|Transform>}
+     */
+    stream (list, options) {
+      return new Promise((resolve, reject) => {
+        options.writable.on('error', reject)
+        options.writable.on('end', resolve)
+
+        if (!isMainThread) list.push(this.hydrate())
+        if (options.transform) list.concat(options.transform)
+        if (options.serialize) list.push(this.serialize())
+
+        return list.reduce((p, c) => p.pipe(c)).pipe(options.writable)
+      })
+    }
+
+    /**
+     * Returns the set of objects satisfying the `filter` if specified;
+     * otherwise returns all objects. If a `writable` stream is provided and
+     * `cached` is false, the list is streamed. Otherwise the list is returned
+     * in an array. One or more custom transforms can be specified to modify the
+     * streamed results. Using {@link createWriteStream}, updates can be streamed
+     * back to the storage provider. With streams, we can support queries of very
+     * large tables, with minimal memory overhead on the node server.
+     *
      * @override
-     * @param {*} options
-     * @returns
+     * @param {import('../../domain/datasource').listOptions} param
      */
     async list (options) {
       try {
-        if (options?.writable)
-          return isMainThread
-            ? super.list(options)
-            : super.list({ ...options, transform: this.transform() })
+        if (options?.query.__count) return this.count()
+        if (options?.query.__cached) return this.listSync(options.query)
 
-        const arr = await super.list(options)
+        const opts = { ...options, streamRequested: options.writable }
+        const list = [await super.list(opts)].flat()
 
-        if (Array.isArray(arr))
-          return isMainThread
-            ? arr
-            : arr.map(model =>
-                ModelFactory.loadModel(broker, this, model, this.name)
-              )
+        if (list.length < 1) return []
+
+        if (list[0] instanceof Readable || list[0] instanceof Transform)
+          return this.stream(list, options)
+
+        return isMainThread
+          ? list
+          : list.map(model =>
+              ModelFactory.loadModel(broker, this, model, this.name)
+            )
       } catch (error) {
         console.error({ fn: this.list.name, error })
         throw error

src/domain/datasource.js

@@ -2,6 +2,38 @@
 
 import { changeDataCapture } from './util/change-data-capture'
 
+/**
+ * @typedef {{
+ *  'some-key': 'some-val',
+ * }} SumOfType
+ */
+
+/**
+ * @typedef {object} QueryType
+ * @property {boolean} [__cached] list cache only
+ * @property {number|SumOfType} [__count] number of object to return
+ * @property {'and'|'or'|'not'} [__operand] operation to use for key-value pairs
+ * @property {string|number|boolean} [key1] key-value pair 1
+ * @property {string|number|boolean} [keyN] key-value pair n
+ */
+
+/**
+ * Query syntax provided by the storage vendor's native API
+ * @typedef {object} VendorType
+ */
+
+/**
+ * @typedef {object} listOptions
+ * @property {QueryType} query url query params
+ * @property {VendorType} options Vendor-specific native query syntax
+ * @property {import('stream').Writable} writable writable stream for output
+ * @property {import('stream').Transform|import('stream').Transform[]} transform
+ * transform stream before writing
+ * @property {boolean} serialize seriailize input to writable
+ * @property {boolean} streamRequested true if caller provided a writable stream -
+ * indicates to the datasource that it should return a readable stream if supported
+ */
+
 /** change data capture */
 const cdcEnabled = false // /true/i.test('CHANGE_DATA_CAPTURE')
 
@@ -122,18 +154,7 @@ export default class DataSource {
 
   /**
    * list model instances
-   * @param {{key1:string, keyN:string}} filter - e.g. http query
-   * @param {{
-   *  writable: WritableStream,
-   *  cached: boolean,
-   *  serialize: boolean,
-   *  transform: Transform
-   * }} options
-   *    - details
-   *    - `serialize` seriailize input to writable
-   *    - `cached` list cache only
-   *    - `transform` transform stream before writing
-   *    - `writable` writable stream for output
+   * @param {listOptions} options
    * @returns {Promise<any[]>}
    */
   async list (options) {