refactor streaming

Author: tysonrm

src/adapters/controllers/get-models.js

@@ -2,6 +2,13 @@
 import { Readable, Transform } from 'stream'
 import getContent from './get-content'
 
+function isStream (models) {
+  return (
+    models?.length > 0 &&
+    (models[0] instanceof Readable || models[0] instanceof Transform)
+  )
+}
+
 /**
  *
  * @param {import("../use-cases/list-models").listModels} listModels
@@ -17,13 +24,7 @@ export default function getModelsFactory (listModels) {
         writable: httpRequest.res
       })
 
-      if (
-        models?.length > 0 &&
-        (models[0] instanceof Readable || models[0] instanceof Transform)
-      ) {
-        httpRequest.stream = true
-        return
-      }
+      if (isStream()) return
 
       const { content, contentType } = getContent(httpRequest, models)
 

src/adapters/controllers/http-adapter.js

@@ -16,7 +16,7 @@
 /**
  * @param {httpController} controller
  */
-export default function buildCallback(controller) {
+export default function buildCallback (controller) {
   /**
    */
   return async (req, res) => {
@@ -29,7 +29,7 @@ export default function buildCallback(controller) {
       path: req.path,
       res: res,
       headers: req.headers,
-      log(func) {
+      log (func) {
         console.info({
           function: func,
           ip: httpRequest.ip,
@@ -44,7 +44,6 @@ export default function buildCallback(controller) {
 
     return controller(httpRequest)
       .then(httpResponse => {
-        if (httpRequest.stream) return
         if (httpResponse.headers) {
           res.set(httpResponse.headers)
         }
@@ -55,4 +54,4 @@ export default function buildCallback(controller) {
         res.status(500).send({ error: 'An unkown error occurred.', e })
       )
   }
-}
+}

src/domain/datasource-factory.js

@@ -171,10 +171,10 @@ const DsCoreExtensions = superclass =>
      */
     async list (options) {
       try {
-        if (options?.query.__count) return this.count()
-        if (options?.query.__cached) return this.listSync(options.query)
+        if (options?.query?.__count) return this.count()
+        if (options?.query?.__cached) return this.listSync(options.query)
 
-        const opts = { ...options, streamRequested: options.writable }
+        const opts = { ...options, streamRequested: options?.writable }
         const list = [await super.list(opts)].flat()
 
         if (list.length < 1) return []

src/domain/use-cases/list-configs.js

@@ -25,20 +25,27 @@ export default function listConfigsFactory ({
       const poolName =
         typeof query.poolName === 'string' ? query.poolName.toUpperCase() : null
 
+      const domain = models.getModelSpec(modelName)?.domain
+
       const configTypes = {
         data: async () =>
           modelName && isMainThread
             ? threadpools
                 .getThreadPool(modelName)
                 .runJob(listConfigs.name, query, modelName)
             : modelName && poolName
-            ? await datasources.getSharedDataSource(poolName).list()
+            ? await datasources.getSharedDataSource(poolName, domain).list()
             : modelName
-            ? await datasources.getSharedDataSource(modelName).list()
+            ? await datasources.getSharedDataSource(modelName, domain).list()
             : await Promise.all(
                 datasources.listDataSources().map(async dsName => ({
                   dsName,
-                  ...(await datasources.getSharedDataSource(dsName).count())
+                  ...(await datasources
+                    .getSharedDataSource(
+                      dsName,
+                      models.getModelSpec(modelName)?.domain
+                    )
+                    .count())
                 }))
               ),