shared datasources

.gitignore

@@ -7,3 +7,4 @@ node_modules
 .idea
 .DS_Store
 tst-rtry.js
+.VSCodeCounter/

src/adapters/datasources/datasource-memory.js

@@ -6,10 +6,6 @@ import DataSource from '../../domain/datasource'
  * Temporary in-memory storage.
  */
 export class DataSourceMemory extends DataSource {
-  constructor (map, factory, name) {
-    super(map, factory, name)
-  }
-
   /**
    * @override
    *
@@ -107,6 +103,10 @@ export class DataSourceMemory extends DataSource {
         id
       })
     }
+    this.deleteSync(id)
+  }
+
+  async deleteSync (id) {
     this.dsMap.delete(id)
   }
 }

src/adapters/datasources/datasource-mongodb.js

@@ -13,7 +13,7 @@ const cacheSize = configRoot.adapters.cacheSize || 3000
  * even when the database is offline.
  */
 export class DataSourceMongoDb extends DataSourceMemory {
-  constructor (maps, factory, name) {
+  constructor (map, factory, name) {
     super(map, factory, name)
     this.url = url
     this.cacheSize = cacheSize

src/adapters/datasources/index.js

@@ -1,26 +1,15 @@
 export * from './datasource-memory'
 export * from './datasource-file'
 export * from './datasource-mongodb'
-export * from './datasource-ipfs'
-//export * from "./datasource-solid-pod";
+// export * from './datasource-ipfs'
+// export * from './datasource-solid-pod'
 
-const config = {
-  getBaseClass (name) {
-    if (name === 'DataSourceFile') {
-      return require('.').DataSourceFile
-    }
-    if (name === 'DataSourceMongoDb') {
-      return require('.').DataSourceMongoDb
-    }
-    if (name === 'DataSourceIpfs') {
-      return require('.').DataSourceIpfs
-    }
-    // if (name === "DataSourceSolidPod") {
-    //   return require(".").DataSourceSolidPod;
-    // }
-    return require('.').DataSourceMemory
-  },
-  MEMORYADAPTER: 'DataSourceMemory'
-}
+import { DataSourceFile } from './datasource-file'
+import { DataSourceMemory } from './datasource-memory'
+import { DataSourceMongoDb } from './datasource-mongodb'
 
-export default config
+export const dsClasses = {
+  [DataSourceFile.name]: DataSourceFile,
+  [DataSourceMemory.name]: DataSourceMemory,
+  [DataSourceMongoDb.name]: DataSourceMongoDb
+}

src/domain/datasource-factory.js

@@ -11,7 +11,7 @@
 
 import ModelFactory from '.'
 import * as adapters from '../adapters/datasources'
-import dsconfig from '../adapters/datasources'
+import { dsClasses } from '../adapters/datasources'
 import sysconf from '../config'
 import DataSource from './datasource'
 import { withSharedMemory } from './shared-memory'
@@ -67,7 +67,7 @@ const DataSourceFactory = (() => {
     const { memoryOnly, ephemeral, adapterName } = options
 
     if (memoryOnly || ephemeral) {
-      return dsconfig.getBaseClass(dsconfig.MEMORYADAPTER)
+      return dsClasses['DataSourceMemory']
     }
 
     if (adapterName) return adapters[adapterName] || DefaultDataSource
@@ -76,7 +76,7 @@ const DataSourceFactory = (() => {
       const url = spec.datasource.url
       const cacheSize = spec.datasource.cacheSize
       const adapterFactory = spec.datasource.factory
-      const BaseClass = dsconfig.getBaseClass(spec.datasource.baseClass)
+      const BaseClass = dsClasses[spec.datasource.baseClass]
       return adapterFactory(url, cacheSize, BaseClass)
     }
 

src/domain/distributed-cache.js

@@ -159,7 +159,7 @@ export default function DistributedCache ({
       eventName === models.getEventName(models.EventTypes.DELETE, modelName)
     ) {
       console.debug('deleting from cache', modelName, event.modelId)
-      await datasources.getSharedDataSource(modelName).delete(event.modelId)
+      await datasources.getDataSource(modelName).delete(event.modelId)
       return true
     }
     return false
@@ -283,7 +283,7 @@ export default function DistributedCache ({
         // find the requested object or objects
         const relatedModels = await relationType[relation.type](
           model,
-          datasources.getSharedDataSource(relation.modelName.toUpperCase()),
+          datasources.getDataSource(relation.modelName.toUpperCase()),
           relation
         )
 

src/domain/federated-query.js

@@ -0,0 +1,34 @@
+'use strict'
+
+import { EventBrokerFactory } from '.'
+import { requireRemoteObject } from './make-relations'
+
+export const FederationMixin = superclass =>
+  class extends superclass {
+    /**
+     * Deserialize
+     * @override
+     * @param {*} id
+     * @returns {import('.').Model}
+     */
+    async find (id) {
+      try {
+        const result = await super.find(id)
+        if (!result || result.length < 1) return this.findRemote(id)
+        return result
+      } catch (error) {
+        console.error({ fn: 'federatedFindById', error })
+      }
+    }
+
+    async findRemote (id) {
+      const event = await requireRemoteObject(
+        null,
+        { type: 'findById', modelName: this.name, id },
+        EventBrokerFactory.getInstance()
+      )
+      if (event?.model) return event.model
+      console.debug('federated findById: no model found')
+      return {}
+    }
+  }

src/domain/thread-pool.js

@@ -7,6 +7,7 @@ import domainEvents from './domain-events'
 import ModelFactory, { DataSourceFactory } from '.'
 import { performance as perf } from 'perf_hooks'
 import os from 'os'
+import { Perf } from 'nats/lib/nats-base-client/util'
 
 const { poolOpen, poolClose, poolDrain, poolAbort } = domainEvents
 const broker = EventBrokerFactory.getInstance()
@@ -53,6 +54,8 @@ async function kill (thread, reason) {
   try {
     return await new Promise(resolve => {
       console.info({ msg: 'killing thread', id: thread.id, reason })
+      const TIMEOUT = 8000
+      const start = perf.now()
 
       const timerId = setTimeout(async () => {
         await thread.mainChannel.terminate()
@@ -62,12 +65,14 @@ async function kill (thread, reason) {
           reason
         })
         resolve(thread.id)
-      }, 8000)
+      }, TIMEOUT)
 
       thread.mainChannel.once('exit', () => {
         clearTimeout(timerId)
         thread.eventChannel.close()
-        console.info('clean exit of thread', thread.id, reason)
+        // the timeout will cause this to run if executed first
+        if (perf.now() - start < TIMEOUT)
+          console.info('clean exit of thread', thread.id, reason)
         resolve(thread.id)
       })
 
@@ -128,7 +133,7 @@ function newThread ({ pool, file, workerData }) {
       const timerId = setTimeout(async () => {
         const message = 'timedout creating thread'
         console.error({ fn: newThread.name, message })
-        reject('timeout')
+        reject(message)
       }, 50000)
 
       worker.once('message', async msg => {
@@ -300,6 +305,7 @@ export class ThreadPool extends EventEmitter {
       this.freeThreads.findIndex(t => t.id === thread.id),
       1
     )
+    return this
   }
 
   /**
@@ -847,8 +853,8 @@ const ThreadPoolFactory = (() => {
   return Object.freeze({
     getThreadPool,
     broadcastEvent,
-    listPools,
     fireEvent,
+    listPools,
     reloadAll,
     reload,
     status,

src/domain/use-cases/broker-events.js

@@ -86,7 +86,7 @@ export default function brokerEvents (
   process.on('message', ({ cmd, id, pid, data, name }) => {
     if (cmd && id && data && process.pid !== pid) {
       if (cmd === 'saveCommand') {
-        const ds = datasources.getSharedDataSource(name)
+        const ds = datasources.getDataSource(name)
         ds.save(id, data, false)
         return
       }