-
Notifications
You must be signed in to change notification settings - Fork 13
/
base.ts
197 lines (160 loc) · 6.41 KB
/
base.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
import { EventEmitter } from 'events';
import {
isFunction, cloneDeep, pMap, Logger, toHumanTime
} 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 { ExecutionContextAPI } from './api';
/**
* A base class for an Execution Context
*/
export default class BaseExecutionContext<T extends OperationLifeCycle> {
readonly config: ExecutionConfig;
readonly context: WorkerContext;
readonly assetIds: string[] = [];
readonly exId: string;
readonly jobId: string;
/** The terafoundation EventEmitter */
readonly events: EventEmitter;
readonly logger: Logger;
protected readonly _loader: OperationLoader;
protected readonly _operations = new Set() as Set<T>;
protected _methodRegistry = new Map<keyof T, Set<number>>();
private readonly _handlers: EventHandlers = {};
constructor(config: ExecutionContextConfig, loggerName: string) {
this.events = config.context.apis.foundation.getSystemEvents();
this._handlers['execution:add-to-lifecycle'] = (op: T) => {
this.addOperation(op);
};
this.events.on('execution:add-to-lifecycle', this._handlers['execution:add-to-lifecycle']);
const executionConfig = cloneDeep(config.executionConfig);
this._loader = new OperationLoader({
terasliceOpPath: config.terasliceOpPath,
assetPath: config.context.sysconfig.teraslice.assets_directory,
});
registerApis(config.context, executionConfig, config.assetIds);
this.context = config.context as WorkerContext;
this.assetIds = config.assetIds || [];
this.config = executionConfig;
this.exId = executionConfig.ex_id;
this.jobId = executionConfig.job_id;
this.logger = this.api.makeLogger(loggerName, { level: executionConfig.log_level });
}
/**
* Called to initialize all of the registered operations
*/
async initialize(initConfig?: unknown): Promise<void> {
// make sure we autoload the apis before we initialize the processors
await pMap((this.config.apis || []), async ({ _name: name }) => {
const api = this.api.apis[name];
if (api.type !== 'api') return;
const startTime = Date.now();
this.logger.info(`[START] "${name}" api instance initialize`);
try {
await this.api.initAPI(name);
} finally {
const diff = toHumanTime(Date.now() - startTime);
this.logger.info(`[FINISH] "${name}" api instance initialize, took ${diff}`);
}
});
await pMap(this._operations, async (op) => {
if (!('initialize' in op)) return;
const startTime = Date.now();
// @ts-expect-error
const name = op.opConfig?._op ?? op.apiConfig?._name ?? op.constructor.name;
this.logger.info(`[START] "${name}" operation initialize`);
try {
await op.initialize(initConfig);
} finally {
const diff = toHumanTime(Date.now() - startTime);
this.logger.info(`[FINISH] "${name}" operation initialize, took ${diff}`);
}
});
}
/**
* Called to cleanup all of the registered operations
*/
async shutdown(): Promise<void> {
await pMap(this._operations, async (op) => {
if (!('shutdown' in op)) return;
const startTime = Date.now();
// @ts-expect-error
const name = op.opConfig?._op ?? op.apiConfig?._name ?? op.constructor.name;
this.logger.info(`[START] "${name}" operation shutdown`);
try {
await op.shutdown();
} finally {
const diff = toHumanTime(Date.now() - startTime);
this.logger.info(`[FINISH] "${name}" operation shutdown, took ${diff}`);
}
}).finally(() => {
Object.entries(this._handlers)
.forEach(([event, listener]) => {
this.events.removeListener(event, listener);
});
});
await this.context.apis.foundation.promMetrics.shutdown();
}
get api(): ExecutionContextAPI {
return this.context.apis.executionContext;
}
/**
* Returns a list of any registered Operation that has been
* initialized.
*/
getOperations(): T[] {
return [...this._operations.values()];
}
/** Add an operation to the lifecycle queue */
protected addOperation(op: T): void {
this._operations.add(op);
this._resetMethodRegistry();
}
/** Run an async method on the operation lifecycle */
protected _runMethodAsync(method: keyof T, ...args: any[]): Promise<any[]> {
const set = this._getMethodSet(method);
if (set.size === 0) return Promise.resolve([]);
let i = 0;
const promises = [];
for (const operation of this._operations) {
const index = i++;
if (set.has(index)) {
// @ts-expect-error because I can't get the type definitions to work right
promises.push(operation[method](...args));
}
}
return Promise.all(promises);
}
/** Run an method */
protected _runMethod(method: keyof T, ...args: any[]): void {
const set = this._getMethodSet(method);
if (set.size === 0) return;
let index = 0;
for (const operation of this._operations) {
if (set.has(index)) {
// @ts-expect-error because I can't get the type definitions to work right
operation[method](...args);
}
index++;
}
}
protected _resetMethodRegistry(): void {
for (const set of this._methodRegistry.values()) {
set.clear();
}
let index = 0;
for (const op of this._operations) {
for (const [method, set] of this._methodRegistry.entries()) {
if (isFunction(op[method])) {
set.add(index);
}
}
index++;
}
}
private _getMethodSet(method: keyof T): Set<number> {
return this._methodRegistry.get(method) || new Set();
}
}