Source: ProcessWrapper.js

  1. const cp = require('child_process')
  2. , EventEmitter = require('events').EventEmitter
  3. , Rx = require('rxjs')
  4. , Observable = Rx.Observable
  5. , fromEvent = Rx.fromEvent
  6. , { mergeObjects } = require('./tools/Objects')
  7. , symbolProcessOutput = Symbol('processOutput');
  11. /**
  12.  * General purpose class that can represent any result of a terminated Process.
  13.  */
  14. class ProcessResult {
  15. 	constructor() { };
  16. };
  19. /**
  20.  * Inherits from ProcessResult and represents an errored Process. This class
  21.  * is usually used for Processes that did not even start or were impossible
  22.  * to launch with the given configuration. A Process that started but errored
  23.  * then is usually terminated with an instance of ProcessExit.
  24.  */
  25. class ProcessErrorResult extends ProcessResult {
  26. 	/**
  27. 	 * @param {any} error 
  28. 	 */
  29. 	constructor(error) {
  30. 		super();
  31. 		this.error = error;
  32. 	};
  33. };
  36. /**
  37.  * Inherits from ProcessResult and represents the details of an exited Process.
  38.  * This result can either be positive (non-faulted) or negative. The result will
  39.  * provide details about the termination, such as exit-code, signal (if available)
  40.  * or the contents of stderr and stdout (if configured to be used).
  41.  */
  42. class ProcessExit extends ProcessResult {
  43. 	/**
  44. 	 * @param {boolean} faulted
  45. 	 * @param {number} code 
  46. 	 * @param {string|null} signal 
  47. 	 * @param {string|null} stdOut 
  48. 	 * @param {string|null} stdErr 
  49. 	 */
  50. 	constructor(faulted, code, signal, stdOut, stdErr) {
  51. 		super();
  52. 		this.faulted = faulted;
  53. 		this.code = code;
  54. 		this.signal = signal;
  55. 		this.stdOut = stdOut;
  56. 		this.stdErr = stdErr;
  57. 	};
  58. }
  61. /**
  62.  * This class is used for whenever a wrapped Process generates any kind
  63.  * of output.
  64.  */
  65. class ProcessOutput {
  66. 	/**
  67. 	 * @param {'stdout'|'stderr'} streamType 
  68. 	 * @param {Buffer|String} chunk 
  69. 	 */
  70. 	constructor(streamType, chunk) {
  71. 		this.streamType = streamType;
  72. 		this.chunk = chunk;
  73. 	};
  75. 	get isStdOut() {
  76. 		return this.streamType === 'stdout';
  77. 	};
  79. 	get isStdErr() {
  80. 		return this.streamType === 'stderr';
  81. 	};
  83. 	/**
  84. 	 * Provides the chunk as string, converted to utf-8.
  85. 	 */
  86. 	get asString() {
  87. 		return this.chunk instanceof Buffer ?
  88. 			this.chunk.toString('utf-8') : this.chunk;
  89. 	};
  90. }
  93. /**
  94.  * Wraps (the execution of) a Process that is created using child_process.
  95.  * Provides convenient async behavior and Observables as well as eventing.
  96.  */
  97. class ProcessWrapper extends EventEmitter {
  98. 	/**
  99. 	 * @param {string} command 
  100. 	 * @param {Array.<string>} args 
  101. 	 * @param {SpawnOptions|Object} options The same options that child_process.spawn(..) supports.
  102. 	 */
  103. 	constructor(command, args = [], options = {}) {
  104. 		super();
  106. 		this.command = command;
  107. 		this.args = args;
  108. 		this.options = options;
  110. 		/** @protected */
  111. 		this._wasStarted = false;
  112. 		/** @protected */
  113. 		this._isRunning = false;
  115. 		/**
  116. 		 * @type {Observable.<ProcessOutput>}
  117. 		 * @protected
  118. 		 */
  119. 		this._observable = fromEvent(this, symbolProcessOutput);
  121. 		/**
  122. 		 * @type {ProcessResult}
  123. 		 * @protected
  124. 		 */
  125. 		this._processResult = void 0;
  127. 		/**
  128. 		 * @type {NodeJS.Process}
  129. 		 * @protected
  130. 		 */
  131. 		this._process = void 0;
  132. 	};
  134. 	/**
  135. 	 * Returns the underlying process as returned by child_process.
  136. 	 * 
  137. 	 * @throws {Error} if this process was not yet started.
  138. 	 * @returns {NodeJS.Process}
  139. 	 */
  140. 	get process() {
  141. 		if (!this.wasStarted) {
  142. 			throw new Error('This process was not yet started.');
  143. 		}
  144. 		return this._process;
  145. 	};
  147. 	/**
  148. 	 * @returns {Boolean}
  149. 	 */
  150. 	get wasStarted() {
  151. 		return this._wasStarted;
  152. 	};
  154. 	/**
  155. 	 * @returns {Boolean}
  156. 	 */
  157. 	get isRunning() {
  158. 		return this._isRunning;
  159. 	};
  161. 	/**
  162. 	 * Creates and returns a general Observable for any output of this Process. The
  163. 	 * Observable will just observe any emits. Thus it never errors and never drains.
  164. 	 * 
  165. 	 * @returns {Observable.<ProcessOutput>}
  166. 	 */
  167. 	get observable() {
  168. 		return this._observable;
  169. 	};
  171. 	/**
  172. 	 * This property will yield a value once the Process terminated. Note that in the
  173. 	 * API of Observable, the complete()-function does not support arguments, so that
  174. 	 * we cannot pass in the result there.
  175. 	 * Instead the wrapper guarantees that, once the Process terminated, its result
  176. 	 * will be made available before any events are omitted or complete() gets called.
  177. 	 * 
  178. 	 * @returns {ProcessResult|ProcessErrorResult|ProcessExit}
  179. 	 */
  180. 	get result() {
  181. 		return this._processResult;
  182. 	};
  184. 	/**
  185. 	 * The default action to run this process. At the moment it defaults to spawn().
  186. 	 * The default behavior may change in the future, so do not rely on this method
  187. 	 * and use the specialized instantiation methods, if you require e.g. fork()
  188. 	 * instead of spawn().
  189. 	 * However regardless, this method guarantees that the process will be executed
  190. 	 * and that its result can be observed or obtained by awaiting its termination
  191. 	 * (that depends on the behavior of 'emitOnly').
  192. 	 * 
  193. 	 * @param {boolean} emitOnly If set to true, the resulting value of the resolving
  194. 	 * Promise will be empty (i.e. stdout and stderr will be empty and not be collected
  195. 	 * while the process is running; instead, all output is emitted (events and obser-
  196. 	 * vable) while the process is running). It is recommended to set this to true, if
  197. 	 * the underlying process outputs a lot of data to the std-streams. That data would
  198. 	 * otherwise be held in the node.js process. However, if set to false, the resulting
  199. 	 * value of the Promise will be the process' entire output.
  200. 	 */
  201. 	async run(emitOnly = true) {
  202. 		if (this.isRunning) {
  203. 			throw new Error('The process is already running.');
  204. 		}
  205. 		return await this.spawnAsync(emitOnly);
  206. 	};
  208. 	/**
  209. 	 * The default action to start and observe this process. At the moment it defaults
  210. 	 * to spawn() and thus spawnObservable(). The default behavior may change in the future,
  211. 	 * so do not rely on this method and use the specialized instantiation methods, if you
  212. 	 * require e.g. fork() instead of spawn().
  213. 	 * However regardless, this method guarantees that the process will be executed and that
  214. 	 * its result can be observed. When the process errors or terminates, the observable will
  215. 	 * forward those errors or call onComplete() accordingly.
  216. 	 * 
  217. 	 * @returns {Observable.<ProcessOutput>}
  218. 	 */
  219. 	runObservable() {
  220. 		return this.spawnObservable();
  221. 	};
  223. 	/**
  224. 	 * Calls 'spawnAsync(emitOnly=true)' so that the entire Process becomes
  225. 	 * observable while it runs. When the process errors or finishes, the
  226. 	 * observable will output an error or complete, respectively.
  227. 	 * 
  228. 	 * @returns {Observable.<ProcessOutput>}
  229. 	 */
  230. 	spawnObservable() {
  231. 		return new Observable(async subs => {
  232. 			const subscription = fromEvent(this, symbolProcessOutput)
  233. 				.subscribe(next => {
  234. 					subs.next(next);
  235. 				});
  237. 			try {
  238. 				await this.spawnAsync(true);
  239. 				subs.complete();
  240. 			} catch (e) {
  241. 				subs.error(e);
  242. 			} finally {
  243. 				subscription.unsubscribe();
  244. 			}
  245. 		});
  246. 	};
  248. 	/**
  249. 	 * Launches the process using spawn().
  250. 	 * 
  251. 	 * @param {boolean} emitOnly Has the same effect as 'run(emitOnly)', please refer
  252. 	 * to the documentation there.
  253. 	 * @returns {Promise.<ProcessResult>} The Promise will be rejected if the process
  254. 	 * exits with a non-zero result or if, when spawning it, an error occurs. In the 
  255. 	 * first case, the result will be of type ProcessExit, while in the latter case, it
  256. 	 * will be of type ProcessErrorResult. Both types inherit from ProcessResult.
  257. 	 */
  258. 	spawnAsync(emitOnly = true) {
  259. 		return new Promise((resolve, reject) => {
  260. 			
  261. 			if (this.isRunning) {
  262. 				throw new Error('The process is already running.');
  263. 			}
  264. 			this._isRunning = true;
  266. 			/**
  267. 			 * @param {ProcessResult} procExit 
  268. 			 */
  269. 			const shutdownFunc = procExit => {
  270. 				this._isRunning = false;
  271. 				this._processResult = procExit;
  273. 				if (procExit instanceof ProcessErrorResult
  274. 					|| (procExit instanceof ProcessExit && procExit.faulted)) {
  275. 					reject(procExit);
  276. 				} else {
  277. 					resolve(procExit);
  278. 				}
  279. 			};
  280. 		
  281. 			/**
  282. 			 * @param {'stdout'|'stderr'} streamName 
  283. 			 * @param {Buffer|String} chunk 
  284. 			 * @param {Array.<String>} streamArr 
  285. 			 */
  286. 			const dataFn = (streamName, chunk, streamArr) => {
  287. 				const out = new ProcessOutput(streamName, chunk);
  288. 				if (!emitOnly) {
  289. 					streamArr.push(out.asString);
  290. 				}
  291. 				this.emit(symbolProcessOutput, out);
  292. 			};
  294. 			const stdOut = [], stdErr = [];
  295. 			const options = mergeObjects({}, this.options);
  296. 			options.stdio = 'pipe';
  298. 			const proc = cp.spawn(this.command, this.args, this.options)
  299. 				.once('error', err =>
  300. 					shutdownFunc(new ProcessErrorResult(err)))
  301. 				.once('exit', (code, sig) =>
  302. 					shutdownFunc(new ProcessExit(
  303. 						code !== 0, code, sig, stdOut.join(''), stdErr.join(''))));
  305. 			this._wasStarted = true;
  307. 			proc.stdout.on('data', chunk => dataFn('stdout', chunk, stdOut));
  308. 			proc.stderr.on('data', chunk => dataFn('stderr', chunk, stdErr));
  310. 			// Expose, after everything is set up:
  311. 			this._process = proc;
  312. 		});
  313. 	};
  314. };
  317. module.exports = Object.freeze({
  318. 	ProcessResult,
  319. 	ProcessErrorResult,
  320. 	ProcessExit,
  321. 	ProcessOutput,
  322. 	ProcessWrapper,
  323. 	symbolProcessOutput
  324. });