Get search working and clean up disconnected client (#23)

* Use ipc instead of pipe

* Run callback passed to child process's send method

* It also returns true

* Correct send signature

* Kill processes when client disconnects

Author: code-asher

packages/protocol/src/browser/command.ts

@@ -30,8 +30,8 @@ export interface ChildProcess {
 
 	kill(signal?: string): void;
 
-	send(message: string | Uint8Array, ipc?: false): void;
-	send(message: any, ipc: true): void;
+	send(message: string | Uint8Array, callback?: () => void, ipc?: false): void;
+	send(message: any, callback: undefined | (() => void), ipc: true): void;
 
 	on(event: "message", listener: (data: any) => void): void;
 	on(event: "error", listener: (err: Error) => void): void;
@@ -101,7 +101,7 @@ export class ServerProcess extends events.EventEmitter implements ChildProcess {
 		this._connected = false;
 	}
 
-	public send(message: string | Uint8Array | any, ipc: boolean = this.ipc): void {
+	public send(message: string | Uint8Array | any, callback?: (error: Error | null) => void, ipc: boolean = this.ipc): boolean {
 		const send = new WriteToSessionMessage();
 		send.setId(this.id);
 		send.setSource(ipc ? WriteToSessionMessage.Source.IPC : WriteToSessionMessage.Source.STDIN);
@@ -113,6 +113,12 @@ export class ServerProcess extends events.EventEmitter implements ChildProcess {
 		const client = new ClientMessage();
 		client.setWriteToSession(send);
 		this.connection.send(client.serializeBinary());
+		// TODO: properly implement?
+		if (callback) {
+			callback(null);
+		}
+
+		return true;
 	}
 
 	public resize(dimensions: TTYDimensions): void {

packages/protocol/src/node/command.ts

@@ -13,11 +13,12 @@ export interface Process {
 	stdin?: stream.Writable;
 	stdout?: stream.Readable;
 	stderr?: stream.Readable;
+	send?: (message: string) => void;
 
 	pid: number;
 	killed?: boolean;
 
-	on(event: "data", cb: (data: string) => void): void;
+	on(event: "data" | "message", cb: (data: string) => void): void;
 	on(event: "exit", listener: (exitCode: number, signal?: number) => void): void;
 	write(data: string | Uint8Array): void;
 	resize?(cols: number, rows: number): void;
@@ -61,7 +62,7 @@ export const handleNewSession = (connection: SendableConnection, newSession: New
 				connection.send(sm.serializeBinary());
 			}
 		}, 200);
-		
+
 		ptyProc.on("exit", () => {
 			clearTimeout(timer);
 		});
@@ -93,6 +94,9 @@ export const handleNewSession = (connection: SendableConnection, newSession: New
 			stderr: proc.stderr,
 			stdout: proc.stdout,
 			stdio: proc.stdio,
+			send: (message): void => {
+				proc.send(message);
+			},
 			on: (...args: any[]): void => ((proc as any).on)(...args), // tslint:disable-line no-any
 			write: (d): boolean => proc.stdin.write(d),
 			kill: (s): void => proc.kill(s || "SIGTERM"),
@@ -105,7 +109,7 @@ export const handleNewSession = (connection: SendableConnection, newSession: New
 
 			let data = msg.toString();
 			if (_source === SessionOutputMessage.Source.IPC) {
-				data = Buffer.from(msg.toString(), "base64").toString();
+				// data = Buffer.from(msg.toString(), "base64").toString();
 			}
 
 			return [
@@ -139,10 +143,10 @@ export const handleNewSession = (connection: SendableConnection, newSession: New
 		});
 	}
 
-	if (process.stdio && process.stdio[3]) {
-		// We have ipc fd
-		process.stdio[3].on("data", (data) => {
-			sendOutput(SessionOutputMessage.Source.IPC, data);
+	// IPC.
+	if (process.send) {
+		process.on("message", (data) => {
+			sendOutput(SessionOutputMessage.Source.IPC, JSON.stringify(data));
 		});
 	}
 

packages/protocol/src/node/server.ts

@@ -1,7 +1,7 @@
 import * as os from "os";
 import * as cp from "child_process";
 import * as path from "path";
-import { mkdir, WriteStream } from "fs";
+import { mkdir } from "fs";
 import { promisify } from "util";
 import { TextDecoder } from "text-encoding";
 import { logger, field } from "@coder/logger";
@@ -37,6 +37,17 @@ export class Server {
 				logger.error("Failed to handle client message", field("length", data.byteLength), field("exception", ex));
 			}
 		});
+		connection.onClose(() => {
+			this.sessions.forEach((s) => {
+				s.kill();
+			});
+			this.connections.forEach((c) => {
+				c.destroy();
+			});
+			this.servers.forEach((s) => {
+				s.close();
+			});
+		});
 
 		if (!options) {
 			logger.warn("No server options provided. InitMessage will not be sent.");
@@ -97,7 +108,12 @@ export class Server {
 	private handleMessage(message: ClientMessage): void {
 		if (message.hasNewEval()) {
 			const evalMessage = message.getNewEval()!;
-			logger.debug("EvalMessage", field("id", evalMessage.getId()));
+			logger.debug(() => [
+				"EvalMessage",
+				field("id", evalMessage.getId()),
+				field("args", evalMessage.getArgsList()),
+				field("function", evalMessage.getFunction()),
+			]);
 			evaluate(this.connection, evalMessage);
 		} else if (message.hasNewSession()) {
 			const sessionMessage = message.getNewSession()!;
@@ -141,10 +157,10 @@ export class Server {
 			const data = new TextDecoder().decode(writeToSessionMessage.getData_asU8());
 			const source = writeToSessionMessage.getSource();
 			if (source === WriteToSessionMessage.Source.IPC) {
-				if (!s.stdio || !s.stdio[3]) {
+				if (!s.send) {
 					throw new Error("Cannot send message via IPC to process without IPC");
 				}
-				(s.stdio[3] as WriteStream).write(data);
+				s.send(JSON.parse(data));
 			} else {
 				s.write(data);
 			}

packages/protocol/test/command.test.ts

@@ -15,7 +15,7 @@ describe("spawn", () => {
 		workingDirectory: "",
 		forkProvider: (msg): cp.ChildProcess => {
 			return cp.spawn(msg.getCommand(), msg.getArgsList(), {
-				stdio: [null, null, null, "pipe"],
+				stdio: [null, null, null, "ipc"],
 			});
 		},
 	});
@@ -144,7 +144,7 @@ describe("spawn", () => {
 			expect(msg.bananas).toBeTruthy();
 			proc.kill();
 		});
-		proc.send({ bananas: true }, true);
+		proc.send({ bananas: true }, undefined, true);
 		proc.on("exit", () => done());
 	});
 });

packages/server/src/vscode/bootstrapFork.ts

@@ -1,30 +1,18 @@
 import * as cp from "child_process";
 import * as fs from "fs";
-import * as net from "net";
 import * as path from "path";
 
 export const requireModule = (modulePath: string): void => {
 	process.env.AMD_ENTRYPOINT = modulePath;
 
 	const xml = require("xhr2");
 
-	(<any>global).XMLHttpRequest = xml.XMLHttpRequest;
+	// tslint:disable-next-line no-any this makes installing extensions work.
+	(global as any).XMLHttpRequest = xml.XMLHttpRequest;
 
 	// Always do this so we can see console.logs.
 	// process.env.VSCODE_ALLOW_IO = "true";
 
-	if (!process.send) {
-		const socket = new net.Socket({ fd: 3 });
-		socket.on("data", (data) => {
-			process.emit("message", JSON.parse(data.toString()), undefined);
-		});
-
-		// tslint:disable-next-line no-any
-		process.send = (message: any): void => {
-			socket.write(JSON.stringify(message));
-		};
-	}
-
 	const content = fs.readFileSync(path.join(process.env.BUILD_DIR as string || path.join(__dirname, "../.."), "./build/bootstrap-fork.js"));
 	eval(content.toString());
 };
@@ -45,7 +33,7 @@ export const forkModule = (modulePath: string, env?: NodeJS.ProcessEnv): cp.Chil
 		args.push("--env", JSON.stringify(env));
 	}
 	const options: cp.SpawnOptions = {
-		stdio: [null, null, null, "pipe"],
+		stdio: [null, null, null, "ipc"],
 	};
 	if (process.env.CLI === "true") {
 		proc = cp.spawn(process.execPath, args, options);

packages/web/yarn.lock

@@ -0,0 +1,4 @@
+# THIS IS AN AUTOGENERATED FILE. DO NOT EDIT THIS FILE DIRECTLY.
+# yarn lockfile v1
+
+