Allow passing proxies back from client to server

This makes things like piping streams possible.

Author: code-asher

packages/protocol/src/browser/client.ts

@@ -4,7 +4,7 @@ import { promisify } from "util";
 import { Emitter } from "@coder/events";
 import { logger, field } from "@coder/logger";
 import { ReadWriteConnection, InitData, SharedProcessData } from "../common/connection";
-import { Module, ServerProxy } from "../common/proxy";
+import { ClientServerProxy, Module, ServerProxy } from "../common/proxy";
 import { argumentToProto, protoToArgument, moduleToProto, protoToModule, protoToOperatingSystem } from "../common/util";
 import { Argument, Ping, ServerMessage, ClientMessage, Method, Event, Callback } from "../proto";
 import { FsModule, ChildProcessModule, NetModule, NodePtyModule, SpdlogModule, TrashModule } from "./modules";
@@ -224,7 +224,11 @@ export class Client {
 			field("method", method),
 		]);
 
-		proxyMessage.setArgsList(args.map((a) => argumentToProto(a, storeCallback)));
+		proxyMessage.setArgsList(args.map((a) => argumentToProto<ClientServerProxy>(
+			a,
+			storeCallback,
+			(p) => p.proxyId,
+		)));
 
 		const clientMessage = new ClientMessage();
 		clientMessage.setMethod(message);
@@ -429,7 +433,7 @@ export class Client {
 	/**
 	 * Return a proxy that makes remote calls.
 	 */
-	private createProxy<T>(proxyId: number | Module, promise: Promise<any> = Promise.resolve()): T {
+	private createProxy<T extends ClientServerProxy>(proxyId: number | Module, promise: Promise<any> = Promise.resolve()): T {
 		logger.trace(() => [
 			"creating proxy",
 			field("proxyId", proxyId),
@@ -449,7 +453,7 @@ export class Client {
 					cb(event.event, ...event.args);
 				});
 			},
-		}, {
+		} as ClientServerProxy, {
 			get: (target: any, name: string): any => {
 				// When resolving a promise with a proxy, it will check for "then".
 				if (name === "then") {

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

@@ -2,13 +2,24 @@ import * as cp from "child_process";
 import * as net from "net";
 import * as stream from "stream";
 import { callbackify } from "util";
-import { ClientProxy } from "../../common/proxy";
-import { ChildProcessModuleProxy, ChildProcessProxy, ChildProcessProxies } from "../../node/modules/child_process";
-import { Readable, Writable } from "./stream";
+import { ClientProxy, Module } from "../../common/proxy";
+import { ChildProcessModuleProxy, ChildProcessProxy } from "../../node/modules/child_process";
+import { ClientWritableProxy, ClientReadableProxy, Readable, Writable } from "./stream";
 
 // tslint:disable completed-docs
 
-export class ChildProcess extends ClientProxy<ChildProcessProxy> implements cp.ChildProcess {
+export interface ClientChildProcessProxy extends ChildProcessProxy {
+	proxyId: number | Module;
+}
+
+export interface ClientChildProcessProxies {
+	childProcess: ClientChildProcessProxy;
+	stdin?: ClientWritableProxy | null;
+	stdout?: ClientReadableProxy | null;
+	stderr?: ClientReadableProxy | null;
+}
+
+export class ChildProcess extends ClientProxy<ClientChildProcessProxy> implements cp.ChildProcess {
 	public readonly stdin: stream.Writable;
 	public readonly stdout: stream.Readable;
 	public readonly stderr: stream.Readable;
@@ -18,7 +29,7 @@ export class ChildProcess extends ClientProxy<ChildProcessProxy> implements cp.C
 	private _killed: boolean = false;
 	private _pid = -1;
 
-	public constructor(proxyPromises: Promise<ChildProcessProxies>) {
+	public constructor(proxyPromises: Promise<ClientChildProcessProxies>) {
 		super(proxyPromises.then((p) => p.childProcess));
 		this.stdin = new Writable(proxyPromises.then((p) => p.stdin!));
 		this.stdout = new Readable(proxyPromises.then((p) => p.stdout!));
@@ -99,8 +110,15 @@ export class ChildProcess extends ClientProxy<ChildProcessProxy> implements cp.C
 	}
 }
 
+interface ClientChildProcessModuleProxy extends ChildProcessModuleProxy {
+	proxyId: number | Module;
+	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>;
+}
+
 export class ChildProcessModule {
-	public constructor(private readonly proxy: ChildProcessModuleProxy) {}
+	public constructor(private readonly proxy: ClientChildProcessModuleProxy) {}
 
 	public exec = (
 		command: string,

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

@@ -1,6 +1,6 @@
 import * as fs from "fs";
 import { callbackify } from "util";
-import { ClientProxy, Batch } from "../../common/proxy";
+import { Batch, ClientProxy, Module } 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,7 +38,11 @@ class ReaddirBatch extends Batch<Buffer[] | fs.Dirent[] | string[], { path: fs.P
 	}
 }
 
-class Watcher extends ClientProxy<WatcherProxy> implements fs.FSWatcher {
+interface ClientWatcherProxy extends WatcherProxy {
+	proxyId: number | Module;
+}
+
+class Watcher extends ClientProxy<ClientWatcherProxy> implements fs.FSWatcher {
 	public close(): void {
 		this.catch(this.proxy.close());
 	}
@@ -48,7 +52,11 @@ class Watcher extends ClientProxy<WatcherProxy> implements fs.FSWatcher {
 	}
 }
 
-class ReadStream extends Readable<ReadStreamProxy> implements fs.ReadStream {
+interface ClientReadStreamProxy extends ReadStreamProxy {
+	proxyId: number | Module;
+}
+
+class ReadStream extends Readable<ClientReadStreamProxy> implements fs.ReadStream {
 	public get bytesRead(): number {
 		throw new Error("not implemented");
 	}
@@ -62,7 +70,11 @@ class ReadStream extends Readable<ReadStreamProxy> implements fs.ReadStream {
 	}
 }
 
-class WriteStream extends Writable<WriteStreamProxy> implements fs.WriteStream {
+interface ClientWriteStreamProxy extends WriteStreamProxy {
+	proxyId: number | Module;
+}
+
+class WriteStream extends Writable<ClientWriteStreamProxy> implements fs.WriteStream {
 	public get bytesWritten(): number {
 		throw new Error("not implemented");
 	}
@@ -76,12 +88,19 @@ class WriteStream extends Writable<WriteStreamProxy> implements fs.WriteStream {
 	}
 }
 
+interface ClientFsModuleProxy extends FsModuleProxy {
+	proxyId: number | Module;
+	createReadStream(path: fs.PathLike, options?: any): Promise<ClientReadStreamProxy>;
+	createWriteStream(path: fs.PathLike, options?: any): Promise<ClientWriteStreamProxy>;
+	watch(filename: fs.PathLike, options?: IEncodingOptions): Promise<ClientWatcherProxy>;
+}
+
 export class FsModule {
 	private readonly statBatch: StatBatch;
 	private readonly lstatBatch: LstatBatch;
 	private readonly readdirBatch: ReaddirBatch;
 
-	public constructor(private readonly proxy: FsModuleProxy) {
+	public constructor(private readonly proxy: ClientFsModuleProxy) {
 		this.statBatch = new StatBatch(this.proxy);
 		this.lstatBatch = new LstatBatch(this.proxy);
 		this.readdirBatch = new ReaddirBatch(this.proxy);

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

@@ -1,16 +1,20 @@
 import * as net from "net";
 import { callbackify } from "util";
-import { ClientProxy } from "../../common/proxy";
+import { ClientProxy, Module } from "../../common/proxy";
 import { NetModuleProxy, NetServerProxy, NetSocketProxy } from "../../node/modules/net";
 import { Duplex } from "./stream";
 
 // tslint:disable completed-docs
 
-export class Socket extends Duplex<NetSocketProxy> implements net.Socket {
+interface ClientNetSocketProxy extends NetSocketProxy {
+	proxyId: number | Module;
+}
+
+export class Socket extends Duplex<ClientNetSocketProxy> implements net.Socket {
 	private _connecting: boolean = false;
 	private _destroyed: boolean = false;
 
-	public constructor(proxyPromise: Promise<NetSocketProxy> | NetSocketProxy, connecting?: boolean) {
+	public constructor(proxyPromise: Promise<ClientNetSocketProxy> | ClientNetSocketProxy, connecting?: boolean) {
 		super(proxyPromise);
 		if (connecting) {
 			this._connecting = connecting;
@@ -126,12 +130,17 @@ export class Socket extends Duplex<NetSocketProxy> implements net.Socket {
 	}
 }
 
-export class Server extends ClientProxy<NetServerProxy> implements net.Server {
+interface ClientNetServerProxy extends NetServerProxy {
+	proxyId: number | Module;
+	onConnection(cb: (proxy: ClientNetSocketProxy) => void): Promise<void>;
+}
+
+export class Server extends ClientProxy<ClientNetServerProxy> implements net.Server {
 	private socketId = 0;
 	private readonly sockets = new Map<number, net.Socket>();
 	private _listening: boolean = false;
 
-	public constructor(proxyPromise: Promise<NetServerProxy> | NetServerProxy) {
+	public constructor(proxyPromise: Promise<ClientNetServerProxy> | ClientNetServerProxy) {
 		super(proxyPromise);
 
 		this.catch(this.proxy.onConnection((socketProxy) => {
@@ -208,11 +217,18 @@ export class Server extends ClientProxy<NetServerProxy> implements net.Server {
 
 type NodeNet = typeof net;
 
+interface ClientNetModuleProxy extends NetModuleProxy {
+	proxyId: number | Module;
+	createSocket(options?: net.SocketConstructorOpts): Promise<ClientNetSocketProxy>;
+	createConnection(target: string | number | net.NetConnectOpts, host?: string): Promise<ClientNetSocketProxy>;
+	createServer(options?: { allowHalfOpen?: boolean, pauseOnConnect?: boolean }): Promise<ClientNetServerProxy>;
+}
+
 export class NetModule implements NodeNet {
 	public readonly Socket: typeof net.Socket;
 	public readonly Server: typeof net.Server;
 
-	public constructor(private readonly proxy: NetModuleProxy) {
+	public constructor(private readonly proxy: ClientNetModuleProxy) {
 		// @ts-ignore this is because Socket is missing things from the Stream
 		// namespace but I'm unsure how best to provide them (finished,
 		// finished.__promisify__, pipeline, and some others) or if it even matters.

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

@@ -1,15 +1,19 @@
 import * as pty from "node-pty";
-import { ClientProxy } from "../../common/proxy";
+import { ClientProxy, Module } from "../../common/proxy";
 import { NodePtyModuleProxy, NodePtyProcessProxy } from "../../node/modules/node-pty";
 
 // tslint:disable completed-docs
 
-export class NodePtyProcess extends ClientProxy<NodePtyProcessProxy> implements pty.IPty {
+interface ClientNodePtyProcessProxy extends NodePtyProcessProxy {
+	proxyId: number | Module;
+}
+
+export class NodePtyProcess extends ClientProxy<ClientNodePtyProcessProxy> implements pty.IPty {
 	private _pid = -1;
 	private _process = "";
 
 	public constructor(
-		private readonly moduleProxy: NodePtyModuleProxy,
+		private readonly moduleProxy: ClientNodePtyModuleProxy,
 		private readonly file: string,
 		private readonly args: string[] | string,
 		private readonly options: pty.IPtyForkOptions,
@@ -18,10 +22,12 @@ export class NodePtyProcess extends ClientProxy<NodePtyProcessProxy> implements
 		this.on("process", (process) => this._process = process);
 	}
 
-	protected initialize(proxyPromise: Promise<NodePtyProcessProxy>): void {
-		super.initialize(proxyPromise);
+	protected initialize(proxyPromise: Promise<ClientNodePtyProcessProxy>): ClientNodePtyProcessProxy {
+		const proxy = super.initialize(proxyPromise);
 		this.catch(this.proxy.getPid().then((p) => this._pid = p));
 		this.catch(this.proxy.getProcess().then((p) => this._process = p));
+
+		return proxy;
 	}
 
 	public get pid(): number {
@@ -53,8 +59,13 @@ export class NodePtyProcess extends ClientProxy<NodePtyProcessProxy> implements
 
 type NodePty = typeof pty;
 
+interface ClientNodePtyModuleProxy extends NodePtyModuleProxy {
+	proxyId: number | Module;
+	spawn(file: string, args: string[] | string, options: pty.IPtyForkOptions): Promise<ClientNodePtyProcessProxy>;
+}
+
 export class NodePtyModule implements NodePty {
-	public constructor(private readonly proxy: NodePtyModuleProxy) {}
+	public constructor(private readonly proxy: ClientNodePtyModuleProxy) {}
 
 	public spawn = (file: string, args: string[] | string, options: pty.IPtyForkOptions): pty.IPty => {
 		return new NodePtyProcess(this.proxy, file, args, options);

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

@@ -1,12 +1,16 @@
 import * as spdlog from "spdlog";
-import { ClientProxy } from "../../common/proxy";
+import { ClientProxy, Module } from "../../common/proxy";
 import { RotatingLoggerProxy, SpdlogModuleProxy } from "../../node/modules/spdlog";
 
 // tslint:disable completed-docs
 
-class RotatingLogger extends ClientProxy<RotatingLoggerProxy> implements spdlog.RotatingLogger {
+interface ClientRotatingLoggerProxy extends RotatingLoggerProxy {
+	proxyId: number | Module;
+}
+
+class RotatingLogger extends ClientProxy<ClientRotatingLoggerProxy> implements spdlog.RotatingLogger {
 	public constructor(
-		private readonly moduleProxy: SpdlogModuleProxy,
+		private readonly moduleProxy: ClientSpdlogModuleProxy,
 		private readonly name: string,
 		private readonly filename: string,
 		private readonly filesize: number,
@@ -31,10 +35,15 @@ class RotatingLogger extends ClientProxy<RotatingLoggerProxy> implements spdlog.
 	}
 }
 
+interface ClientSpdlogModuleProxy extends SpdlogModuleProxy {
+	proxyId: number | Module;
+	createLogger(name: string, filePath: string, fileSize: number, fileCount: number): Promise<ClientRotatingLoggerProxy>;
+}
+
 export class SpdlogModule {
 	public readonly RotatingLogger: typeof spdlog.RotatingLogger;
 
-	public constructor(private readonly proxy: SpdlogModuleProxy) {
+	public constructor(private readonly proxy: ClientSpdlogModuleProxy) {
 		this.RotatingLogger = class extends RotatingLogger {
 			public constructor(name: string, filename: string, filesize: number, filecount: number) {
 				super(proxy, name, filename, filesize, filecount);

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

@@ -1,11 +1,16 @@
 import * as stream from "stream";
 import { callbackify } from "util";
-import { ClientProxy } from "../../common/proxy";
-import { DuplexProxy, IReadableProxy, WritableProxy } from "../../node/modules/stream";
+import { ClientProxy, Module } from "../../common/proxy";
+import { isPromise } from "../../common/util";
+import { DuplexProxy, ReadableProxy, WritableProxy } from "../../node/modules/stream";
 
 // tslint:disable completed-docs
 
-export class Writable<T extends WritableProxy = WritableProxy> extends ClientProxy<T> implements stream.Writable {
+export interface ClientWritableProxy extends WritableProxy {
+	proxyId: number | Module;
+}
+
+export class Writable<T extends ClientWritableProxy = ClientWritableProxy> extends ClientProxy<T> implements stream.Writable {
 	public get writable(): boolean {
 		throw new Error("not implemented");
 	}
@@ -88,7 +93,11 @@ export class Writable<T extends WritableProxy = WritableProxy> extends ClientPro
 	}
 }
 
-export class Readable<T extends IReadableProxy = IReadableProxy> extends ClientProxy<T> implements stream.Readable {
+export interface ClientReadableProxy extends ReadableProxy {
+	proxyId: number | Module;
+}
+
+export class Readable<T extends ClientReadableProxy = ClientReadableProxy> extends ClientProxy<T> implements stream.Readable {
 	public get readable(): boolean {
 		throw new Error("not implemented");
 	}
@@ -141,8 +150,16 @@ export class Readable<T extends IReadableProxy = IReadableProxy> extends ClientP
 		throw new Error("not implemented");
 	}
 
-	public pipe<T>(): T {
-		throw new Error("not implemented");
+	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;
+		this.catch(
+			isPromise(writableProxy)
+				? writableProxy.then((p) => this.proxy.pipe(p, options))
+				: this.proxy.pipe(writableProxy, options),
+		);
+
+		return destination;
 	}
 
 	// tslint:disable-next-line no-any
@@ -164,7 +181,11 @@ export class Readable<T extends IReadableProxy = IReadableProxy> extends ClientP
 	}
 }
 
-export class Duplex<T extends DuplexProxy = DuplexProxy> extends Writable<T> implements stream.Duplex, stream.Readable {
+export interface ClientDuplexProxy extends DuplexProxy {
+	proxyId: number | Module;
+}
+
+export class Duplex<T extends ClientDuplexProxy = ClientDuplexProxy> extends Writable<T> implements stream.Duplex, stream.Readable {
 	private readonly _readable: Readable;
 
 	public constructor(proxyPromise: Promise<T> | T) {

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

@@ -1,10 +1,15 @@
 import * as trash from "trash";
+import {  Module } from "../../common/proxy";
 import { TrashModuleProxy } from "../../node/modules/trash";
 
 // tslint:disable completed-docs
 
+interface ClientTrashModuleProxy extends TrashModuleProxy {
+	proxyId: number | Module;
+}
+
 export class TrashModule {
-	public constructor(private readonly proxy: TrashModuleProxy) {}
+	public constructor(private readonly proxy: ClientTrashModuleProxy) {}
 
 	public trash = (path: string, options?: trash.Options): Promise<void> => {
 		return this.proxy.trash(path, options);