ipc: multicast events

src/vs/base/common/arrays.ts

@@ -584,3 +584,7 @@ export function mapArrayOrNot<T, U>(items: T | T[], fn: (_: T) => U): U | U[] {
 export function asArray<T>(x: T | T[]): T[] {
 	return Array.isArray(x) ? x : [x];
 }
+
+export function getRandomElement<T>(arr: T[]): T {
+	return arr[Math.floor(Math.random() * arr.length)];
+}

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

@@ -9,6 +9,7 @@ import { CancelablePromise, createCancelablePromise, timeout } from 'vs/base/com
 import { CancellationToken, CancellationTokenSource } from 'vs/base/common/cancellation';
 import * as errors from 'vs/base/common/errors';
 import { VSBuffer } from 'vs/base/common/buffer';
+import { getRandomElement } from 'vs/base/common/arrays';
 
 /**
  * An `IChannel` is an abstraction over a collection of commands.
@@ -117,8 +118,7 @@ export interface IClientRouter<TContext = string> {
  * order to pick the right one.
  */
 export interface IRoutingChannelClient<TContext = string> {
-	getChannel<T extends IChannel>(channelName: string, router: IClientRouter<TContext>): T;
-	getBroadcastChannel<T extends IChannel>(channelName: string): T;
+	getChannel<T extends IChannel>(channelName: string, router?: IClientRouter<TContext>): T;
 }
 
 interface IReader {
@@ -700,18 +700,26 @@ export class IPCServer<TContext = string> implements IChannelServer<TContext>, I
 		});
 	}
 
-	getChannel<T extends IChannel>(channelName: string, router: IClientRouter<TContext>): T {
+	getChannel<T extends IChannel>(channelName: string, router?: IClientRouter<TContext>): T {
 		const that = this;
 
 		return {
 			call(command: string, arg?: any, cancellationToken?: CancellationToken): Promise<T> {
-				const channelPromise = router.routeCall(that, command, arg)
+				const connectionPromise = router
+					? router.routeCall(that, command, arg)
+					: Promise.resolve(getRandomElement(that.connections));
+
+				const channelPromise = connectionPromise
 					.then(connection => (connection as Connection<TContext>).channelClient.getChannel(channelName));
 
 				return getDelayedChannel(channelPromise)
 					.call(command, arg, cancellationToken);
 			},
 			listen(event: string, arg: any): Event<T> {
+				if (!router) {
+					return that.getMulticastEvent(channelName, event, arg);
+				}
+
 				const channelPromise = router.routeEvent(that, event, arg)
 					.then(connection => (connection as Connection<TContext>).channelClient.getChannel(channelName));
 
@@ -721,64 +729,56 @@ export class IPCServer<TContext = string> implements IChannelServer<TContext>, I
 		} as T;
 	}
 
-	getBroadcastChannel<T extends IChannel>(channelName: string): T {
+	private getMulticastEvent<T extends IChannel>(channelName: string, eventName: string, arg: any): Event<T> {
 		const that = this;
+		let disposables = new DisposableStore();
 
-		return {
-			call(_): Promise<T> {
-				throw new Error('IPC broadcast channels are not supported for calls');
-			},
-			listen(eventName: string, arg: any): Event<T> {
-				let disposables = new DisposableStore();
-
-				// Create an emitter which hooks up to all clients
-				// as soon as first listener is added. It also
-				// disconnects from all clients as soon as the last listener
-				// is removed.
-				const emitter = new Emitter<T>({
-					onFirstListenerAdd: () => {
-						disposables = new DisposableStore();
-
-						// The event multiplexer is useful since the active
-						// client list is dynamic. We need to hook up and disconnection
-						// to/from clients as they come and go.
-						const eventMultiplexer = new EventMultiplexer<T>();
-						const map = new Map<Connection<TContext>, IDisposable>();
-
-						const onDidAddConnection = (connection: Connection<TContext>) => {
-							const channel = connection.channelClient.getChannel(channelName);
-							const event = channel.listen<T>(eventName, arg);
-							const disposable = eventMultiplexer.add(event);
-
-							map.set(connection, disposable);
-						};
-
-						const onDidRemoveConnection = (connection: Connection<TContext>) => {
-							const disposable = map.get(connection);
-
-							if (!disposable) {
-								return;
-							}
-
-							disposable.dispose();
-							map.delete(connection);
-						};
-
-						that.connections.forEach(onDidAddConnection);
-						that.onDidAddConnection(onDidAddConnection, undefined, disposables);
-						that.onDidRemoveConnection(onDidRemoveConnection, undefined, disposables);
-						eventMultiplexer.event(emitter.fire, emitter, disposables);
-
-						disposables.add(eventMultiplexer);
-					},
-					onLastListenerRemove: () => {
-						disposables.dispose();
+		// Create an emitter which hooks up to all clients
+		// as soon as first listener is added. It also
+		// disconnects from all clients as soon as the last listener
+		// is removed.
+		const emitter = new Emitter<T>({
+			onFirstListenerAdd: () => {
+				disposables = new DisposableStore();
+
+				// The event multiplexer is useful since the active
+				// client list is dynamic. We need to hook up and disconnection
+				// to/from clients as they come and go.
+				const eventMultiplexer = new EventMultiplexer<T>();
+				const map = new Map<Connection<TContext>, IDisposable>();
+
+				const onDidAddConnection = (connection: Connection<TContext>) => {
+					const channel = connection.channelClient.getChannel(channelName);
+					const event = channel.listen<T>(eventName, arg);
+					const disposable = eventMultiplexer.add(event);
+
+					map.set(connection, disposable);
+				};
+
+				const onDidRemoveConnection = (connection: Connection<TContext>) => {
+					const disposable = map.get(connection);
+
+					if (!disposable) {
+						return;
 					}
-				});
 
-				return emitter.event;
+					disposable.dispose();
+					map.delete(connection);
+				};
+
+				that.connections.forEach(onDidAddConnection);
+				that.onDidAddConnection(onDidAddConnection, undefined, disposables);
+				that.onDidRemoveConnection(onDidRemoveConnection, undefined, disposables);
+				eventMultiplexer.event(emitter.fire, emitter, disposables);
+
+				disposables.add(eventMultiplexer);
+			},
+			onLastListenerRemove: () => {
+				disposables.dispose();
 			}
-		} as T;
+		});
+
+		return emitter.event;
 	}
 
 	registerChannel(channelName: string, channel: IServerChannel<TContext>): void {

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

@@ -454,9 +454,9 @@ suite('Base IPC', function () {
 			client1.registerChannel('channel', clientChannel1);
 
 			const pings: string[] = [];
-			const broadcastChannel = server.getBroadcastChannel('channel');
-			const broadcastService = new TestChannelClient(broadcastChannel);
-			broadcastService.onPong(msg => pings.push(msg));
+			const channel = server.getChannel('channel');
+			const service = new TestChannelClient(channel);
+			service.onPong(msg => pings.push(msg));
 
 			await timeout(1);
 			clientService1.ping('hello 1');