fix mongo streaming

Author: tysonrm

src/adapters/datasources/datasource-mongodb.js

@@ -14,8 +14,8 @@ const processQuery = qpm({
   converters: {
     nullstring: val => {
       return { $type: 10 }
-    }, // reference BSON datatypes https://www.mongodb.com/docs/manual/reference/bson-types/
-  },
+    } // reference BSON datatypes https://www.mongodb.com/docs/manual/reference/bson-types/
+  }
 })
 
 const url = process.env.MONGODB_URL || 'mongodb://localhost:27017'
@@ -136,16 +136,16 @@ export class DataSourceMongoDb extends DataSource {
     }
   }
 
-  async collection() {
+  async collection () {
     return (await this.connection()).db(this.namespace).collection(this.name)
   }
 
-  async createIndexes(client) {
+  async createIndexes (client) {
     const indexOperations = this.options.connOpts.indexes.map(index => {
       return {
         name: index.fields.join('_'),
         key: index.fields.reduce((a, v) => ({ ...a, [v]: 1 }), {}),
-        ...index.options,
+        ...index.options
       }
     })
 
@@ -278,7 +278,7 @@ export class DataSourceMongoDb extends DataSource {
    * @returns
    */
 
-  async mongoCount({ filter = {} } = {}) {
+  async mongoCount ({ filter = {} } = {}) {
     return filter == {}
       ? await this.countDb()
       : (await this.collection()).count(filter)
@@ -295,21 +295,20 @@ export class DataSourceMongoDb extends DataSource {
    * }} param0
    * @returns
    */
-  streamList({ writable, serialize, transform, options }) {
+  streamList ({ writable, serialize, transform, options }) {
     try {
-
       const serializer = new Transform({
         writableObjectMode: true,
 
         // start of array
-        construct(callback) {
+        construct (callback) {
           this.first = true
           this.push('[')
           callback()
         },
 
         // each chunk is a record
-        transform(chunk, _encoding, next) {
+        transform (chunk, _encoding, next) {
           // comma-separate
           if (this.first) this.first = false
           else this.push(',')
@@ -320,32 +319,32 @@ export class DataSourceMongoDb extends DataSource {
         },
 
         // end of array
-        flush(callback) {
+        flush (callback) {
           this.push(']')
           callback()
-        },
+        }
       })
 
       const paginate = new Transform({
         writableObjectMode: true,
 
         // start of array
-        construct(callback) {
+        construct (callback) {
           this.push('{ ')
           callback()
         },
 
         // first chunk is the pagination data, rest is result
-        transform(chunk, _encoding, next) {
+        transform (chunk, _encoding, next) {
           this.push(chunk)
           next()
         },
 
         // end of array
-        flush(callback) {
+        flush (callback) {
           this.push('}')
           callback()
-        },
+        }
       })
 
       return new Promise(async (resolve, reject) => {
@@ -354,7 +353,7 @@ export class DataSourceMongoDb extends DataSource {
 
         // optionally transform db stream then pipe to output
         pipeArgs.push(readable)
-        
+
         if (options.page) {
           const count = ~~(await this.mongoCount(options.filter))
           pipeArgs.push(paginate)
@@ -366,19 +365,18 @@ export class DataSourceMongoDb extends DataSource {
             () => console.log('paginated query', options)
           )
         }
-        
+
         return pipeArgs
-   
       })
     } catch (error) {
       console.error({
         fn: this.streamList.name,
-        error,
+        error
       })
     }
   }
 
-  processOptions({ options = {}, query = {} }) {
+  processOptions ({ options = {}, query = {} }) {
     return { ...processQuery(query), ...options } // options must overwite the query not otherwise
   }
 
@@ -389,13 +387,14 @@ export class DataSourceMongoDb extends DataSource {
    */
   async list (param) {
     try {
+      let result
       const options = this.processOptions(param)
-      if (0 < ~~query.__page) {
+      if (0 < ~~options.__page) {
         // qpm > processOptions weeds out __page - add it back properly as an integer
         options.page = ~~query.__page
         options.skip = (options.page - 1) * options.limit || 0
       }
-      if (query.__aggregate) {
+      if (options.__aggregate) {
         // qpm > processOptions weeds out __aggregate - add it back properly as parsed json
         try {
           const aggregateQuery = JSON.parse(query.__aggregate)
@@ -406,7 +405,7 @@ export class DataSourceMongoDb extends DataSource {
       }
       console.log({ options })
 
-      if (streamRequested && !query.__json)
+      if (options.streamRequested && !query.__json)
         this.streamList({ writable, serialize, transform, options })
       else {
         const data = (await this.mongoFind(options)).toArray()
@@ -415,7 +414,7 @@ export class DataSourceMongoDb extends DataSource {
           ...options,
           data,
           count,
-          total: Math.ceil(count / options.limit),
+          total: Math.ceil(count / options.limit)
         }
       }
 

src/domain/datasource-factory.js

@@ -38,15 +38,15 @@ const DefaultDataSource =
  */
 const DsCoreExtensions = superclass =>
   class extends superclass {
-    constructor(map, name, namespace, options) {
+    constructor (map, name, namespace, options) {
       super(map, name, namespace, options)
     }
 
-    set factory(value) {
+    set factory (value) {
       this[FACTORY] = value
     }
 
-    get factory() {
+    get factory () {
       return this[FACTORY]
     }
 
@@ -61,7 +61,7 @@ const DsCoreExtensions = superclass =>
      * @param {string} id
      * @param {Model} data
      */
-    async save(id, data) {
+    async save (id, data) {
       try {
         this.saveSync(id, data)
         await super.save(id, JSON.parse(JSON.stringify(data)))
@@ -78,7 +78,7 @@ const DsCoreExtensions = superclass =>
      * @param {string} id
      * @returns {Promise<Model>|undefined}
      */
-    async find(id) {
+    async find (id) {
       try {
         const cached = this.findSync(id)
         if (cached) return cached
@@ -98,33 +98,33 @@ const DsCoreExtensions = superclass =>
       }
     }
 
-    hydrate() {
+    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() {
+    serialize () {
       let first = true
 
       return new Transform({
         objectMode: true,
 
         // start of array
-        construct(callback) {
+        construct (callback) {
           this.push('[')
           callback()
         },
 
         // each chunk is a record
-        transform(chunk, encoding, next) {
+        transform (chunk, encoding, next) {
           // comma-separate
           if (first) first = false
           else this.push(',')
@@ -135,10 +135,10 @@ const DsCoreExtensions = superclass =>
         },
 
         // end of array
-        flush(callback) {
+        flush (callback) {
           this.push(']')
           callback()
-        },
+        }
       })
     }
 
@@ -148,7 +148,7 @@ const DsCoreExtensions = superclass =>
      * @param {import('./datasource').listOptions} options
      * @returns {Array<Readable|Transform>}
      */
-    stream(list, options) {
+    stream (list, options) {
       return new Promise((resolve, reject) => {
         options.writable.on('error', reject)
         options.writable.on('end', resolve)
@@ -173,14 +173,15 @@ const DsCoreExtensions = superclass =>
      * @override
      * @param {import('../../domain/datasource').listOptions} param
      */
-    async list(options) {
+    async list (options) {
       try {
         if (options?.query?.__count) return this.count()
         if (options?.query?.__cached) return this.listSync(options.query)
 
         const opts = { ...options, streamRequested: options?.writable }
         const list = [await super.list(opts)].flat()
 
+        console.debug({ list })
         if (list.length < 1) return []
 
         if (list[0] instanceof Readable || list[0] instanceof Transform)
@@ -202,7 +203,7 @@ const DsCoreExtensions = superclass =>
      * @param {*} id
      * @returns
      */
-    async delete(id) {
+    async delete (id) {
       try {
         await super.delete(id)
         // only if super succeeds
@@ -238,11 +239,11 @@ const DataSourceFactory = (() => {
    * @param {*} name
    * @returns
    */
-  function hasDataSource(name) {
+  function hasDataSource (name) {
     return dataSources.has(name)
   }
 
-  function listDataSources() {
+  function listDataSources () {
     return [...dataSources.keys()]
   }
 
@@ -267,7 +268,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} options
    * @returns {typeof DataSource}
    */
-  function createDataSourceClass(spec, options) {
+  function createDataSourceClass (spec, options) {
     const { memoryOnly, ephemeral, adapterName } = options
 
     if (memoryOnly || ephemeral) return dsClasses['DataSourceMemory']
@@ -295,7 +296,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} options
    * @returns {typeof DataSource}
    */
-  function extendDataSourceClass(DsClass, options = {}) {
+  function extendDataSourceClass (DsClass, options = {}) {
     const mixins = [extendClass].concat(options.mixins || [])
     return compose(...mixins)(DsClass)
   }
@@ -306,7 +307,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} [options]
    * @returns {DataSource}
    */
-  function createDataSource(name, namespace, options) {
+  function createDataSource (name, namespace, options) {
     const spec = ModelFactory.getModelSpec(name)
     const dsMap = options.dsMap || new Map()
 
@@ -332,7 +333,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} options
    * @returns {import('./datasource').default}
    */
-  function getDataSource(name, namespace = null, options = {}) {
+  function getDataSource (name, namespace = null, options = {}) {
     if (!dataSources) dataSources = new Map()
     if (!namespace) return dataSources.get(name)
     if (dataSources.has(name)) return dataSources.get(name)
@@ -347,7 +348,7 @@ const DataSourceFactory = (() => {
    * @param {dsOpts} [options]
    * @returns
    */
-  function getSharedDataSource(name, namespace = null, options = {}) {
+  function getSharedDataSource (name, namespace = null, options = {}) {
     if (!dataSources) dataSources = new Map()
     if (!namespace) return dataSources.get(name)
     if (dataSources.has(name)) return dataSources.get(name)
@@ -362,20 +363,20 @@ const DataSourceFactory = (() => {
    * @param {string} name
    * @returns {ProxyHandler<DataSource>}
    */
-  function getRestrictedDataSource(name, namespace, options) {
+  function getRestrictedDataSource (name, namespace, options) {
     return new Proxy(getDataSource(name, namespace, options), {
-      get(target, key) {
+      get (target, key) {
         if (key === 'factory') {
           throw new Error('unauthorized')
         }
       },
-      ownKeys(target) {
+      ownKeys (target) {
         return []
-      },
+      }
     })
   }
 
-  function close() {
+  function close () {
     dataSources.forEach(ds => ds.close())
   }
 
@@ -390,7 +391,7 @@ const DataSourceFactory = (() => {
     getRestrictedDataSource,
     hasDataSource,
     listDataSources,
-    close,
+    close
   })
 })()