meshlink

src/adapters/serverless/index.js

@@ -1,7 +1,7 @@
 import { makeServerlessAdapter } from './serverless-adapter'
 import * as localParsers from './parsers'
 
-const getRemoteParsers = async () => import('aegis-services/parsers')
+const getRemoteParsers = async () => null //import('aegis-services/parsers')
 
 export const getParsers = async function () {
   try {
@@ -14,4 +14,4 @@ export const getParsers = async function () {
   return localParsers
 }
 
-export const ServerlessAdapter = () => makeServerlessAdapter(getParsers)                             
+export const ServerlessAdapter = () => makeServerlessAdapter(getParsers)

src/adapters/webassembly/wasm-import.js

@@ -50,7 +50,7 @@ export async function importWebAssembly (remoteEntry, type = 'model') {
           return
         }
 
-        observer.on(event, eventData => adapter.callWasmFunction(fn, eventData))
+        observer.on(event, eventData => adapter.callWasmFunction(fn, eventData), true)
       },
 
       /**

src/adapters/webassembly/wasm-interop.js

@@ -106,7 +106,7 @@ exports.WasmInterop = function (module) {
       const obj = __getArray(ptr)
         .map(inner => __getArray(inner))
         .map(tuple => ({ [__getString(tuple[0])]: __getString(tuple[1]) }))
-        .reduce((prop1, prop2) => ({ ...prop1, ...prop2 }))
+        .reduce((obj1, obj2) => ({ ...obj1, ...obj2 }))
 
       const immutableClone = Object.freeze({ ...obj })
       !unpin || __unpin(ptr)

src/domain/distributed-cache.js

@@ -319,8 +319,10 @@ export default function DistributedCache ({
    * @param {*} externalEvent
    */
   const forwardSearchRequest = (internalEvent, externalEvent) =>
-    observer.on(internalEvent, async event =>
-      publish({ ...event, eventName: externalEvent })
+    observer.on(
+      internalEvent,
+      async event => publish({ ...event, eventName: externalEvent }),
+      true
     )
 
   /**
@@ -336,7 +338,7 @@ export default function DistributedCache ({
    */
   const broadcastCrudEvent = eventName =>
     observer.on(eventName, async event =>
-      publish({ ...event, eventName: externalCrudEvent(eventName) })
+      publish({ ...event, eventName: externalCrudEvent(eventName) }, true)
     )
 
   /**

src/domain/make-relations.js

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

src/domain/observer.js

@@ -80,9 +80,13 @@ async function notify (eventName, eventData, forward = false) {
   try {
     if (this.handlers.has(eventName)) {
       await Promise.allSettled(
-        this.handlers
-          .get(eventName)
-          .map(handler => run(eventName, eventData, handler, forward))
+        this.handlers.get(eventName).map(handler => {
+          console.debug('hander running', {
+            eventName,
+            handler: handlers.toString()
+          })
+          run(eventName, eventData, handler, forward)
+        })
       )
     }
 
@@ -129,6 +133,18 @@ class ObserverImpl extends Observer {
     }
     return true
   }
+
+  serialize () {
+    return JSON.stringify(
+      [...this.handlers].map(([k, v]) => ({ [k]: v.map(fn => fn.toString()) })),
+      null,
+      2
+    )
+  }
+
+  toString () {
+    console.log('toString', this.serialize())
+  }
 }
 
 /**

src/services/dns/index.js

@@ -1,7 +1,7 @@
 'use strict'
 
 import * as localClients from './providers'
-const getRemoteClients = async () => import('aegis-services/dns')
+const getRemoteClients = async () => null //import('aegis-services/dns')
 
 const dns = async function () {
   const name = process.env.DNS_SERVICE

src/services/service-mesh/mesh-link.js

@@ -5,7 +5,7 @@ const nanoid = require('nanoid').nanoid
 const begins = Date.now()
 const uptime = () => Math.round(Math.abs((Date.now() - begins) / 1000 / 60))
 const userConfig = require('../../../../microlib/public/aegis.config.json')
-const DEBUG =
+const debug =
   /true/i.test(userConfig.services.serviceMesh.MeshLink.debug) || false
 
 const defaultCfg = {
@@ -23,11 +23,7 @@ const defaultCfg = {
 
 const cfg = userConfig.services.serviceMesh.MeshLink.config || defaultCfg
 
-let started = false
-const subscriptions = new Map()
-
 function numericHash (str) {
-  if (subscriptions.has(str)) return null
   let hash = 0
   let i
   let chr
@@ -37,8 +33,7 @@ function numericHash (str) {
     hash = (hash << 5) - hash + chr
     hash |= 0 // Convert to 32bit integer
   }
-  const handleId = Math.abs(parseInt(hash % 10000)) //keep under 0xffff
-  subscriptions.set(str, handleId)
+  return Math.abs(parseInt(hash % 10000)) //keep under 0xffff
 }
 
 const sharedObjects = new Map()
@@ -101,25 +96,22 @@ const SharedObjEvent = {
 }
 
 const registerSharedObjEvents = observer =>
-  observer.on(/^externalCrudEvent_.*/, async (eventName, eventData) =>
-    SharedObjEvent[eventName.split('_')[1].substr(0, 6)](eventData)
+  observer.on(
+    /^externalCrudEvent_.*/,
+    async (eventName, eventData) =>
+      SharedObjEvent[eventName.split('_')[1].substr(0, 6)](eventData),
+    true
   )
 
 /**
  *
  * @param {cfg} config
  * @returns
  */
-async function start (config = defaultCfg, observer = null) {
-  if (started) return
-  started = true
-  observer && registerSharedObjEvents(observer)
-
+async function start (config = cfg) {
   mlink
     .start(config)
     .then(() => {
-      // connect to uplink if configured
-      !config.uplink || uplink(config)
       console.info('meshlink started')
     })
     .catch(error => {
@@ -128,23 +120,18 @@ async function start (config = defaultCfg, observer = null) {
 }
 
 async function publish (event, observer) {
+  console.debug('publish called', event.eventName, event, observer)
   const deserEvent = JSON.parse(JSON.stringify(event))
   const handlerId = numericHash(event.eventName)
-  DEBUG && console.debug('mlink publish', handlerId, deserEvent)
-
-  start(cfg, observer).then(() => {
-    return mlink.send(
-      handlerId,
-      mlink.getNodeEndPoints(),
-      deserEvent,
-      response => {
-        console.debug('response to publish ', handlerId, response)
-        const eventData = JSON.parse(response)
-        if (eventData?.eventName) {
-          observer.notify(eventData.eventName, eventData)
-        }
-      }
-    )
+  console.debug('mlink stringify', handlerId, JSON.stringify(event))
+  console.debug('mlink parse', handlerId, JSON.parse(JSON.stringify(event)))
+
+  mlink.send(handlerId, mlink.getNodeEndPoints(), deserEvent, response => {
+    console.debug('response to publish ', handlerId, response)
+    const eventData = JSON.parse(response)
+    if (eventData?.eventName) {
+      observer.notify(eventData.eventName, eventData)
+    }
   })
   try {
     global.broadcast(JSON.stringify(event), {
@@ -158,18 +145,18 @@ async function publish (event, observer) {
 async function subscribe (eventName, callback) {
   const handlerId = numericHash(eventName)
   if (!handlerId) return // we've already registered a callback for this event
-  DEBUG && console.debug('mlink subscribe', eventName, handlerId)
-  start(cfg).then(() =>
-    mlink.handler(handlerId, (data, cb) => {
-      console.log('mlink.handler called with data', handlerId, data)
-      cb(callback(data))
-    })
-  )
+  debug && console.debug('mlink subscribe', eventName, handlerId)
+
+  mlink.handler(handlerId, (data, cb) => {
+    console.log('mlink.handler called with data', handlerId, data)
+    callback(data)
+    cb(data)
+  })
 }
 
 function attachServer (server) {
   let messagesSent = 0
-
+  start()
   /**
    *
    * @param {object} data
@@ -178,7 +165,7 @@ function attachServer (server) {
   server.broadcast = function (data, sender) {
     server.clients.forEach(function (client) {
       if (client.OPEN && client.info.id !== sender.info.id) {
-        !DEBUG || console.debug('sending client', client.info, data.toString())
+        debug && console.debug('sending client', client.info, data.toString())
         client.send(data)
         messagesSent++
       }
@@ -196,9 +183,11 @@ function attachServer (server) {
   server.sendStatus = function (client) {
     client.send(
       JSON.stringify({
+        servicePlugin: 'MeshLink',
         uptimeMinutes: uptime(),
         messagesSent,
-        clientsConnected: server.clients.size
+        clientsConnected: server.clients.size,
+        meshLinkNodes: mlink.getNodeEndPoints()
       })
     )
   }
@@ -207,7 +196,7 @@ function attachServer (server) {
     client.info = { address: client._socket.address(), id: nanoid() }
 
     client.addListener('ping', function () {
-      !DEBUG || console.debug('responding to client ping', client.info)
+      debug && console.debug('responding to client ping', client.info)
       client.pong(0xa)
     })
 

src/services/service-mesh/webswitch/web-node.js

@@ -15,6 +15,8 @@ const config = configFile.services.serviceMesh.WebSwitch
 const DEBUG = /true|yes|y/i.test(config.debug) || false
 const heartbeat = config.heartbeat || 10000
 
+if (!configFile) console.error('WebSwitch', 'cannot access config file')
+
 /**
  * @type import("ws/lib/websocket")
  */
@@ -125,7 +127,7 @@ function setupHeartBeat (ws) {
 }
 
 export async function subscribe (eventName, callback, observer) {
-  observer.on(eventName, callback)
+  observer.on(eventName, callback, true)
 }
 
 /**

src/use-cases/edit-model.js

@@ -28,10 +28,10 @@ export default function makeEditModel ({
 } = {}) {
   const eventType = models.EventTypes.UPDATE
   const eventName = models.getEventName(eventType, modelName)
-  handlers.forEach(handler => observer.on(eventName, handler))
+  handlers.forEach(handler => observer.on(eventName, handler, false))
 
   // Add an event that can be used to edit this model
-  observer.on(domainEvents.editModel(modelName), editModelHandler)
+  observer.on(domainEvents.editModel(modelName), editModelHandler, false)
 
   async function editModel (id, changes, command) {
     const model = await repository.find(id)

src/use-cases/forward-events.js

@@ -72,8 +72,10 @@ export function forwardEvents ({ observer, models, publish, subscribe }) {
     )
 
     producerEvents.forEach(producerEvent =>
-      observer.on(producerEvent, eventData =>
-        publish(producerEvent, eventData || 'no data')
+      observer.on(
+        producerEvent,
+        eventData => publish(producerEvent, eventData || 'no data'),
+        true
       )
     )
   }
@@ -82,8 +84,10 @@ export function forwardEvents ({ observer, models, publish, subscribe }) {
    * Forward events so marked.
    */
   function forwardUserEvents () {
-    observer.on(domainEvents.forwardEvent, eventData =>
-      publish(eventData.eventName, eventData)
+    observer.on(
+      domainEvents.forwardEvent(),
+      eventData => publish(eventData.eventName, eventData),
+      false
     )
   }
 

src/use-cases/remove-model.js

@@ -28,7 +28,7 @@ export default function removeModelFactory ({
 } = {}) {
   const eventType = models.EventTypes.DELETE
   const eventName = models.getEventName(eventType, modelName)
-  handlers.forEach(handler => observer.on(eventName, handler))
+  handlers.forEach(handler => observer.on(eventName, handler, false))
 
   return async function removeModel (id) {
     const model = await repository.find(id)

target/npmlist.json

@@ -1,5 +1,5 @@
 {
-  "version": "0.0.0-alpha.95",
+  "version": "1.0.0-alpha.95",
   "name": "@module-federation/aegis",
   "dependencies": {
     "@octokit/core": {

wasm/index.js

@@ -64,16 +64,13 @@ async function importWebAssembly () {
             wasm.exports.__getString(callbackName)
           )
 
-
-
-            if (typeof fn === 'function') {
-              observer.on(eventName, eventData => {
-
+          if (typeof fn === 'function') {
+            observer.on(eventName, eventData => {
               adapter.callWasmFunction(fn, eventData, false)
               return
-            }
+            })
             console.log('no command found')
-          })
+          }
         },
 
         /**