@nxtedition/agent - npm Package Compare versions

@nxtedition/agent

Package Overview

Dependencies

Maintainers

Advanced tools

Install Socket

Detect and block malicious and high-risk dependencies

Install

Comparing version 2.0.0 to 3.1.0

147

index.js

		const objectHash = require('object-hash')
		const { Observable, BehaviorSubject } = require('rxjs')

		function hash (obj, len = 10) {
		const hash = `${objectHash(obj).slice(0, len)}`
		function hash (obj) {
		const hash = `${objectHash(obj).slice(0, 10)}`
		if (hash[0] === '_') {
		@@ -29,146 +28,6 @@ hash[0] = '0'

		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,
		AgentFactory
		spawn
		}

package.json

		{
		"name": "@nxtedition/agent",
		"version": "2.0.0",
		"version": "3.1.0",
		"main": "index.js",
		@@ -5,0 +5,0 @@ "author": "Robert Nagy <robert.nagy@boffins.se>",

Worsened metrics