support for wasm streaming compilation and adapter refactoring

Author: tysonrm

src/adapters/datasources/datasource-mongodb.js

@@ -315,29 +315,8 @@ export class DataSourceMongoDb extends DataSource {
   async delete (id) {
     try {
       await (await this.collection()).deleteOne({ _id: id })
-      this.deleteSync(id)
     } catch (error) {
       console.error(error)
     }
   }
-
-  /**
-   * Flush the cache to disk.
-   * @override
-   */
-  flush () {
-    try {
-      this.dsMap.reduce((a, b) => a.then(() => this.saveDb(b.getId(), b)), {})
-    } catch (error) {
-      console.error(error)
-    }
-  }
-
-  /**
-   * Process terminating, flush cache, close connections.
-   * @override
-   */
-  close () {
-    this.flush()
-  }
 }

src/adapters/webassembly/wasm-import.js

@@ -48,6 +48,7 @@ exports.importWebAssembly = function (remoteEntry) {
         liftArray: __liftArray,
         notnull: __notnull,
         store_ref: __store_ref,
+        _exports: exports,
         memory,
 
         getModelName () {
@@ -65,10 +66,6 @@ exports.importWebAssembly = function (remoteEntry) {
           return __liftString(exports.getDomain() >>> 0)
         },
 
-        callExportedFn (fn, kv) {
-          return exports[fn](kv)
-        },
-
         ArrayOfStrings_ID: {
           // assembly/index/ArrayOfStrings_ID: u32
           valueOf () {
@@ -79,9 +76,14 @@ exports.importWebAssembly = function (remoteEntry) {
           }
         },
 
-        modelFactory (kv) {
-          // assembly/index/modelFactory(~lib/array/Array<~lib/array/Array<~lib/string/String>>) => ~lib/array/Array<~lib/array/Array<~lib/string/String>>
-          kv =
+        modelFactory (obj) {
+          const entries = Object.entries(obj)
+            .filter(([k, v]) =>
+              ['string', 'number', 'boolean'].includes(typeof v)
+            )
+            .map(([k, v]) => [k, v.toString()])
+
+          const kv =
             __lowerArray(
               (pointer, value) => {
                 __store_ref(
@@ -98,7 +100,7 @@ exports.importWebAssembly = function (remoteEntry) {
               },
               5,
               2,
-              kv
+              entries
             ) || __notnull()
 
           return __liftArray(
@@ -134,6 +136,23 @@ exports.importWebAssembly = function (remoteEntry) {
             .reduce((a, b) => ({ ...a, ...b }))
         },
 
+        getCommands () {
+          // assembly/index/getCommands() => ~lib/array/Array<~lib/array/Array<~lib/string/String>>
+          return __liftArray(
+            pointer =>
+              __liftArray(
+                pointer =>
+                  __liftString(new Uint32Array(memory.buffer)[pointer >>> 2]),
+                2,
+                new Uint32Array(memory.buffer)[pointer >>> 2]
+              ),
+            2,
+            exports.getCommands() >>> 0
+          )
+            .map(([k, v]) => ({ [k]: v }))
+            .reduce((a, b) => ({ ...a, ...b }))
+        },
+
         emitEvent (kv) {
           // assembly/index/emitEvent(~lib/array/Array<~lib/array/Array<~lib/string/String>>) => ~lib/array/Array<~lib/array/Array<~lib/string/String>>
           kv =
@@ -172,23 +191,6 @@ exports.importWebAssembly = function (remoteEntry) {
           )
         },
 
-        getCommands () {
-          // assembly/index/getCommands() => ~lib/array/Array<~lib/array/Array<~lib/string/String>>
-          return __liftArray(
-            pointer =>
-              __liftArray(
-                pointer =>
-                  __liftString(new Uint32Array(memory.buffer)[pointer >>> 2]),
-                2,
-                new Uint32Array(memory.buffer)[pointer >>> 2]
-              ),
-            2,
-            exports.getCommands() >>> 0
-          )
-            .map(([k, v]) => ({ [k]: v }))
-            .reduce((a, b) => ({ ...a, ...b }))
-        },
-
         onUpdate (kv) {
           // assembly/index/onUpdate(~lib/array/Array<~lib/array/Array<~lib/string/String>>) => ~lib/array/Array<~lib/array/Array<~lib/string/String>>
           kv =
@@ -353,8 +355,8 @@ exports.importWebAssembly = function (remoteEntry) {
       }
     }
 
-    function __notnull () {
-      throw TypeError('value must not be null')
+    function __notnull (key) {
+      throw TypeError(`value must not be null ${key}`)
     }
 
     function __store_ref (pointer, value) {

src/adapters/webassembly/wasm-interop.js

@@ -26,7 +26,7 @@ exports.WasmInterop = function (wasmExports) {
     store_ref,
     notnull,
     memory,
-    callExportedFn
+    _exports
   } = wasmExports
 
   /**
@@ -44,7 +44,7 @@ exports.WasmInterop = function (wasmExports) {
           new Uint32Array(memory.buffer)[pointer >>> 2]
         ),
       2,
-      callExportedFn(fn, kv) >>> 0
+      _exports[fn](kv) >>> 0
     )
       .map(([k, v]) => ({ [k]: v }))
       .reduce((a, b) => ({ ...a, ...b }))
@@ -78,6 +78,12 @@ exports.WasmInterop = function (wasmExports) {
     )
   }
 
+  function cleanse (obj) {
+    return Object.entries(obj)
+      .filter(([k, v]) => ['string', 'number', 'boolean'].includes(typeof v))
+      .map(([k, v]) => [k, v.toString()])
+  }
+
   /**
    * Parse the input object into a multidimensional array of key-value pairs
    * and pass it as an argument to the exported wasm function. Do the reverse for
@@ -92,8 +98,8 @@ exports.WasmInterop = function (wasmExports) {
    * @returns {object|number} object
    */
   function callWasmFunction (fn, obj) {
-    const entries = Object.entries(obj)
-    const kv = lower(entries)
+    const props = cleanse(obj)
+    const kv = lower(props)
     return lift(fn, kv)
   }
 
@@ -151,14 +157,22 @@ exports.WasmInterop = function (wasmExports) {
               producesEvent,
               callback: data => callWasmFunction(callback, data),
               undo: data => callWasmFunction(undo, data),
-              inbound: (port, args, id) =>
+              inbound: function inbound (port, args, id) {
                 callWasmFunction(inbound, { port, ...args, id })
+              }
             }
           }
         })
         .reduce((p, c) => ({ ...p, ...c }))
     },
 
+    importWasmPortFunctions () {
+      const ports = getPorts()
+      return Object.values(ports)
+        .filter(v => v.inbound)
+        .reduce((a, b) => [...a, ...b], [])
+    },
+
     constructObject (ptr) {
       return constructObject(ptr, false)
     }

src/adapters/webassembly/wasm-wrappers.js

@@ -46,8 +46,8 @@ exports.wrapWasmModelSpec = function (wasmExports) {
     ports: {
       ...interop.importWasmPorts()
     },
-    getInboundPortFunctions () {
-      return Object.values(this.ports).flatMap(p => p.inboundFn)
+    portFunctions: {
+      ...interop.importWasmPortFunctions()
     }
   }
   console.debug(wrappedSpec)

src/domain/circuit-breaker.js

@@ -319,7 +319,7 @@ const CircuitBreaker = function (id, protectedCall, thresholds) {
     async invoke (...args) {
       const breaker = Switch(id, thresholds)
       errorEvents.forEach(monitorErrors.bind(this))
-      breaker.incrementInvocationCounter()
+      breaker.appendLog()
 
       // check breaker status
       if (breaker.closed()) {

src/domain/datasource-factory.js

@@ -9,6 +9,7 @@ import { isMainThread } from 'worker_threads'
  * @property {boolean} memoryOnly - if true returns memory adapter and caches it
  * @property {boolean} ephemeral - if true returns memory adapter but doesn't cache it
  * @property {string} adapterName - name of adapter to use
+ * @property {Array<function():typeof import('./datasource').default>} mixins
  */
 
 import ModelFactory from '.'
@@ -132,12 +133,6 @@ const DsCoreExtensions = superclass =>
       }
     }
 
-    listSync (options) {
-      const count = options?.query?.__count
-      if (count) return this.handleCount(count)
-      return super.listSync(options)
-    }
-
     /**
      * @override
      * @param {*} id
@@ -146,12 +141,11 @@ const DsCoreExtensions = superclass =>
     async delete (id) {
       try {
         await super.delete(id)
+        // only if super succeeds
+        this.deleteSync(id)
       } catch (error) {
         console.error(error)
         throw error
-      } finally {
-        // only if super succeeds
-        this.deleteSync(id)
       }
     }
   }
@@ -260,7 +254,8 @@ const DataSourceFactory = (() => {
 
     if (!options.ephemeral) dataSources.set(name, newDs)
 
-    debug && console.debug({ newDs })
+    //debug &&
+    console.debug({ newDs })
     return newDs
   }
 

src/domain/datasource.js

@@ -36,10 +36,14 @@ function roughSizeOfObject (...objects) {
   return bytes
 }
 
-/**
- * Data source base class
- */
 export default class DataSource {
+  /**
+   * Data source base class
+   * @param {Map} map
+   * @param {string} name
+   * @param {string} namespace
+   * @param {*} options
+   */
   constructor (map, name, namespace, options = {}) {
     this.dsMap = map
     this.name = name
@@ -133,8 +137,7 @@ export default class DataSource {
    * @returns {Promise<any[]>}
    */
   async list (options) {
-    const count = options.filter.__count || options.query.__count
-    if (count) return this.handleCount(count)
+    throw new Error('unimplemented abstract method')
   }
 
   /**
@@ -143,6 +146,7 @@ export default class DataSource {
    * @returns
    */
   listSync (query) {
+    if (query?.__count) return this.handleCount(query.__count)
     const list = this.generateList()
     return query ? this.filterList(query, list) : list
   }
@@ -196,6 +200,7 @@ export default class DataSource {
       }
 
       const operand = query.__operand ? query.__operand.toLowerCase() : 'and'
+
       if (typeof operands[operand] !== 'function')
         throw new Error('invalid query')
 

src/domain/index.js

@@ -260,9 +260,9 @@ function register ({
 
   const bindings = bindAdapters({
     portSpec: model.ports,
+    ports: { ...model.portFunctions, ...ports },
     adapters,
-    services,
-    ports
+    services
   })
 
   const dependencies = {