Change EventBroker to call subscriber callbacks syncrhonously.

Don't use JSON to serialize Error - comes back empty.
Listen for thread exit in case error event isn't fired and worker crashes. Remove thread from pool.

Author: tysonrm

__test__/use-cases/make-relations.js

@@ -32,7 +32,7 @@ function requireRemoteObject (model, relation, broker, ...args) {
   return new Promise(async function (resolve) {
     setTimeout(resolve, maxwait)
     broker.on(response, resolve)
-    await broker.emit(request, requestData)
+    broker.emit(request, requestData)
   })
 }
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@module-federation/aegis",
-  "version": "1.3.0-beta",
+  "version": "1.3.1-beta",
   "module": "lib/index.js",
   "main": "lib/index.js",
   "repository": "[email protected]:module-federation/aegis.git",

src/adapters/datasources/datasource-mongodb.js

@@ -64,6 +64,7 @@ export class DataSourceMongoDb extends DataSourceMemory {
       return (await this.connection()).db(this.name).collection(this.name)
     } catch (error) {
       console.error({ fn: this.collection.name, error })
+      throw error
     }
   }
 
@@ -250,7 +251,7 @@ export class DataSourceMongoDb extends DataSourceMemory {
     let cursor = (await this.collection()).find(filter)
     if (sort) cursor = cursor.sort(sort)
     if (limit) cursor = cursor.limit(parseInt(limit))
-    if (aggregate) cursor = cursor.limit(aggregate)
+    if (aggregate) cursor = cursor.aggregate(aggregate)
     return cursor
   }
 
@@ -260,7 +261,6 @@ export class DataSourceMongoDb extends DataSourceMemory {
    *
    * @param  {{
    *  filter:*
-   *  writable:Writable
    *  transform:Transform
    *  serialize:boolean
    * }} param0

src/adapters/webassembly/wasm-import.js

@@ -61,7 +61,7 @@ export async function importWebAssembly (remoteEntry) {
         const event = wasm.exports.__getString(eventName)
         const data = interop.constructObject(eventData)
         console.debug('fireEvent', data)
-        await broker.notify(event, data)
+        broker.notify(event, data)
       },
 
       /**

src/domain/distributed-cache.js

@@ -366,6 +366,7 @@ export default function DistributedCache ({
       publish({ ...event, eventName: externalCrudEvent(eventName) })
     )
 
+  function handlecrudeEvent (modelSpecs) {}
   /**
    * Subcribe to external CRUD events for related models.
    * Also listen for request and response events for locally

src/domain/event-broker.js

@@ -64,7 +64,7 @@ export class EventBroker {
    * @param {Event} eventData - the import of the event
    * @param {{forward:boolean}} options - forward this event externally
    */
-  async notify (eventName, eventData, options) {
+  notify (eventName, eventData, options) {
     throw new Error('unimplemented abstract method')
   }
 
@@ -93,7 +93,7 @@ const handleError = error => {
  * @param {eventHandler} handle
  * @param {boolean} forward
  */
-async function runHandler (eventName, eventData = {}, handle) {
+function runHandler (eventName, eventData = {}, handle) {
   const abort = eventData ? false : true
 
   if (abort) {
@@ -111,7 +111,7 @@ async function runHandler (eventName, eventData = {}, handle) {
   /**
    * @type {eventHandler}
    */
-  await handle(eventData)
+  handle(eventData)
 }
 
 /**
@@ -121,7 +121,7 @@ async function runHandler (eventName, eventData = {}, handle) {
  * @param {brokerOptions} options
  * @fires eventName
  */
-async function notify (eventName, eventData = {}, options = {}) {
+function notify (eventName, eventData = {}, options = {}) {
   const run = runHandler.bind(this)
   const data =
     typeof eventData === 'object'
@@ -134,20 +134,21 @@ async function notify (eventName, eventData = {}, options = {}) {
   try {
     if (handlers.has(eventName)) {
       match = true
-      await Promise.allSettled(
-        handlers.get(eventName).map(async fn => {
-          await run(eventName, formattedEvent, fn)
-        })
-      )
+      // await Promise.allSettled(
+      //   handlers.get(eventName).map(async fn => {
+      //     await run(eventName, formattedEvent, fn)
+      //   })
+      // )
+      handlers.get(eventName).forEach(fn => run(eventName, formattedEvent, fn))
     }
 
     if (options.regexOff && match) return
 
-    await Promise.allSettled(
-      [...handlers]
-        .filter(([k]) => k instanceof RegExp && k.test(eventName))
-        .map(([, v]) => v.map(fn => run(eventName, formattedEvent, fn)))
-    )
+    //await Promise.allSettled(
+    ;[...handlers]
+      .filter(([k]) => k instanceof RegExp && k.test(eventName))
+      .forEach(([, v]) => v.f(fn => run(eventName, formattedEvent, fn)))
+    //)
   } catch (error) {
     handleError(notify.name, error)
   }

src/domain/index.js

@@ -407,6 +407,6 @@ export { default as ThreadPoolFactory } from './thread-pool'
 
 export { default as DomainEvents } from './domain-events'
 
-export { AppError } from './util/app-error'
+export { AppError } from '../domain/util/app-error'
 
 export default ModelFactory

src/domain/make-relations.js

@@ -152,7 +152,7 @@ export function requireRemoteObject (model, relation, broker, ...args) {
   return new Promise(async function (resolve) {
     setTimeout(resolve, maxwait)
     broker.on(response, resolve)
-    await broker.notify(request, requestData)
+    broker.notify(request, requestData)
   })
 }
 

src/domain/model.js

@@ -240,7 +240,7 @@ const Model = (() => {
        * defaults to `false`
        */
       async emit (eventName, eventData, options) {
-        await broker.notify(
+        broker.notify(
           eventName,
           {
             eventName,

src/domain/thread-pool.js

@@ -126,7 +126,7 @@ export class ThreadPool extends EventEmitter {
     broker.on('to_worker', async event => port1.postMessage(event))
     // on receipt of event from worker thread fire 'from_worker'
     port1.onmessage = async event =>
-      event.data.eventName && (await broker.notify('from_worker', event.data))
+      event.data.eventName && (broker.notify('from_worker', event.data))
   }
 
   /**