v0.53.1 - Fix/Improve Operation API usage

jest.config.base.js

@@ -8,7 +8,7 @@ module.exports = (projectDir) => {
     const name = path.basename(projectDir);
     const workspaceName = name === 'e2e' ? 'e2e' : 'packages';
     const rootDir = name === 'e2e' ? '../' : '../../';
-    const projectRoot = name === 'e2e' ? '<rootDir>/e2e' : `<rootDir>/${workspaceName}/${name}`;
+    const packageRoot = name === 'e2e' ? '<rootDir>/e2e' : `<rootDir>/${workspaceName}/${name}`;
     const isTypescript = fs.pathExistsSync(path.join(projectDir, 'tsconfig.json'));
     const runInPackage = projectDir === process.cwd();
 
@@ -18,13 +18,13 @@ module.exports = (projectDir) => {
         displayName: name,
         verbose: true,
         testEnvironment: 'node',
-        setupFilesAfterEnv: ['jest-extended'],
-        testMatch: [`${projectRoot}/test/**/*-spec.{ts,js}`, `${projectRoot}/test/*-spec.{ts,js}`],
+        setupFilesAfterEnv: ['jest-extended', '<rootDir>/scripts/add-test-env.js'],
+        testMatch: [`${packageRoot}/test/**/*-spec.{ts,js}`, `${packageRoot}/test/*-spec.{ts,js}`],
         testPathIgnorePatterns: [
             '<rootDir>/assets',
             `<rootDir>/${workspaceName}/*/node_modules`,
             `<rootDir>/${workspaceName}/*/dist`,
-            `<rootDir>/${workspaceName}/teraslice-cli/test/fixtures/`,
+            `<rootDir>/${workspaceName}/teraslice-cli/test/fixtures/`
         ],
         transformIgnorePatterns: ['^.+\\.js$'],
         moduleNameMapper: lernaAliases({ mainFields: ['srcMain', 'main'] }),
@@ -33,63 +33,63 @@ module.exports = (projectDir) => {
         coveragePathIgnorePatterns: ['/node_modules/', '/test/'],
         watchPathIgnorePatterns: [],
         coverageReporters: runInPackage ? ['html'] : ['lcov', 'text', 'html'],
-        coverageDirectory: `${projectRoot}/coverage`,
+        coverageDirectory: `${packageRoot}/coverage`,
         preset: 'ts-jest',
-        watchPlugins: ['jest-watch-typeahead/filename', 'jest-watch-typeahead/testname'],
+        watchPlugins: ['jest-watch-typeahead/filename', 'jest-watch-typeahead/testname']
     };
 
     if (fs.pathExistsSync(path.join(projectDir, 'test/global.setup.js'))) {
-        config.globalSetup = `${projectRoot}/test/global.setup.js`;
+        config.globalSetup = `${packageRoot}/test/global.setup.js`;
     }
 
     if (fs.pathExistsSync(path.join(projectDir, 'test/global.teardown.js'))) {
-        config.globalTeardown = `${projectRoot}/test/global.teardown.js`;
+        config.globalTeardown = `${packageRoot}/test/global.teardown.js`;
     }
 
     if (fs.pathExistsSync(path.join(projectDir, 'test/test.setup.js'))) {
-        config.setupFilesAfterEnv.push(`${projectRoot}/test/test.setup.js`);
+        config.setupFilesAfterEnv.push(`${packageRoot}/test/test.setup.js`);
     }
 
     config.globals = {
-        availableExtensions: ['.js', '.ts'],
+        availableExtensions: ['.js', '.ts']
     };
 
     if (isTypescript) {
         if (runInPackage) {
             config.globals['ts-jest'] = {
                 tsConfig: './tsconfig.json',
                 diagnostics: true,
-                pretty: true,
+                pretty: true
             };
         } else {
             config.globals['ts-jest'] = {
                 tsConfig: `./${workspaceName}/${name}/tsconfig.json`,
                 diagnostics: true,
-                pretty: true,
+                pretty: true
             };
         }
     } else {
         config.globals['ts-jest'] = {
             diagnostics: true,
-            pretty: true,
+            pretty: true
         };
     }
 
-    config.roots = [`${projectRoot}/test`];
+    config.roots = [`${packageRoot}/test`];
 
     if (fs.pathExistsSync(path.join(projectDir, 'lib'))) {
-        config.roots.push(`${projectRoot}/lib`);
+        config.roots.push(`${packageRoot}/lib`);
     } else if (fs.pathExistsSync(path.join(projectDir, 'index.js'))) {
-        config.roots.push(`${projectRoot}`);
+        config.roots.push(`${packageRoot}`);
     }
 
     if (fs.pathExistsSync(path.join(projectDir, 'src'))) {
-        config.roots.push(`${projectRoot}/src`);
+        config.roots.push(`${packageRoot}/src`);
     }
 
     if (fs.pathExistsSync(path.join(projectDir, 'peg'))) {
-        config.watchPathIgnorePatterns.push(`${projectRoot}/peg/*engine*.js`);
-        config.roots.push(`${projectRoot}/peg`);
+        config.watchPathIgnorePatterns.push(`${packageRoot}/peg/*engine*.js`);
+        config.roots.push(`${packageRoot}/peg`);
     }
 
     return config;

packages/data-access-plugin/package.json

@@ -46,7 +46,7 @@
         "xlucene-evaluator": "^0.9.3"
     },
     "devDependencies": {
-        "@terascope/job-components": "^0.19.0",
+        "@terascope/job-components": "^0.20.0",
         "@types/express": "^4.17.0",
         "@types/got": "^9.6.0",
         "@types/graphql-iso-date": "^3.3.1",

packages/job-components/package.json

@@ -1,6 +1,6 @@
 {
     "name": "@terascope/job-components",
-    "version": "0.19.0",
+    "version": "0.20.0",
     "publishConfig": {
         "access": "public"
     },

packages/job-components/src/execution-context/api.ts

@@ -1,102 +1,138 @@
-import { OpAPI, Context, ExecutionConfig, APIConfig } from '../interfaces';
-import { OperationAPIConstructor } from '../operations';
-
-// WeakMaps are used as a memory efficient reference to private data
-const _registry = new WeakMap();
-const _apis = new WeakMap();
-const _config = new WeakMap();
+import { EventEmitter } from 'events';
+import { AnyObject, Logger, isTest } from '@terascope/utils';
+import { OpAPI, Context, ExecutionConfig, APIConfig, WorkerContext } from '../interfaces';
+import { isOperationAPI, getOperationAPIType, makeContextLogger } from './utils';
+import { Observer, APIConstructor } from '../operations';
+import { JobAPIInstances } from './interfaces';
 
 /**
  * A utility API exposed on the Terafoundation Context APIs.
  * The following functionality is included:
  *  - Registering Operation API
  *  - Creating an API (usually done from an Operation),
  *    it also includes attaching the API to the Execution LifeCycle.
- *    An API can only be created once.
+ *    An API will only be created once.
  *  - Getting a reference to an API
-*/
+ */
 export class ExecutionContextAPI {
+    private readonly _apis: JobAPIInstances = {};
+    private readonly _context: WorkerContext;
+    private readonly _events: EventEmitter;
+    private readonly _executionConfig: ExecutionConfig;
+    private readonly _logger: Logger;
+
     constructor(context: Context, executionConfig: ExecutionConfig) {
-        _config.set(this, {
-            context,
-            executionConfig,
-            events: context.apis.foundation.getSystemEvents(),
-        });
-
-        _registry.set(this, {});
-        _apis.set(this, {});
+        this._context = context as WorkerContext;
+        this._events = context.apis.foundation.getSystemEvents();
+        this._executionConfig = executionConfig;
+        this._logger = this.makeLogger('execution_context_api');
     }
 
-    /** Add an API constructor to the registry */
-    addToRegistry(name: string, api: OperationAPIConstructor) {
-        const registry = this.registry;
-        registry[name] = api;
-        _registry.set(this, registry);
+    /** For backwards compatibility */
+    get registry() {
+        return {};
     }
 
-    /** Get all of the registered API constructors */
-    get registry(): APIRegistry {
-        return _registry.get(this);
+    get apis(): Readonly<JobAPIInstances> {
+        return this._apis;
     }
 
-    /** Get all of the initalized APIs */
-    get apis(): APIS {
-        return _apis.get(this);
-    }
+    /** Add an API constructor to the registry */
+    addToRegistry(name: string, API: APIConstructor) {
+        if (this._apis[name] != null) {
+            throw new Error(`Cannot register API "${name}" due to conflict`);
+        }
+
+        const { apis = [] } = this._executionConfig;
 
-    /** Set an api to be used */
-    addAPI(name: string, opAPI: OpAPI) {
-        const apis = _apis.get(this);
-        apis[name] = opAPI;
+        const apiConfig = apis.find((a: APIConfig) => a._name === name) || {
+            _name: name,
+        };
+
+        const instance = new API(this._context, apiConfig, this._executionConfig);
+        const type = getOperationAPIType(instance);
+        this._apis[name] = {
+            instance,
+            type,
+        };
+
+        const eventName = 'execution:add-to-lifecycle';
+        const count = this._events.listenerCount(eventName);
+        if (!count) {
+            if (isTest) return;
+            this._logger.warn(`no listener ${eventName} available but is needed to register the api`);
+        } else {
+            this._events.emit(eventName, instance);
+            this._logger.trace(`registered api ${name}`);
+        }
     }
 
     /**
-     * Get a reference to a specific API,
-     * the must be initialized.
+     * Get a reference to a specific operation API,
+     * the must be initialized and created
      */
-    getAPI(name: string) {
-        if (this.apis[name] == null) {
+    getObserver<T extends Observer = Observer>(name: string): T {
+        const api = this._apis[name];
+        if (api == null) {
             throw new Error(`Unable to find API by name "${name}"`);
         }
-        return this.apis[name];
+        if (api.type !== 'observer') {
+            throw new Error(`Unable to find observer by name "${name}"`);
+        }
+        return api.instance as T;
     }
 
     /**
-     * Initalize an API and attach it
-     * to the lifecycle of an Execution.
+     * Get a reference to a specific operation API,
+     * the must be initialized and created
      */
-    async initAPI(name: string, ...params: any[]) {
-        const config = _config.get(this);
-        if (this.registry[name] == null) {
+    getAPI<T extends OpAPI = OpAPI>(name: string): T {
+        const api = this._apis[name];
+        if (api == null) {
             throw new Error(`Unable to find API by name "${name}"`);
         }
-
-        if (this.apis[name] != null) {
-            throw new Error(`API "${name}" can only be initalized once`);
+        if (api.opAPI == null) {
+            throw new Error(`Unable to find created API by name "${name}"`);
         }
+        return api.opAPI as T;
+    }
 
-        const API = this.registry[name];
-
-        const { apis = [] } = config.executionConfig;
-
-        const apiConfig = apis.find((a: APIConfig) => a._name === name) || {
-            _name: name,
-        };
+    /**
+     * Create an instance of the API
+     *
+     * @param name the name of API to create
+     * @param params any additional options that the API may need
+     */
+    async initAPI(name: string, ...params: any[]): Promise<OpAPI> {
+        const api = this._apis[name];
+        if (api == null) {
+            throw new Error(`Unable to find API by name "${name}"`);
+        }
 
-        const api = new API(config.context, apiConfig, config.executionConfig);
-        await api.initialize();
+        if (!isOperationAPI(api.instance)) {
+            throw new Error('Observers cannot be created');
+        }
 
-        config.events.emit('execution:add-to-lifecycle', api);
+        if (api.opAPI != null) {
+            const msg = `using existing instance of api: "${name}"`;
+            if (params.length) {
+                this._logger.warn(`${msg}, ignoring params`);
+            } else {
+                this._logger.debug(`${msg}`);
+            }
+            return api.opAPI;
+        }
 
-        this.addAPI(name, await api.createAPI(...params));
-        return this.apis[name];
+        api.opAPI = await api.instance.createAPI(...params);
+        this._logger.trace(`initialized api ${name}`);
+        return api.opAPI;
     }
-}
-
-interface APIS {
-    [name: string]: OpAPI;
-}
 
-interface APIRegistry {
-    [name: string]: OperationAPIConstructor;
+    /**
+     * Make a logger with a the job_id and ex_id in the logger context
+     */
+    makeLogger(moduleName: string, extra: AnyObject = {}) {
+        const { ex_id, job_id } = this._executionConfig;
+        return makeContextLogger(this._context, moduleName, { ex_id, job_id, ...extra });
+    }
 }

packages/job-components/src/execution-context/base.ts

@@ -3,14 +3,11 @@ import { isFunction, cloneDeep } from '@terascope/utils';
 import { OperationLoader } from '../operation-loader';
 import { registerApis } from '../register-apis';
 import { ExecutionConfig, WorkerContext, OperationLifeCycle } from '../interfaces';
-import {
-    EventHandlers,
-    ExecutionContextConfig,
-} from './interfaces';
+import { EventHandlers, ExecutionContextConfig } from './interfaces';
 
 /**
  * A base class for an Execution Context
-*/
+ */
 export default class BaseExecutionContext<T extends OperationLifeCycle> {
     readonly config: ExecutionConfig;
     readonly context: WorkerContext;
@@ -56,7 +53,7 @@ export default class BaseExecutionContext<T extends OperationLifeCycle> {
 
     /**
      * Called to initialize all of the registered operations
-    */
+     */
     async initialize(initConfig?: any) {
         const promises = [];
         for (const op of this.getOperations()) {
@@ -68,7 +65,7 @@ export default class BaseExecutionContext<T extends OperationLifeCycle> {
 
     /**
      * Called to cleanup all of the registered operations
-    */
+     */
     async shutdown() {
         const promises = [];
         for (const op of this.getOperations()) {
@@ -77,17 +74,20 @@ export default class BaseExecutionContext<T extends OperationLifeCycle> {
 
         await Promise.all(promises);
 
-        Object.keys(this._handlers)
-            .forEach((event) => {
-                const listener = this._handlers[event];
-                this.events.removeListener(event, listener);
-            });
+        Object.keys(this._handlers).forEach(event => {
+            const listener = this._handlers[event];
+            this.events.removeListener(event, listener);
+        });
+    }
+
+    get api() {
+        return this.context.apis.executionContext;
     }
 
     /**
      * Returns a list of any registered Operation that has been
      * initialized.
-    */
+     */
     getOperations() {
         return this._operations.values();
     }