Synchronously bind to proxy events

This eliminates any chance whatsoever of missing events due to binding
too late.

Author: code-asher

packages/protocol/src/browser/modules/child_process.ts

@@ -2,15 +2,13 @@ import * as cp from "child_process";
 import * as net from "net";
 import * as stream from "stream";
 import { callbackify } from "util";
-import { ClientProxy, Module } from "../../common/proxy";
+import { ClientProxy, ClientServerProxy } from "../../common/proxy";
 import { ChildProcessModuleProxy, ChildProcessProxy } from "../../node/modules/child_process";
 import { ClientWritableProxy, ClientReadableProxy, Readable, Writable } from "./stream";
 
 // tslint:disable completed-docs
 
-export interface ClientChildProcessProxy extends ChildProcessProxy {
-	proxyId: number | Module;
-}
+export interface ClientChildProcessProxy extends ChildProcessProxy, ClientServerProxy {}
 
 export interface ClientChildProcessProxies {
 	childProcess: ClientChildProcessProxy;
@@ -110,8 +108,7 @@ export class ChildProcess extends ClientProxy<ClientChildProcessProxy> implement
 	}
 }
 
-interface ClientChildProcessModuleProxy extends ChildProcessModuleProxy {
-	proxyId: number | Module;
+interface ClientChildProcessModuleProxy extends ChildProcessModuleProxy, ClientServerProxy {
 	exec(command: string, options?: { encoding?: string | null } & cp.ExecOptions | null, callback?: ((error: cp.ExecException | null, stdin: string | Buffer, stdout: string | Buffer) => void)): Promise<ClientChildProcessProxies>;
 	fork(modulePath: string, args?: string[], options?: cp.ForkOptions): Promise<ClientChildProcessProxies>;
 	spawn(command: string, args?: string[], options?: cp.SpawnOptions): Promise<ClientChildProcessProxies>;

packages/protocol/src/browser/modules/fs.ts

@@ -1,6 +1,6 @@
 import * as fs from "fs";
 import { callbackify } from "util";
-import { Batch, ClientProxy, Module } from "../../common/proxy";
+import { Batch, ClientProxy, ClientServerProxy } from "../../common/proxy";
 import { IEncodingOptions, IEncodingOptionsCallback } from "../../common/util";
 import { FsModuleProxy, ReadStreamProxy, Stats as IStats, WatcherProxy, WriteStreamProxy } from "../../node/modules/fs";
 import { Readable, Writable  } from "./stream";
@@ -38,9 +38,7 @@ class ReaddirBatch extends Batch<Buffer[] | fs.Dirent[] | string[], { path: fs.P
 	}
 }
 
-interface ClientWatcherProxy extends WatcherProxy {
-	proxyId: number | Module;
-}
+interface ClientWatcherProxy extends WatcherProxy, ClientServerProxy {}
 
 class Watcher extends ClientProxy<ClientWatcherProxy> implements fs.FSWatcher {
 	public close(): void {
@@ -52,9 +50,7 @@ class Watcher extends ClientProxy<ClientWatcherProxy> implements fs.FSWatcher {
 	}
 }
 
-interface ClientReadStreamProxy extends ReadStreamProxy {
-	proxyId: number | Module;
-}
+interface ClientReadStreamProxy extends ReadStreamProxy, ClientServerProxy {}
 
 class ReadStream extends Readable<ClientReadStreamProxy> implements fs.ReadStream {
 	public get bytesRead(): number {
@@ -70,9 +66,7 @@ class ReadStream extends Readable<ClientReadStreamProxy> implements fs.ReadStrea
 	}
 }
 
-interface ClientWriteStreamProxy extends WriteStreamProxy {
-	proxyId: number | Module;
-}
+interface ClientWriteStreamProxy extends WriteStreamProxy, ClientServerProxy {}
 
 class WriteStream extends Writable<ClientWriteStreamProxy> implements fs.WriteStream {
 	public get bytesWritten(): number {
@@ -88,8 +82,7 @@ class WriteStream extends Writable<ClientWriteStreamProxy> implements fs.WriteSt
 	}
 }
 
-interface ClientFsModuleProxy extends FsModuleProxy {
-	proxyId: number | Module;
+interface ClientFsModuleProxy extends FsModuleProxy, ClientServerProxy {
 	createReadStream(path: fs.PathLike, options?: any): Promise<ClientReadStreamProxy>;
 	createWriteStream(path: fs.PathLike, options?: any): Promise<ClientWriteStreamProxy>;
 	watch(filename: fs.PathLike, options?: IEncodingOptions): Promise<ClientWatcherProxy>;

packages/protocol/src/browser/modules/net.ts

@@ -1,14 +1,12 @@
 import * as net from "net";
 import { callbackify } from "util";
-import { ClientProxy, Module } from "../../common/proxy";
+import { ClientProxy, ClientServerProxy } from "../../common/proxy";
 import { NetModuleProxy, NetServerProxy, NetSocketProxy } from "../../node/modules/net";
 import { Duplex } from "./stream";
 
 // tslint:disable completed-docs
 
-interface ClientNetSocketProxy extends NetSocketProxy {
-	proxyId: number | Module;
-}
+interface ClientNetSocketProxy extends NetSocketProxy, ClientServerProxy {}
 
 export class Socket extends Duplex<ClientNetSocketProxy> implements net.Socket {
 	private _connecting: boolean = false;
@@ -130,8 +128,7 @@ export class Socket extends Duplex<ClientNetSocketProxy> implements net.Socket {
 	}
 }
 
-interface ClientNetServerProxy extends NetServerProxy {
-	proxyId: number | Module;
+interface ClientNetServerProxy extends NetServerProxy, ClientServerProxy {
 	onConnection(cb: (proxy: ClientNetSocketProxy) => void): Promise<void>;
 }
 
@@ -217,8 +214,7 @@ export class Server extends ClientProxy<ClientNetServerProxy> implements net.Ser
 
 type NodeNet = typeof net;
 
-interface ClientNetModuleProxy extends NetModuleProxy {
-	proxyId: number | Module;
+interface ClientNetModuleProxy extends NetModuleProxy, ClientServerProxy {
 	createSocket(options?: net.SocketConstructorOpts): Promise<ClientNetSocketProxy>;
 	createConnection(target: string | number | net.NetConnectOpts, host?: string): Promise<ClientNetSocketProxy>;
 	createServer(options?: { allowHalfOpen?: boolean, pauseOnConnect?: boolean }): Promise<ClientNetServerProxy>;

packages/protocol/src/browser/modules/node-pty.ts

@@ -1,12 +1,10 @@
 import * as pty from "node-pty";
-import { ClientProxy, Module } from "../../common/proxy";
+import { ClientProxy, ClientServerProxy } from "../../common/proxy";
 import { NodePtyModuleProxy, NodePtyProcessProxy } from "../../node/modules/node-pty";
 
 // tslint:disable completed-docs
 
-interface ClientNodePtyProcessProxy extends NodePtyProcessProxy {
-	proxyId: number | Module;
-}
+interface ClientNodePtyProcessProxy extends NodePtyProcessProxy, ClientServerProxy {}
 
 export class NodePtyProcess extends ClientProxy<ClientNodePtyProcessProxy> implements pty.IPty {
 	private _pid = -1;
@@ -59,8 +57,7 @@ export class NodePtyProcess extends ClientProxy<ClientNodePtyProcessProxy> imple
 
 type NodePty = typeof pty;
 
-interface ClientNodePtyModuleProxy extends NodePtyModuleProxy {
-	proxyId: number | Module;
+interface ClientNodePtyModuleProxy extends NodePtyModuleProxy, ClientServerProxy {
 	spawn(file: string, args: string[] | string, options: pty.IPtyForkOptions): Promise<ClientNodePtyProcessProxy>;
 }
 

packages/protocol/src/browser/modules/spdlog.ts

@@ -1,12 +1,10 @@
 import * as spdlog from "spdlog";
-import { ClientProxy, Module } from "../../common/proxy";
+import { ClientProxy, ClientServerProxy } from "../../common/proxy";
 import { RotatingLoggerProxy, SpdlogModuleProxy } from "../../node/modules/spdlog";
 
 // tslint:disable completed-docs
 
-interface ClientRotatingLoggerProxy extends RotatingLoggerProxy {
-	proxyId: number | Module;
-}
+interface ClientRotatingLoggerProxy extends RotatingLoggerProxy, ClientServerProxy {}
 
 class RotatingLogger extends ClientProxy<ClientRotatingLoggerProxy> implements spdlog.RotatingLogger {
 	public constructor(
@@ -35,8 +33,7 @@ class RotatingLogger extends ClientProxy<ClientRotatingLoggerProxy> implements s
 	}
 }
 
-interface ClientSpdlogModuleProxy extends SpdlogModuleProxy {
-	proxyId: number | Module;
+interface ClientSpdlogModuleProxy extends SpdlogModuleProxy, ClientServerProxy {
 	createLogger(name: string, filePath: string, fileSize: number, fileCount: number): Promise<ClientRotatingLoggerProxy>;
 }
 

packages/protocol/src/browser/modules/stream.ts

@@ -1,14 +1,12 @@
 import * as stream from "stream";
 import { callbackify } from "util";
-import { ClientProxy, Module } from "../../common/proxy";
+import { ClientProxy, ClientServerProxy } from "../../common/proxy";
 import { isPromise } from "../../common/util";
 import { DuplexProxy, ReadableProxy, WritableProxy } from "../../node/modules/stream";
 
 // tslint:disable completed-docs
 
-export interface ClientWritableProxy extends WritableProxy {
-	proxyId: number | Module;
-}
+export interface ClientWritableProxy extends WritableProxy, ClientServerProxy {}
 
 export class Writable<T extends ClientWritableProxy = ClientWritableProxy> extends ClientProxy<T> implements stream.Writable {
 	public get writable(): boolean {
@@ -93,9 +91,7 @@ export class Writable<T extends ClientWritableProxy = ClientWritableProxy> exten
 	}
 }
 
-export interface ClientReadableProxy extends ReadableProxy {
-	proxyId: number | Module;
-}
+export interface ClientReadableProxy extends ReadableProxy, ClientServerProxy {}
 
 export class Readable<T extends ClientReadableProxy = ClientReadableProxy> extends ClientProxy<T> implements stream.Readable {
 	public get readable(): boolean {
@@ -153,6 +149,9 @@ export class Readable<T extends ClientReadableProxy = ClientReadableProxy> exten
 	public pipe<P extends NodeJS.WritableStream>(destination: P, options?: { end?: boolean }): P {
 		// tslint:disable-next-line no-any this will be a Writable instance.
 		const writableProxy = (destination as any as Writable).proxyPromise;
+		if (!writableProxy) {
+			throw new Error("can only pipe stream proxies");
+		}
 		this.catch(
 			isPromise(writableProxy)
 				? writableProxy.then((p) => this.proxy.pipe(p, options))
@@ -181,9 +180,7 @@ export class Readable<T extends ClientReadableProxy = ClientReadableProxy> exten
 	}
 }
 
-export interface ClientDuplexProxy extends DuplexProxy {
-	proxyId: number | Module;
-}
+export interface ClientDuplexProxy extends DuplexProxy, ClientServerProxy {}
 
 export class Duplex<T extends ClientDuplexProxy = ClientDuplexProxy> extends Writable<T> implements stream.Duplex, stream.Readable {
 	private readonly _readable: Readable;

packages/protocol/src/browser/modules/trash.ts

@@ -1,12 +1,10 @@
 import * as trash from "trash";
-import {  Module } from "../../common/proxy";
+import { ClientServerProxy } from "../../common/proxy";
 import { TrashModuleProxy } from "../../node/modules/trash";
 
 // tslint:disable completed-docs
 
-interface ClientTrashModuleProxy extends TrashModuleProxy {
-	proxyId: number | Module;
-}
+interface ClientTrashModuleProxy extends TrashModuleProxy, ClientServerProxy {}
 
 export class TrashModule {
 	public constructor(private readonly proxy: ClientTrashModuleProxy) {}

packages/protocol/src/common/proxy.ts

@@ -116,10 +116,14 @@ export abstract class ClientProxy<T extends ClientServerProxy> extends EventEmit
 
 /**
  * Proxy to the actual instance on the server. Every method must only accept
- * serializable arguments and must return promises with serializable values. If
- * a proxy itself has proxies on creation (like how ChildProcess has stdin),
+ * serializable arguments and must return promises with serializable values.
+ *
+ * If a proxy itself has proxies on creation (like how ChildProcess has stdin),
  * then it should return all of those at once, otherwise you will miss events
  * from those child proxies and fail to dispose them properly.
+ *
+ * Events listeners are added client-side (since all events automatically
+ * forward to the client), so onDone and onEvent do not need to be asynchronous.
  */
 export interface ServerProxy {
 	/**
@@ -131,17 +135,20 @@ export interface ServerProxy {
 	 * This is used instead of an event to force it to be implemented since there
 	 * would be no guarantee the implementation would remember to emit the event.
 	 */
-	onDone(cb: () => void): Promise<void>;
+	onDone(cb: () => void): void;
 
 	/**
 	 * Listen to all possible events. On the client, this is to reduce boilerplate
 	 * that would just be a bunch of error-prone forwarding of each individual
-	 * event from the proxy to its own emitter. It also fixes a timing issue
-	 * because we just always send all events from the server, so we never miss
-	 * any due to listening too late.
+	 * event from the proxy to its own emitter.
+	 *
+	 * It also fixes a timing issue because we just always send all events from
+	 * the server, so we never miss any due to listening too late.
+	 *
+	 * This cannot be async because then we can bind to the events too late.
 	 */
 	// tslint:disable-next-line no-any
-	onEvent(cb: (event: string, ...args: any[]) => void): Promise<void>;
+	onEvent(cb: (event: string, ...args: any[]) => void): void;
 }
 
 /**

packages/protocol/src/node/modules/child_process.ts

@@ -43,7 +43,7 @@ export class ChildProcessProxy implements ServerProxy {
 		return this.process.pid;
 	}
 
-	public async onDone(cb: () => void): Promise<void> {
+	public onDone(cb: () => void): void {
 		this.process.on("close", cb);
 	}
 
@@ -53,7 +53,7 @@ export class ChildProcessProxy implements ServerProxy {
 	}
 
 	// tslint:disable-next-line no-any
-	public async onEvent(cb: (event: string, ...args: any[]) => void): Promise<void> {
+	public onEvent(cb: (event: string, ...args: any[]) => void): void {
 		this.process.on("close", (code, signal) => cb("close", code, signal));
 		this.process.on("disconnect", () => cb("disconnect"));
 		this.process.on("error", (error) => cb("error", error));

packages/protocol/src/node/modules/fs.ts

@@ -48,8 +48,8 @@ export class ReadStreamProxy extends ReadableProxy<fs.ReadStream> {
 	}
 
 	// tslint:disable-next-line no-any
-	public async onEvent(cb: (event: string, ...args: any[]) => void): Promise<void> {
-		await super.onEvent(cb);
+	public onEvent(cb: (event: string, ...args: any[]) => void): void {
+		super.onEvent(cb);
 		this.stream.on("open", (fd) => cb("open", fd));
 	}
 }
@@ -65,8 +65,8 @@ export class WriteStreamProxy extends WritableProxy<fs.WriteStream> {
 	}
 
 	// tslint:disable-next-line no-any
-	public async onEvent(cb: (event: string, ...args: any[]) => void): Promise<void> {
-		await super.onEvent(cb);
+	public onEvent(cb: (event: string, ...args: any[]) => void): void {
+		super.onEvent(cb);
 		this.stream.on("open", (fd) => cb("open", fd));
 	}
 }
@@ -83,13 +83,13 @@ export class WatcherProxy implements ServerProxy {
 		this.watcher.removeAllListeners();
 	}
 
-	public async onDone(cb: () => void): Promise<void> {
+	public onDone(cb: () => void): void {
 		this.watcher.on("close", cb);
 		this.watcher.on("error", cb);
 	}
 
 	// tslint:disable-next-line no-any
-	public async onEvent(cb: (event: string, ...args: any[]) => void): Promise<void> {
+	public onEvent(cb: (event: string, ...args: any[]) => void): void {
 		this.watcher.on("change", (event, filename) => cb("change", event, filename));
 		this.watcher.on("close", () => cb("close"));
 		this.watcher.on("error", (error) => cb("error", error));

packages/protocol/src/node/modules/net.ts

@@ -24,13 +24,13 @@ export class NetSocketProxy extends DuplexProxy<net.Socket> {
 		this.stream.unref();
 	}
 
-	public async onDone(cb: () => void): Promise<void> {
+	public onDone(cb: () => void): void {
 		this.stream.on("close", cb);
 	}
 
 	// tslint:disable-next-line no-any
-	public async onEvent(cb: (event: string, ...args: any[]) => void): Promise<void> {
-		await super.onEvent(cb);
+	public onEvent(cb: (event: string, ...args: any[]) => void): void {
+		super.onEvent(cb);
 		this.stream.on("connect", () => cb("connect"));
 		this.stream.on("lookup", (error, address, family, host) => cb("lookup", error, address, family, host));
 		this.stream.on("timeout", () => cb("timeout"));
@@ -65,12 +65,12 @@ export class NetServerProxy implements ServerProxy {
 		this.server.removeAllListeners();
 	}
 
-	public async onDone(cb: () => void): Promise<void> {
+	public onDone(cb: () => void): void {
 		this.server.on("close", cb);
 	}
 
 	// tslint:disable-next-line no-any
-	public async onEvent(cb: (event: string, ...args: any[]) => void): Promise<void> {
+	public onEvent(cb: (event: string, ...args: any[]) => void): void {
 		this.server.on("close", () => cb("close"));
 		this.server.on("error", (error) => cb("error", error));
 		this.server.on("listening", () => cb("listening"));

packages/protocol/src/node/modules/node-pty.ts

@@ -47,7 +47,7 @@ export class NodePtyProcessProxy implements ServerProxy {
 		this.process.write(data);
 	}
 
-	public async onDone(cb: () => void): Promise<void> {
+	public onDone(cb: () => void): void {
 		this.process.on("exit", cb);
 	}
 
@@ -58,7 +58,7 @@ export class NodePtyProcessProxy implements ServerProxy {
 	}
 
 	// tslint:disable-next-line no-any
-	public async onEvent(cb: (event: string, ...args: any[]) => void): Promise<void> {
+	public onEvent(cb: (event: string, ...args: any[]) => void): void {
 		this.emitter.on("process", (process) => cb("process", process));
 		this.process.on("data", (data) => cb("data", data));
 		this.process.on("exit", (exitCode, signal) => cb("exit", exitCode, signal));

packages/protocol/src/node/modules/spdlog.ts

@@ -21,7 +21,7 @@ export class RotatingLoggerProxy implements ServerProxy {
 	public async flush (): Promise<void> { this.logger.flush(); }
 	public async drop (): Promise<void> { this.logger.drop(); }
 
-	public async onDone(cb: () => void): Promise<void> {
+	public onDone(cb: () => void): void {
 		this.emitter.on("dispose", cb);
 	}
 
@@ -32,7 +32,7 @@ export class RotatingLoggerProxy implements ServerProxy {
 	}
 
 	// tslint:disable-next-line no-any
-	public async onEvent(_cb: (event: string, ...args: any[]) => void): Promise<void> {
+	public onEvent(_cb: (event: string, ...args: any[]) => void): void {
 		// No events.
 	}
 }