error handling improvements

Author: tysonrm

__test__/use-cases/make-relations.js

@@ -36,7 +36,8 @@ function requireRemoteObject (model, relation, broker, ...args) {
   })
 }
 
-const broker = new EventEmitter()
+const broker = new EventEmitter({ captureRejections: true })
+broker.on('error', err => console.log(err))
 
 const model = {
   getName: () => 'model1',

src/adapters/datasources/datasource-mongodb.js

@@ -60,19 +60,14 @@ export class DataSourceMongoDb extends DataSourceMemory {
   }
 
   async collection () {
-    try {
-      return (await this.connection()).db(this.name).collection(this.name)
-    } catch (error) {
-      console.error({ fn: this.collection.name, error })
-      throw error
-    }
+    return (await this.connection()).db(this.name).collection(this.name)
   }
 
   /**
    * @override
    * @param {{
    *  hydrate:function(Map<string,import("../../domain").Model>),
-   *  serializer:import("../../lib/serializer").Serializer
+   *  seriali zer:import("../../lib/serializer").Serializer
    * }} options
    */
   load ({ hydrate, serializer }) {

src/domain/bind-adapters.js

@@ -1,5 +1,7 @@
 'use strict'
 
+const debug = /true/i.test(process.env.DEBUG)
+
 function DefaultInboundAdapter (port) {
   return async function ({ model, args }) {
     return port.apply(model, args)
@@ -8,28 +10,35 @@ function DefaultInboundAdapter (port) {
 
 /**
  * In a hex arch, ports and adapters control I/O between
- * the application core (domain) and the outside world.
- * Either a port invokes an outbound adapter or it is invoked
- * by an inbound one. This function calls adapter factory
- * functions to inject each adapter with its port or service
- * dependency. 
- * 
- * It returns an object containing the set of port functions 
- * defined for the domain model in the model spec. These functions 
- * are invoked by a universal function that handles error recovery, 
+ * the application core and the outside world. Inbound
+ * adapters invoke ports and ports invoke outbound adapters.
+ * Optionally, outbound adapters invoke services.
+ *
+ * To set the above each adapter's factory function
+ * to inject its port or service dependency--I.e. to bind
+ * it to a port or service.
+ *
+ * It returns an object containing the set of port functions
+ * defined in the model spec for the domain model. These functions
+ * are invoked by a  port function, which handles error recovery,
  * instrumentation, authorization, flow control and other port features.
  *
- * @param {import('.').ports} ports - domain interfaces
- * @param {{[x:string]:function(*):function(*):any}} adapters - service adapters
- * @param {*} [services] - (micro-)services
+ * @param {{
+ *  portSpec:import('.').ports
+ *  ports:{[x:string]:function()}
+ *  adapters:{[x:string]:function()}
+ *  services:{[x:string]:function()}
+ * }}
+ *
  */
 export default function bindAdapters ({
   ports,
   adapters,
   services,
   portSpec
 } = {}) {
-  if (!portSpec || !adapters) {
+  if (!portSpec || !adapters || !ports) {
+    debug && console.debug('missing params')
     return
   }
 
@@ -44,26 +53,18 @@ export default function bindAdapters ({
 
   return Object.keys(portSpec)
     .map(portName => {
-      try {
-        const spec = portSpec[portName]
-        const type = spec.type
-        const port = ports[portName]
-        const adapter = adapters[portName]
-        const service = services[spec.service]
-
-        if (!adapter && type === 'outbound') {
-          console.warn('no adapter for port', portName, spec)
-          return
-        }
+      const spec = portSpec[portName]
+      const type = spec?.type ? spec.type : null
+      const port = ports[portName]
+      const adapter = adapters[portName]
+      const service = services && spec.service ? services[spec.service] : null
 
-        return bindings[type](portName, port, adapter, service)
-      } catch (error) {
-        console.warn({
-          fn: bindAdapters.name,
-          error,
-          spec: portSpec[portName]
-        })
+      if (!spec || !type || !port || !adapter) {
+        debug && console.debug('bad port configuration', portName, spec)
+        return
       }
+
+      return bindings[type](portName, port, adapter, service)
     })
     .reduce((p, c) => ({ ...p, ...c }))
 }

src/domain/make-ports.js

@@ -61,7 +61,7 @@ function setPortTimeout (options) {
   // Retry the port on timeout
   const timerId = setTimeout(async () => {
     // Notify interested parties
-    await model.emit(portTimeout(model.getName(), portName), timerArgs)
+    model.emit(portTimeout(model.getName(), portName), timerArgs)
     // Invoke optional custom handler
     if (handler) handler(options)
     // Count retries by passing `timerArgs` to ourselves on the stack
@@ -230,7 +230,7 @@ export default function makePorts (ports, adapters, broker) {
 
           // Signal the next port to run.
           if (rememberPort) {
-            await model.emit(portConf.producesEvent, portName)
+            model.emit(portConf.producesEvent, portName)
           }
 
           // the result can be something other than a model

src/domain/mixins.js

@@ -129,7 +129,7 @@ export const withDeserializers = (...funcs) => o => {
 export const withbroker = broker => o => {
   return {
     ...o,
-    async emit (eventName, eventData) {
+    emit (eventName, eventData) {
       broker.notify(eventName, eventData)
     },
     subscribe (eventName, callback) {

src/domain/model.js

@@ -14,7 +14,7 @@
  * @property {function()} toJSON - de/serialization logic
  * @property {function(eventName,function(eventName,Model):void)} addListener listen
  * for domain events
- * @property {function(eventName,Model):Promise<void>} emit emit a domain event
+ * @property {function(eventName,Model):void} emit emit a domain event
  * @property {function()} [mixinMethod] - when the user
  * specifies a mixin, it is applied to the model on creation - adding methods is
  * a common result.
@@ -239,7 +239,7 @@ const Model = (() => {
        * @param {boolean} [forward] - forward event to service mesh,
        * defaults to `false`
        */
-      async emit (eventName, eventData, options) {
+      emit (eventName, eventData, options) {
         broker.notify(
           eventName,
           {

src/domain/orchestrator.js

@@ -35,7 +35,7 @@ export async function runWorkflow ({ wfName }) {
     DataSourceFactory.getSharedDataSource(wfName),
     wfName
   )
-  await model.emit(wfName)
+  model.emit(wfName)
   console.info(wfName, 'workflow started')
 }
 
@@ -45,23 +45,21 @@ export async function runWorkflow ({ wfName }) {
  *
  * @param {Array<import(".").Model>} list
  */
-export async function resumeWorkflow (list) {
+export function resumeWorkflow (list) {
   if (list?.length > 0) {
-    await Promise.all(
-      list.map(async function (model) {
-        const history = model.getPortFlow()
-        const ports = model.getSpec().ports
+    list.forEach(function (model) {
+      const history = model.getPortFlow()
+      const ports = model.getSpec().ports
 
-        if (history?.length > 0 && !model.compensate) {
-          const lastPort = history.length - 1
-          const nextPort = ports[history[lastPort]].producesEvent
+      if (history?.length > 0 && !model.compensate) {
+        const lastPort = history.length - 1
+        const nextPort = ports[history[lastPort]].producesEvent
 
-          if (nextPort && history[lastPort] !== 'workflowComplete') {
-            await async(model.emit(nextPort, resumeWorkflow.name))
-          }
+        if (nextPort && history[lastPort] !== 'workflowComplete') {
+          model.emit(nextPort, resumeWorkflow.name)
         }
-      })
-    ).catch(error => console.error(error))
+      }
+    })
   }
 }