@nxtedition/agent - npm Package Compare versions

Comparing version 1.3.0 to 2.0.0

index.js

 const objectHash = require('object-hash')
+const { Observable, BehaviorSubject } = require('rxjs')
 
-function hash (obj) {
-  const hash = `${objectHash(obj).slice(0, 10)}`
+function hash (obj, len = 10) {
+  const hash = `${objectHash(obj).slice(0, len)}`
   if (hash[0] === '_') {
@@ -17,3 +18,3 @@ hash[0] = '0'
   }
-  ds.record.update(`${id}:general.tags`, doc => ({ value: [ 'hidden' ], ...doc }))
+  ds.record.update(`${id}:general.tags`, doc => ({ value: [ 'nxt-hidden' ], ...doc }))
   ds.record.update(`${id}:general.created`, doc => ({ value: new Date().toISOString(), ...doc }))
@@ -29,6 +30,146 @@ ds.record.set(`${id}:agent.query`, query)
 
+class AgentFactory {
+  constructor ({ ds }) {
+    this._ds = ds
+    this._workers$ = new BehaviorSubject({})
+
+    this._ds.record.provide(`(.+):agent`, key => {
+      const id = key.slice(0, -`agent`.length)
+      return this._workers$
+        .pluck(id)
+        .distinctUntilChanged()
+        .map(worker =>  {
+          if (!worker) {
+            return null
+          }
+
+          const refs = []
+
+          const { 
+            methods$ = Observable.of({}), 
+            stats$ = Observable.of({}), 
+            result$ = Observable.of({})
+          } = worker({
+            id,
+            ref () {
+              refs.push(this._ds.record.getRecord(key))
+            },
+            unref () {
+              refs.pop().discard()
+            }
+          })
+
+          return Observable
+            .combineLatest(
+              this._methods(id, methods$),
+              stats$,
+              result$,
+              (methods, stats, result) => ({ methods, stats, result })
+            )
+        })
+    }, true)
+  }
+
+  create ({ normalize, spawn }) {
+    return this._ds.record.provide(`^(.+):agent(?:\\.(methods|stats|result))?$`, key => {
+      const [ , id, prop ] = key.match(/^(.+):agent(?:\.(methods|stats|result))?$/)
+      return this._ds.record
+        .observe(`${id}:agent.query`)
+        .first(data => Object.keys(data).length > 0)
+        .map(query => {
+          const description = normalize(query)
+
+          if (!description) {
+            return null
+          }
+          
+          const child = `${hash(description, 14)}:agent`
+          return Observable
+            .using(() => {
+              const workers = { ...this._workers$.getValue() }
+              workers[child] = agent => spawn({ ...agent, description })
+              this._workers$.next(workers)
+              return {
+                unsubscribe: () => {
+                  const workers = { ...this._workers$.getValue() }
+                  delete workers[child]
+                  this._workers$.next(workers)
+                }
+              }
+            },
+            () => prop
+              ? this._ds.record
+                  .observe(child)
+                  .pluck(prop)
+                  .map(x => x || {})
+              : Observable
+                  .of({ ...query, child })
+          )
+        })
+    }, true)
+  }
+
+  _methods (id, methods$) {
+    return Observable.create(o => {
+      let curr = {}
+
+      const subscription = methods$
+        .subscribe({
+          next: methods => {
+            const prev = { ...curr }
+            curr = {}
+            for (const [ key, val ] of Object.entries(methods)) {
+              const func = typeof val === 'function' ? val : val.func
+
+              if (prev[key] && prev[key].func === func) {
+                const rpcId = prev[key].rpcId
+                delete prev[key]
+                curr[key] = { rpcId, func }
+              } else if (func) {
+                const rpcId = `${id}.${key}`
+                this._ds.rpc.provide(rpcId, func)
+                curr[key] = { rpcId, func }
+              }
+    
+              if (curr[key]) {
+                curr[key].rpcData = val.data
+                curr[key].title = val.title || val.name
+                curr[key].description = val.description
+              }
+            }
+
+            for (const { rpcId } of Object.values(prev)) {
+              this._ds.rpc.unprovide(rpcId)
+            }
+
+            o.next(Object
+              .entries(curr)
+              .reduce(([ key, val ]) => ({ [key]: { ...val, func: undefined } }), {})
+            )
+          },
+          error: err => {
+            o.next({})
+            o.error(err)
+          },
+          complete: () => {
+            o.next({})
+            o.complete()
+          }
+        })
+
+      return () => {
+        for (const { rpcId } of Object.values(curr)) {
+          this._ds.rpc.unprovide(rpcId)
+        }
+        subscription.unsubscribe()
+      }
+    })
+  }
+}
+
 module.exports = {
   hash,
   get,
-  spawn
+  spawn,
+  AgentFactory
 }

package.json

 {
   "name": "@nxtedition/agent",
-  "version": "1.3.0",
+  "version": "2.0.0",
   "main": "index.js",
@@ -5,0 +5,0 @@ "author": "Robert Nagy <[e-mail]>",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

6068

increased by196.58%

Lines of code

154

increased by450%

No dependency changes