Add evaluation helpers (#33)

* Add evaluation helpers

* Make some helpers only available server-side

They don't make any sense on the client side.

* Fork the right thing

packages/protocol/src/browser/client.ts

@@ -1,10 +1,10 @@
 import { EventEmitter } from "events";
 import { Emitter } from "@coder/events";
 import { logger, field } from "@coder/logger";
-import { ReadWriteConnection, InitData, OperatingSystem, SharedProcessData } from "../common/connection";
-import { Disposer, stringify, parse } from "../common/util";
 import { NewEvalMessage, ServerMessage, EvalDoneMessage, EvalFailedMessage, ClientMessage, WorkingInitMessage, EvalEventMessage } from "../proto";
-import { ActiveEval } from "./evaluate";
+import { ReadWriteConnection, InitData, OperatingSystem, SharedProcessData } from "../common/connection";
+import { ActiveEvalHelper, EvalHelper, Disposer, ServerActiveEvalHelper } from "../common/helpers";
+import { stringify, parse } from "../common/util";
 
 /**
  * Client accepts an arbitrary connection intended to communicate with the Server.
@@ -56,13 +56,13 @@ export class Client {
 		return this.initDataPromise;
 	}
 
-	public run(func: (ae: ActiveEval) => Disposer): ActiveEval;
-	public run<T1>(func: (ae: ActiveEval, a1: T1) => Disposer, a1: T1): ActiveEval;
-	public run<T1, T2>(func: (ae: ActiveEval, a1: T1, a2: T2) => Disposer, a1: T1, a2: T2): ActiveEval;
-	public run<T1, T2, T3>(func: (ae: ActiveEval, a1: T1, a2: T2, a3: T3) => Disposer, a1: T1, a2: T2, a3: T3): ActiveEval;
-	public run<T1, T2, T3, T4>(func: (ae: ActiveEval, a1: T1, a2: T2, a3: T3, a4: T4) => Disposer, a1: T1, a2: T2, a3: T3, a4: T4): ActiveEval;
-	public run<T1, T2, T3, T4, T5>(func: (ae: ActiveEval, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5) => Disposer, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5): ActiveEval;
-	public run<T1, T2, T3, T4, T5, T6>(func: (ae: ActiveEval, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5, a6: T6) => Disposer, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5, a6: T6): ActiveEval;
+	public run(func: (helper: ServerActiveEvalHelper) => Disposer): ActiveEvalHelper;
+	public run<T1>(func: (helper: ServerActiveEvalHelper, a1: T1) => Disposer, a1: T1): ActiveEvalHelper;
+	public run<T1, T2>(func: (helper: ServerActiveEvalHelper, a1: T1, a2: T2) => Disposer, a1: T1, a2: T2): ActiveEvalHelper;
+	public run<T1, T2, T3>(func: (helper: ServerActiveEvalHelper, a1: T1, a2: T2, a3: T3) => Disposer, a1: T1, a2: T2, a3: T3): ActiveEvalHelper;
+	public run<T1, T2, T3, T4>(func: (helper: ServerActiveEvalHelper, a1: T1, a2: T2, a3: T3, a4: T4) => Disposer, a1: T1, a2: T2, a3: T3, a4: T4): ActiveEvalHelper;
+	public run<T1, T2, T3, T4, T5>(func: (helper: ServerActiveEvalHelper, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5) => Disposer, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5): ActiveEvalHelper;
+	public run<T1, T2, T3, T4, T5, T6>(func: (helper: ServerActiveEvalHelper, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5, a6: T6) => Disposer, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5, a6: T6): ActiveEvalHelper;
 	/**
 	 * Run a function on the server and provide an event emitter which allows
 	 * listening and emitting to the emitter provided to that function. The
@@ -70,7 +70,7 @@ export class Client {
 	 * disconnects and for notifying when disposal has happened outside manual
 	 * activation.
 	 */
-	public run<T1, T2, T3, T4, T5, T6>(func: (ae: ActiveEval, a1?: T1, a2?: T2, a3?: T3, a4?: T4, a5?: T5, a6?: T6) => Disposer, a1?: T1, a2?: T2, a3?: T3, a4?: T4, a5?: T5, a6?: T6): ActiveEval {
+	public run<T1, T2, T3, T4, T5, T6>(func: (helper: ServerActiveEvalHelper, a1?: T1, a2?: T2, a3?: T3, a4?: T4, a5?: T5, a6?: T6) => Disposer, a1?: T1, a2?: T2, a3?: T3, a4?: T4, a5?: T5, a6?: T6): ActiveEvalHelper {
 		const doEval = this.doEvaluate(func, a1, a2, a3, a4, a5, a6, true);
 
 		// This takes server events and emits them to the client's emitter.
@@ -89,9 +89,9 @@ export class Client {
 			eventEmitter.emit("error", ex);
 		});
 
-		// This takes client events and emits them to the server's emitter and
-		// listens to events received from the server (via the event hook above).
-		return {
+		return new ActiveEvalHelper({
+			// This takes client events and emits them to the server's emitter and
+			// listens to events received from the server (via the event hook above).
 			// tslint:disable no-any
 			on: (event: string, cb: (...args: any[]) => void): EventEmitter => eventEmitter.on(event, cb),
 			emit: (event: string, ...args: any[]): void => {
@@ -105,21 +105,21 @@ export class Client {
 			},
 			removeAllListeners: (event: string): EventEmitter => eventEmitter.removeAllListeners(event),
 			// tslint:enable no-any
-		};
+		});
 	}
 
-	public evaluate<R>(func: () => R | Promise<R>): Promise<R>;
-	public evaluate<R, T1>(func: (a1: T1) => R | Promise<R>, a1: T1): Promise<R>;
-	public evaluate<R, T1, T2>(func: (a1: T1, a2: T2) => R | Promise<R>, a1: T1, a2: T2): Promise<R>;
-	public evaluate<R, T1, T2, T3>(func: (a1: T1, a2: T2, a3: T3) => R | Promise<R>, a1: T1, a2: T2, a3: T3): Promise<R>;
-	public evaluate<R, T1, T2, T3, T4>(func: (a1: T1, a2: T2, a3: T3, a4: T4) => R | Promise<R>, a1: T1, a2: T2, a3: T3, a4: T4): Promise<R>;
-	public evaluate<R, T1, T2, T3, T4, T5>(func: (a1: T1, a2: T2, a3: T3, a4: T4, a5: T5) => R | Promise<R>, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5): Promise<R>;
-	public evaluate<R, T1, T2, T3, T4, T5, T6>(func: (a1: T1, a2: T2, a3: T3, a4: T4, a5: T5, a6: T6) => R | Promise<R>, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5, a6: T6): Promise<R>;
+	public evaluate<R>(func: (helper: EvalHelper) => R | Promise<R>): Promise<R>;
+	public evaluate<R, T1>(func: (helper: EvalHelper, a1: T1) => R | Promise<R>, a1: T1): Promise<R>;
+	public evaluate<R, T1, T2>(func: (helper: EvalHelper, a1: T1, a2: T2) => R | Promise<R>, a1: T1, a2: T2): Promise<R>;
+	public evaluate<R, T1, T2, T3>(func: (helper: EvalHelper, a1: T1, a2: T2, a3: T3) => R | Promise<R>, a1: T1, a2: T2, a3: T3): Promise<R>;
+	public evaluate<R, T1, T2, T3, T4>(func: (helper: EvalHelper, a1: T1, a2: T2, a3: T3, a4: T4) => R | Promise<R>, a1: T1, a2: T2, a3: T3, a4: T4): Promise<R>;
+	public evaluate<R, T1, T2, T3, T4, T5>(func: (helper: EvalHelper, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5) => R | Promise<R>, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5): Promise<R>;
+	public evaluate<R, T1, T2, T3, T4, T5, T6>(func: (helper: EvalHelper, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5, a6: T6) => R | Promise<R>, a1: T1, a2: T2, a3: T3, a4: T4, a5: T5, a6: T6): Promise<R>;
 	/**
 	 * Evaluates a function on the server.
 	 * To pass variables, ensure they are serializable and passed through the included function.
 	 * @example
-	 * const returned = await this.client.evaluate((value) => {
+	 * const returned = await this.client.evaluate((helper, value) => {
 	 *     return value;
 	 * }, "hi");
 	 * console.log(returned);
@@ -127,7 +127,7 @@ export class Client {
 	 * @param func Function to evaluate
 	 * @returns Promise rejected or resolved from the evaluated function
 	 */
-	public evaluate<R, T1, T2, T3, T4, T5, T6>(func: (a1?: T1, a2?: T2, a3?: T3, a4?: T4, a5?: T5, a6?: T6) => R | Promise<R>, a1?: T1, a2?: T2, a3?: T3, a4?: T4, a5?: T5, a6?: T6): Promise<R> {
+	public evaluate<R, T1, T2, T3, T4, T5, T6>(func: (helper: EvalHelper, a1?: T1, a2?: T2, a3?: T3, a4?: T4, a5?: T5, a6?: T6) => R | Promise<R>, a1?: T1, a2?: T2, a3?: T3, a4?: T4, a5?: T5, a6?: T6): Promise<R> {
 		return this.doEvaluate(func, a1, a2, a3, a4, a5, a6, false).completed;
 	}
 

packages/protocol/src/browser/evaluate.ts

@@ -1,8 +0,0 @@
-export interface ActiveEval {
-	removeAllListeners(event?: string): void;
-
-	// tslint:disable no-any
-	emit(event: string, ...args: any[]): void;
-	on(event: string, cb: (...args: any[]) => void): void;
-	// tslint:disable no-any
-}

packages/protocol/src/common/helpers.ts

@@ -0,0 +1,399 @@
+import { ChildProcess, SpawnOptions, ForkOptions } from "child_process";
+import { EventEmitter } from "events";
+import { Socket } from "net";
+import { Duplex, Readable, Writable } from "stream";
+import { IDisposable } from "@coder/disposable";
+import { logger } from "@coder/logger";
+
+// tslint:disable no-any
+
+export type ForkProvider = (modulePath: string, args: string[], options: ForkOptions, dataDir?: string) => ChildProcess;
+
+export interface Disposer extends IDisposable {
+	onDidDispose: (cb: () => void) => void;
+}
+
+interface ActiveEvalEmitter {
+	removeAllListeners(event?: string): void;
+	emit(event: string, ...args: any[]): void;
+	on(event: string, cb: (...args: any[]) => void): void;
+}
+
+/**
+ * Helper class for evaluations.
+ */
+export class EvalHelper {
+	/**
+	 * Some spawn code tries to preserve the env (the debug adapter for instance)
+	 * but the env is mostly blank (since we're in the browser), so we'll just
+	 * always preserve the main process.env here, otherwise it won't have access
+	 * to PATH, etc.
+	 * TODO: An alternative solution would be to send the env to the browser?
+	*/
+	public preserveEnv(options: SpawnOptions | ForkOptions): void {
+		if (options && options.env) {
+			options.env = { ...process.env, ...options.env };
+		}
+	}
+}
+
+/**
+ * Helper class for active evaluations.
+ */
+export class ActiveEvalHelper extends EvalHelper implements ActiveEvalEmitter {
+	public constructor(private readonly emitter: ActiveEvalEmitter) {
+		super();
+	}
+
+	public removeAllListeners(event?: string): void {
+		this.emitter.removeAllListeners(event);
+	}
+
+	public emit(event: string, ...args: any[]): void {
+		this.emitter.emit(event, ...args);
+	}
+
+	public on(event: string, cb: (...args: any[]) => void): void {
+		this.emitter.on(event, cb);
+	}
+
+	/**
+	 * Create a new helper to make unique events for an item.
+	 */
+	public createUnique(id: number | "stdout" | "stderr" | "stdin"): ActiveEvalHelper {
+		return new ActiveEvalHelper(this.createUniqueEmitter(id));
+	}
+
+	/**
+	 * Wrap the evaluation emitter to make unique events for an item to prevent
+	 * conflicts when it shares that emitter with other items.
+	 */
+	protected createUniqueEmitter(id: number | "stdout" | "stderr" | "stdin"): ActiveEvalEmitter {
+		let events = <string[]>[];
+
+		return {
+			removeAllListeners: (event?: string): void => {
+				if (!event) {
+					events.forEach((e) => this.removeAllListeners(e));
+					events = [];
+				} else {
+					const index = events.indexOf(event);
+					if (index !== -1) {
+						events.splice(index, 1);
+						this.removeAllListeners(`${event}:${id}`);
+					}
+				}
+			},
+			emit: (event: string, ...args: any[]): void => {
+				this.emit(`${event}:${id}`, ...args);
+			},
+			on: (event: string, cb: (...args: any[]) => void): void => {
+				if (!events.includes(event)) {
+					events.push(event);
+				}
+				this.on(`${event}:${id}`, cb);
+			},
+		};
+	}
+}
+
+/**
+ * Helper class for server-side active evaluations.
+ */
+export class ServerActiveEvalHelper extends ActiveEvalHelper {
+	public constructor(emitter: ActiveEvalEmitter, public readonly fork: ForkProvider) {
+		super(emitter);
+	}
+
+	/**
+	 * If there is a callback ID, return a function that emits the callback event
+	 * on the active evaluation with that ID and all arguments passed to it.
+	 * Otherwise, return undefined.
+	 */
+	public maybeCallback(callbackId?: number): ((...args: any[])  => void) | undefined {
+		return typeof callbackId !== "undefined" ? (...args: any[]): void => {
+			this.emit("callback", callbackId, ...args);
+		} : undefined;
+	}
+
+	/**
+	 * Bind a socket to an active evaluation and returns a disposer.
+	 */
+	public bindSocket(socket: Socket): Disposer {
+		socket.on("connect", () => this.emit("connect"));
+		socket.on("lookup", (error, address, family, host) => this.emit("lookup", error, address, family, host));
+		socket.on("timeout", () => this.emit("timeout"));
+
+		this.on("connect", (options, callbackId) => socket.connect(options, this.maybeCallback(callbackId)));
+		this.on("ref", () => socket.ref());
+		this.on("setKeepAlive", (enable, initialDelay) => socket.setKeepAlive(enable, initialDelay));
+		this.on("setNoDelay", (noDelay) => socket.setNoDelay(noDelay));
+		this.on("setTimeout", (timeout, callbackId) => socket.setTimeout(timeout, this.maybeCallback(callbackId)));
+		this.on("unref", () => socket.unref());
+
+		this.bindReadable(socket);
+		this.bindWritable(socket);
+
+		return {
+			onDidDispose: (cb): Socket => socket.on("close", cb),
+			dispose: (): void => {
+				socket.removeAllListeners();
+				socket.end();
+				socket.destroy();
+				socket.unref();
+			},
+		};
+	}
+
+	/**
+	 * Bind a writable stream to the active evaluation.
+	 */
+	public bindWritable(writable: Writable | Duplex): void {
+		if (!((writable as Readable).read)) { // To avoid binding twice.
+			writable.on("close", () => this.emit("close"));
+			writable.on("error", (error) => this.emit("error", error));
+
+			this.on("destroy", () => writable.destroy());
+		}
+
+		writable.on("drain", () => this.emit("drain"));
+		writable.on("finish", () => this.emit("finish"));
+		writable.on("pipe", () => this.emit("pipe"));
+		writable.on("unpipe", () => this.emit("unpipe"));
+
+		this.on("cork", () => writable.cork());
+		this.on("end", (chunk, encoding, callbackId) => writable.end(chunk, encoding, this.maybeCallback(callbackId)));
+		this.on("setDefaultEncoding", (encoding) => writable.setDefaultEncoding(encoding));
+		this.on("uncork", () => writable.uncork());
+		// Sockets can pass an fd instead of a callback but streams cannot.
+		this.on("write", (chunk, encoding, fd, callbackId) => writable.write(chunk, encoding, this.maybeCallback(callbackId) || fd));
+	}
+
+	/**
+	 * Bind a readable stream to the active evaluation.
+	 */
+	public bindReadable(readable: Readable): void {
+		// Streams don't have an argument on close but sockets do.
+		readable.on("close", (...args: any[]) => this.emit("close", ...args));
+		readable.on("data", (data) => this.emit("data", data));
+		readable.on("end", () => this.emit("end"));
+		readable.on("error", (error) => this.emit("error", error));
+		readable.on("readable", () => this.emit("readable"));
+
+		this.on("destroy", () => readable.destroy());
+		this.on("pause", () => readable.pause());
+		this.on("push", (chunk, encoding) => readable.push(chunk, encoding));
+		this.on("resume", () => readable.resume());
+		this.on("setEncoding", (encoding) => readable.setEncoding(encoding));
+		this.on("unshift", (chunk) => readable.unshift(chunk));
+	}
+
+	public createUnique(id: number | "stdout" | "stderr" | "stdin"): ServerActiveEvalHelper {
+		return new ServerActiveEvalHelper(this.createUniqueEmitter(id), this.fork);
+	}
+}
+
+/**
+ * An event emitter that can store callbacks with IDs in a map so we can pass
+ * them back and forth through an active evaluation using those IDs.
+ */
+export class CallbackEmitter extends EventEmitter {
+	private _ae: ActiveEvalHelper | undefined;
+	private callbackId = 0;
+	private readonly callbacks = new Map<number, Function>();
+
+	public constructor(ae?: ActiveEvalHelper) {
+		super();
+		if (ae) {
+			this.ae = ae;
+		}
+	}
+
+	protected get ae(): ActiveEvalHelper {
+		if (!this._ae) {
+			throw new Error("trying to access active evaluation before it has been set");
+		}
+
+		return this._ae;
+	}
+
+	protected set ae(ae: ActiveEvalHelper) {
+		if (this._ae) {
+			throw new Error("cannot override active evaluation");
+		}
+		this._ae = ae;
+		this.ae.on("callback", (callbackId, ...args: any[]) => this.runCallback(callbackId, ...args));
+	}
+
+	/**
+	 * Store the callback and return and ID referencing its location in the map.
+	 */
+	protected storeCallback(callback?: Function): number | undefined {
+		if (!callback) {
+			return undefined;
+		}
+
+		const callbackId = this.callbackId++;
+		this.callbacks.set(callbackId, callback);
+
+		return callbackId;
+	}
+
+	/**
+	 * Call the function with the specified ID and delete it from the map.
+	 * If the ID is undefined or doesn't exist, nothing happens.
+	 */
+	private runCallback(callbackId?: number, ...args: any[]): void {
+		const callback = typeof callbackId !== "undefined" && this.callbacks.get(callbackId);
+		if (callback && typeof callbackId !== "undefined") {
+			this.callbacks.delete(callbackId);
+			callback(...args);
+		}
+	}
+}
+
+/**
+ * A writable stream over an active evaluation.
+ */
+export class ActiveEvalWritable extends CallbackEmitter implements Writable {
+	public constructor(ae: ActiveEvalHelper) {
+		super(ae);
+		// Streams don't have an argument on close but sockets do.
+		this.ae.on("close", (...args: any[]) => this.emit("close", ...args));
+		this.ae.on("drain", () => this.emit("drain"));
+		this.ae.on("error", (error) => this.emit("error", error));
+		this.ae.on("finish", () => this.emit("finish"));
+		this.ae.on("pipe", () => logger.warn("pipe is not supported"));
+		this.ae.on("unpipe", () => logger.warn("unpipe is not supported"));
+	}
+
+	public get writable(): boolean { throw new Error("not implemented"); }
+	public get writableHighWaterMark(): number { throw new Error("not implemented"); }
+	public get writableLength(): number { throw new Error("not implemented"); }
+	public _write(): void { throw new Error("not implemented"); }
+	public _destroy(): void { throw new Error("not implemented"); }
+	public _final(): void { throw new Error("not implemented"); }
+	public pipe<T>(): T { throw new Error("not implemented"); }
+
+	public cork(): void { this.ae.emit("cork"); }
+	public destroy(): void { this.ae.emit("destroy"); }
+	public setDefaultEncoding(encoding: string): this {
+		this.ae.emit("setDefaultEncoding", encoding);
+
+		return this;
+	}
+	public uncork(): void { this.ae.emit("uncork"); }
+
+	public write(chunk: any, encoding?: string | ((error?: Error | null) => void), callback?: (error?: Error | null) => void): boolean {
+		if (typeof encoding === "function") {
+			callback = encoding;
+			encoding = undefined;
+		}
+
+		// Sockets can pass an fd instead of a callback but streams cannot..
+		this.ae.emit("write", chunk, encoding, undefined, this.storeCallback(callback));
+
+		// Always true since we can't get this synchronously.
+		return true;
+	}
+
+	public end(data?: any, encoding?: string | Function, callback?: Function): void {
+		if (typeof encoding === "function") {
+			callback = encoding;
+			encoding = undefined;
+		}
+		this.ae.emit("end", data, encoding, this.storeCallback(callback));
+	}
+}
+
+/**
+ * A readable stream over an active evaluation.
+ */
+export class ActiveEvalReadable extends CallbackEmitter implements Readable {
+	public constructor(ae: ActiveEvalHelper) {
+		super(ae);
+		this.ae.on("close", () => this.emit("close"));
+		this.ae.on("data", (data) => this.emit("data", data));
+		this.ae.on("end", () => this.emit("end"));
+		this.ae.on("error", (error) => this.emit("error", error));
+		this.ae.on("readable", () => this.emit("readable"));
+	}
+
+	public get readable(): boolean { throw new Error("not implemented"); }
+	public get readableHighWaterMark(): number { throw new Error("not implemented"); }
+	public get readableLength(): number { throw new Error("not implemented"); }
+	public _read(): void { throw new Error("not implemented"); }
+	public read(): any { throw new Error("not implemented"); }
+	public isPaused(): boolean { throw new Error("not implemented"); }
+	public pipe<T>(): T { throw new Error("not implemented"); }
+	public unpipe(): this { throw new Error("not implemented"); }
+	public unshift(): this { throw new Error("not implemented"); }
+	public wrap(): this { throw new Error("not implemented"); }
+	public push(): boolean { throw new Error("not implemented"); }
+	public _destroy(): void { throw new Error("not implemented"); }
+	public [Symbol.asyncIterator](): AsyncIterableIterator<any> { throw new Error("not implemented"); }
+
+	public destroy(): void { this.ae.emit("destroy"); }
+	public pause(): this { return this.emitReturnThis("pause"); }
+	public resume(): this { return this.emitReturnThis("resume"); }
+	public setEncoding(encoding?: string): this { return this.emitReturnThis("setEncoding", encoding); }
+
+	// tslint:disable-next-line no-any
+	protected emitReturnThis(event: string, ...args: any[]): this {
+		this.ae.emit(event, ...args);
+
+		return this;
+	}
+}
+
+/**
+ * An duplex stream over an active evaluation.
+ */
+export class ActiveEvalDuplex extends ActiveEvalReadable implements Duplex {
+	// Some unfortunate duplication here since we can't have multiple extends.
+	public constructor(ae: ActiveEvalHelper) {
+		super(ae);
+		this.ae.on("drain", () => this.emit("drain"));
+		this.ae.on("finish", () => this.emit("finish"));
+		this.ae.on("pipe", () => logger.warn("pipe is not supported"));
+		this.ae.on("unpipe", () => logger.warn("unpipe is not supported"));
+	}
+
+	public get writable(): boolean { throw new Error("not implemented"); }
+	public get writableHighWaterMark(): number { throw new Error("not implemented"); }
+	public get writableLength(): number { throw new Error("not implemented"); }
+	public _write(): void { throw new Error("not implemented"); }
+	public _destroy(): void { throw new Error("not implemented"); }
+	public _final(): void { throw new Error("not implemented"); }
+	public pipe<T>(): T { throw new Error("not implemented"); }
+
+	public cork(): void { this.ae.emit("cork"); }
+	public destroy(): void { this.ae.emit("destroy"); }
+	public setDefaultEncoding(encoding: string): this {
+		this.ae.emit("setDefaultEncoding", encoding);
+
+		return this;
+	}
+	public uncork(): void { this.ae.emit("uncork"); }
+
+	public write(chunk: any, encoding?: string | ((error?: Error | null) => void), callback?: (error?: Error | null) => void): boolean {
+		if (typeof encoding === "function") {
+			callback = encoding;
+			encoding = undefined;
+		}
+
+		// Sockets can pass an fd instead of a callback but streams cannot..
+		this.ae.emit("write", chunk, encoding, undefined, this.storeCallback(callback));
+
+		// Always true since we can't get this synchronously.
+		return true;
+	}
+
+	public end(data?: any, encoding?: string | Function, callback?: Function): void {
+		if (typeof encoding === "function") {
+			callback = encoding;
+			encoding = undefined;
+		}
+		this.ae.emit("end", data, encoding, this.storeCallback(callback));
+	}
+}

packages/protocol/src/common/util.ts

@@ -1,5 +1,3 @@
-import { IDisposable } from "@coder/disposable";
-
 /**
  * Return true if we're in a browser environment (including web workers).
  */
@@ -84,7 +82,3 @@ export const parse = (arg: string): any => { // tslint:disable-line no-any
 
 	return result;
 };
-
-export interface Disposer extends IDisposable {
-	onDidDispose: (cb: () => void) => void;
-}

packages/protocol/src/index.ts

@@ -1,4 +1,4 @@
 export * from "./browser/client";
-export * from "./browser/evaluate";
 export * from "./common/connection";
+export * from "./common/helpers";
 export * from "./common/util";

packages/protocol/src/node/evaluate.ts

@@ -1,8 +1,10 @@
+import { fork as cpFork } from "child_process";
 import { EventEmitter } from "events";
 import * as vm from "vm";
 import { logger, field } from "@coder/logger";
 import { NewEvalMessage, EvalFailedMessage, EvalDoneMessage, ServerMessage, EvalEventMessage } from "../proto";
 import { SendableConnection } from "../common/connection";
+import { ServerActiveEvalHelper, EvalHelper, ForkProvider } from "../common/helpers";
 import { stringify, parse } from "../common/util";
 
 export interface ActiveEvaluation {
@@ -11,7 +13,7 @@ export interface ActiveEvaluation {
 }
 
 declare var __non_webpack_require__: typeof require;
-export const evaluate = (connection: SendableConnection, message: NewEvalMessage, onDispose: () => void): ActiveEvaluation | void => {
+export const evaluate = (connection: SendableConnection, message: NewEvalMessage, onDispose: () => void, fork?: ForkProvider): ActiveEvaluation | void => {
 	/**
 	 * Send the response and call onDispose.
 	 */
@@ -46,7 +48,10 @@ export const evaluate = (connection: SendableConnection, message: NewEvalMessage
 
 	let eventEmitter = message.getActive() ? new EventEmitter(): undefined;
 	const sandbox = {
-		eventEmitter: eventEmitter ? {
+		helper: eventEmitter ? new ServerActiveEvalHelper({
+			removeAllListeners: (event?: string): void => {
+				eventEmitter!.removeAllListeners(event);
+			},
 			// tslint:disable no-any
 			on: (event: string, cb: (...args: any[]) => void): void => {
 				eventEmitter!.on(event, (...args: any[]) => {
@@ -73,7 +78,7 @@ export const evaluate = (connection: SendableConnection, message: NewEvalMessage
 				connection.send(serverMsg.serializeBinary());
 			},
 			// tslint:enable no-any
-		} : undefined,
+		}, fork || cpFork) : new EvalHelper(),
 		_Buffer: Buffer,
 		// When the client is ran from Webpack, it will replace
 		// __non_webpack_require__ with require, which we then need to provide to
@@ -94,7 +99,7 @@ export const evaluate = (connection: SendableConnection, message: NewEvalMessage
 
 	let value: any; // tslint:disable-line no-any
 	try {
-		const code = `(${message.getFunction()})(${eventEmitter ? "eventEmitter, " : ""}...args);`;
+		const code = `(${message.getFunction()})(helper, ...args);`;
 		value = vm.runInNewContext(code, sandbox, {
 			// If the code takes longer than this to return, it is killed and throws.
 			timeout: message.getTimeout() || 15000,

packages/protocol/src/node/server.ts

@@ -5,12 +5,14 @@ import { promisify } from "util";
 import { logger, field } from "@coder/logger";
 import { ClientMessage, WorkingInitMessage, ServerMessage } from "../proto";
 import { evaluate, ActiveEvaluation } from "./evaluate";
+import { ForkProvider } from "../common/helpers";
 import { ReadWriteConnection } from "../common/connection";
 
 export interface ServerOptions {
 	readonly workingDirectory: string;
 	readonly dataDirectory: string;
 	readonly builtInExtensionsDirectory: string;
+	readonly fork?: ForkProvider;
 }
 
 export class Server {
@@ -105,7 +107,7 @@ export class Server {
 				logger.trace(() => [
 					`dispose ${evalMessage.getId()}, ${this.evals.size} left`,
 				]);
-			});
+			}, this.options ? this.options.fork : undefined);
 			if (resp) {
 				this.evals.set(evalMessage.getId(), resp);
 			}

packages/protocol/test/evaluate.test.ts

@@ -13,7 +13,7 @@ describe("Evaluate", () => {
 
 	it("should compute from string", async () => {
 		const start = "ban\%\$\"``a,,,,asdasd";
-		const value = await client.evaluate((a) => {
+		const value = await client.evaluate((_helper, a) => {
 			return a;
 		}, start);
 
@@ -21,7 +21,7 @@ describe("Evaluate", () => {
 	}, 100);
 
 	it("should compute from object", async () => {
-		const value = await client.evaluate((arg) => {
+		const value = await client.evaluate((_helper, arg) => {
 			return arg.bananas * 2;
 		}, { bananas: 1 });