move streaming logic into DataSourceFactory

Author: tysonrm

src/adapters/controllers/get-models.js

@@ -1,4 +1,5 @@
 'use strict'
+import { Readable, Transform } from 'stream'
 import getContent from './get-content'
 
 /**
@@ -16,11 +17,14 @@ export default function getModelsFactory (listModels) {
         writable: httpRequest.res
       })
 
-      if (!models) {
+      if (
+        models?.length > 0 &&
+        (models[0] instanceof Readable || models[0] instanceof Transform)
+      ) {
         httpRequest.stream = true
         return
       }
-      
+
       const { content, contentType } = getContent(httpRequest, models)
 
       return {

src/adapters/datasources/datasource-mongodb.js

@@ -69,7 +69,7 @@ export class DataSourceMongoDb extends DataSource {
           }
         }
         const breaker = CircuitBreaker(
-          'mongo.conn',
+          'mongodb.connect',
           this.connect(client),
           thresholds
         )
@@ -215,55 +215,6 @@ export class DataSourceMongoDb extends DataSource {
    * }} 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
@@ -294,22 +245,14 @@ export class DataSourceMongoDb extends DataSource {
    *    - `writable` writable stream for output
    */
   async list (param = {}) {
-    const {
-      writable = null,
-      transform = null,
-      serialize = false,
-      query = {}
-    } = param
+    const { writable = null, transform = null, serialize = false } = 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 })
+        return (await this.mongoFind(options)).stream()
       }
 
       return (await this.mongoFind(options)).toArray()

src/domain/datasource-factory.js

@@ -1,6 +1,6 @@
 'use strict'
 
-import { Transform } from 'stream'
+import { Readable, Transform } from 'node:stream'
 import { isMainThread } from 'worker_threads'
 /** @typedef {import('.').Model} Model */
 
@@ -94,39 +94,85 @@ 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()
+        }
+      })
+    }
+
+    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((a, b) => a.pipe(b)).pipe(options.writable)
+      })
+    }
+
     /**
      * @override
      * @param {*} options
      * @returns
      */
     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 list = [await super.list(options)].flat(1)
 
-        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