ipc: use Event for onMessage

src/vs/base/parts/ipc/common/ipc.electron.ts

@@ -5,7 +5,7 @@
 
 import { IDisposable, dispose } from 'vs/base/common/lifecycle';
 import { IMessagePassingProtocol } from 'vs/base/parts/ipc/common/ipc';
-import Event, { chain } from 'vs/base/common/event';
+import Event, { Emitter, chain } from 'vs/base/common/event';
 import { fromEventEmitter } from 'vs/base/node/event';
 import { ChannelServer as IPCServer, ChannelClient as IPCClient, IChannelServer, IChannelClient, IChannel } from 'vs/base/parts/ipc/common/ipc';
 
@@ -23,14 +23,17 @@ class Protocol implements IMessagePassingProtocol {
 
 	private listener: IDisposable;
 
-	constructor(private sender: Sender, private onMessageEvent: Event<any>) {}
+	private _onMessage: Event<any>;
+	get onMessage(): Event<any> { return this._onMessage; }
 
-	send(message: any): void {
-		this.sender.send(Message, message);
+	constructor(private sender: Sender, private onMessageEvent: Event<any>) {
+		const emitter = new Emitter<any>();
+		onMessageEvent(msg => emitter.fire(msg));
+		this._onMessage = emitter.event;
 	}
 
-	onMessage(callback: (message: any) => void): void {
-		this.listener = this.onMessageEvent(callback);
+	send(message: any): void {
+		this.sender.send(Message, message);
 	}
 
 	dispose(): void {

src/vs/base/parts/ipc/common/ipc.ts

@@ -50,7 +50,7 @@ interface IHandler {
 
 export interface IMessagePassingProtocol {
 	send(request: any): void;
-	onMessage(callback: (response: any) => void): void;
+	onMessage: Event<any>;
 }
 
 enum State {
@@ -82,11 +82,12 @@ export class ChannelServer {
 
 	private channels: { [name: string]: IChannel };
 	private activeRequests: { [id: number]: IDisposable; };
+	private protocolListener: IDisposable;
 
 	constructor(private protocol: IMessagePassingProtocol) {
 		this.channels = Object.create(null);
 		this.activeRequests = Object.create(null);
-		this.protocol.onMessage(r => this.onMessage(r));
+		this.protocolListener = this.protocol.onMessage(r => this.onMessage(r));
 		this.protocol.send(<IRawResponse> { type: MessageType.ResponseInitialize });
 	}
 
@@ -150,6 +151,9 @@ export class ChannelServer {
 	}
 
 	public dispose(): void {
+		this.protocolListener.dispose();
+		this.protocolListener = null;
+
 		Object.keys(this.activeRequests).forEach(id => {
 			this.activeRequests[<any>id].dispose();
 		});
@@ -165,6 +169,7 @@ export class ChannelClient implements IChannelClient, IDisposable {
 	private bufferedRequests: IRequest[];
 	private handlers: { [id: number]: IHandler; };
 	private lastRequestId: number;
+	private protocolListener: IDisposable;
 
 	constructor(private protocol: IMessagePassingProtocol) {
 		this.state = State.Uninitialized;
@@ -172,7 +177,7 @@ export class ChannelClient implements IChannelClient, IDisposable {
 		this.bufferedRequests = [];
 		this.handlers = Object.create(null);
 		this.lastRequestId = 0;
-		this.protocol.onMessage(r => this.onMessage(r));
+		this.protocolListener = this.protocol.onMessage(r => this.onMessage(r));
 	}
 
 	getChannel<T extends IChannel>(channelName: string): T {
@@ -298,6 +303,9 @@ export class ChannelClient implements IChannelClient, IDisposable {
 	}
 
 	dispose(): void {
+		this.protocolListener.dispose();
+		this.protocolListener = null;
+
 		this.activeRequests.forEach(r => r.cancel());
 		this.activeRequests = [];
 	}

src/vs/base/parts/ipc/node/ipc.cp.ts

@@ -8,13 +8,15 @@ import { IDisposable } from 'vs/base/common/lifecycle';
 import { Promise} from 'vs/base/common/winjs.base';
 import { Delayer } from 'vs/base/common/async';
 import { clone, assign } from 'vs/base/common/objects';
+import { Emitter } from 'vs/base/common/event';
+import { fromEventEmitter } from 'vs/base/node/event';
 import { ChannelServer as IPCServer, ChannelClient as IPCClient, IChannelClient, IChannel } from 'vs/base/parts/ipc/common/ipc';
 
 export class Server extends IPCServer {
 	constructor() {
 		super({
 			send: r => { try { process.send(r); } catch (e) { /* not much to do */ } },
-			onMessage: cb => process.on('message', cb)
+			onMessage: fromEventEmitter(process, 'message', msg => msg)
 		});
 
 		process.once('disconnect', () => this.dispose());
@@ -121,32 +123,37 @@ export class Client implements IChannelClient, IDisposable {
 			}
 
 			this.child = fork(this.modulePath, args, forkOpts);
-			this._client = new IPCClient({
-				send: r => this.child && this.child.connected && this.child.send(r),
-				onMessage: cb => {
-					this.child.on('message', (msg) => {
-
-						// Handle console logs specially
-						if (msg && msg.type === '__$console') {
-							let args = ['%c[IPC Library: ' + this.options.serverName + ']', 'color: darkgreen'];
-							try {
-								const parsed = JSON.parse(msg.arguments);
-								args = args.concat(Object.getOwnPropertyNames(parsed).map(o => parsed[o]));
-							} catch (error) {
-								args.push(msg.arguments);
-							}
-
-							console[msg.severity].apply(console, args);
-						}
-
-						// Anything else goes to the outside
-						else {
-							cb(msg);
-						}
-					});
+
+			const onMessageEmitter = new Emitter<any>();
+			const onRawMessage = fromEventEmitter(this.child, 'message', msg => msg);
+
+			onRawMessage(msg => {
+				// Handle console logs specially
+				if (msg && msg.type === '__$console') {
+					let args = ['%c[IPC Library: ' + this.options.serverName + ']', 'color: darkgreen'];
+					try {
+						const parsed = JSON.parse(msg.arguments);
+						args = args.concat(Object.getOwnPropertyNames(parsed).map(o => parsed[o]));
+					} catch (error) {
+						args.push(msg.arguments);
+					}
+
+					console[msg.severity].apply(console, args);
+					return null;
+				}
+
+				// Anything else goes to the outside
+				else {
+					onMessageEmitter.fire(msg);
 				}
 			});
 
+			const send = r => this.child && this.child.connected && this.child.send(r);
+			const onMessage = onMessageEmitter.event;
+			const protocol = { send, onMessage };
+
+			this._client = new IPCClient(protocol);
+
 			const onExit = () => this.disposeClient();
 			process.once('exit', onExit);
 

src/vs/base/parts/ipc/node/ipc.net.ts

@@ -23,7 +23,9 @@ function bufferIndexOf(buffer: Buffer, value: number, start = 0) {
 class Protocol implements IMessagePassingProtocol {
 
 	private static Boundary = new Buffer([0]);
-	private onMessageEvent: Event<any>;
+
+	private _onMessage: Event<any>;
+	get onMessage(): Event<any> { return this._onMessage; }
 
 	constructor(private socket: Socket) {
 		let buffer = null;
@@ -58,7 +60,7 @@ class Protocol implements IMessagePassingProtocol {
 			}
 		});
 
-		this.onMessageEvent = emitter.event;
+		this._onMessage = emitter.event;
 	}
 
 	public send(message: any): void {
@@ -69,10 +71,6 @@ class Protocol implements IMessagePassingProtocol {
 			// noop
 		}
 	}
-
-	public onMessage(callback: (message: any) => void): void {
-		this.onMessageEvent(callback);
-	}
 }
 
 class RoutingChannelClient implements IMultiChannelClient {