import { MessageLogger , MessageStreamer } from "@adpt/utils" ; import db from "debug" ; import * as ld from "lodash" ; import * as util from "util" ; import { createObserverManagerDeployment , ExecutedQuery , Observations , ObserverManagerDeployment , } from "./obs_manager_deployment" ; import { ObserverPlugin } from "./plugin" ; const debug = db ( "adapt:observers" ) ; interface ObserverRecord { [ name ] : ObserverPlugin < any > ; } export interface ObserverNameHolder { observerName : string ; } const observers = { } ; export function makeObserverManagerDeployment ( observations ) { const mgr = createObserverManagerDeployment ( ) ; for ( const name in observers ) { if ( ! Object . hasOwnProperty . call ( observers , name ) ) continue ; const o = observers [ name ] ; const obs = observations [ name ] ; mgr . registerSchema ( { observerName : name } , o . schema , obs ? obs . observations : { } ) ; } return mgr ; } export function registerObserver < D = object , C = any > ( obs , nameIn ? ) { const constructor = obs . constructor ; const name = nameIn ? nameIn : constructor . name ; if ( name in observers ) throw new Error ( `Attempt to register observer with duplicate name '${ name }'` ) ; observers [ name ] = obs ; if ( constructor . observerName === undefined ) constructor . observerName = name ; return name ; } export function findObserver ( observer ) { return observers [ observer . observerName ] ; } export async function observe ( executedQueries , logger = new MessageStreamer ( "observe" ) , observerNames = Object . keys ( observers ) ) { const ret = { } ; const waitFors = [ ] ; const errors = [ ] ; for ( const name of observerNames ) { const obs = observers [ name ] ; if ( obs === undefined ) continue ; const queries = executedQueries [ name ] ? executedQueries [ name ] : [ ] ; const waitP = ( async ( ) => { try { debug ( `starting observations for ${ name } with ${ queries . length } queries` ) ; const observations = await obs . observe ( queries ) ; debug ( `finished observations for ${ name }` ) ; ret [ name ] = { observations , queries } ; } catch ( e ) { debug ( `errored observations for ${ name }` ) ; if ( ! ld . isError ( e ) ) e = new Error ( util . inspect ( e ) ) ; const msg = `Error observing for ${ name }: ${ e . message }` ; logger . warning ( msg ) ; errors . push ( new Error ( msg ) ) ; } } ) ( ) ; waitFors . push ( waitP ) ; } await Promise . all ( waitFors ) ; if ( errors . length !== 0 ) { const msgs = errors . map ( ( val ) => val . message ) ; const e = new Error ( "Errors during observations:\n" + msgs . join ( "\n" ) ) ; e . observations = ret ; throw e ; } return ret ; }