src/vs/base/parts/ipc/common/ipc.ts
1341 LOC · 1092 covered · 249 uncovered · 197 ranges · 205 concepts · 32 introducers · 113 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
>
/*---------------------------------------------------------------------------------------------
ipc.ts ×60
2
>
* Copyright (c) Microsoft Corporation. All rights reserved.
3
>
* Licensed under the MIT License. See License.txt in the project root for license information.
4
>
*--------------------------------------------------------------------------------------------*/
5
>
6
>
import { getRandomElement } from '../../../common/arrays.js';
7
>
import { CancelablePromise, createCancelablePromise, timeout } from '../../../common/async.js';
8
>
import { VSBuffer } from '../../../common/buffer.js';
9
>
import { CancellationToken, CancellationTokenSource } from '../../../common/cancellation.js';
10
>
import { memoize } from '../../../common/decorators.js';
11
>
import { CancellationError, ErrorNoTelemetry } from '../../../common/errors.js';
12
>
import { Emitter, Event, EventMultiplexer, Relay } from '../../../common/event.js';
13
>
import { createSingleCallFunction } from '../../../common/functional.js';
14
>
import { DisposableStore, dispose, IDisposable, toDisposable } from '../../../common/lifecycle.js';
15
>
import { revive } from '../../../common/marshalling.js';
16
>
import * as strings from '../../../common/strings.js';
17
>
import { isFunction, isUndefinedOrNull } from '../../../common/types.js';
18
>
19
>
/**
20
>
* An `IChannel` is an abstraction over a collection of commands.
21
>
* You can `call` several commands on a channel, each taking at
22
>
* most one single argument. A `call` always returns a promise
23
>
* with at most one single return value.
24
>
*/
25
>
export interface IChannel {
26
>
call<T>(command: string, arg?: any, cancellationToken?: CancellationToken): Promise<T>;
27
>
listen<T>(event: string, arg?: any): Event<T>;
28
>
}
29
>
30
>
/**
31
>
* An `IServerChannel` is the counter part to `IChannel`,
32
>
* on the server-side. You should implement this interface
33
>
* if you'd like to handle remote promises or events.
34
>
*/
35
>
export interface IServerChannel<TContext = string> {
36
>
call<T>(ctx: TContext, command: string, arg?: any, cancellationToken?: CancellationToken): Promise<T>;
37
>
listen<T>(ctx: TContext, event: string, arg?: any): Event<T>;
38
>
}
39
>
40
>
const enum RequestType {
41
>
Promise = 100,
42
>
PromiseCancel = 101,
43
>
EventListen = 102,
44
>
EventDispose = 103
45
>
}
46
>
47
function requestTypeToStr(type: RequestType): string {
48
switch (type) {
49
case RequestType.Promise:
50
return 'req';
51
case RequestType.PromiseCancel:
52
return 'cancel';
53
case RequestType.EventListen:
54
return 'subscribe';
55
case RequestType.EventDispose:
56
return 'unsubscribe';
57
}
58
}
60
>
type IRawPromiseRequest = { type: RequestType.Promise; id: number; channelName: string; name: string; arg: any };
61
>
type IRawPromiseCancelRequest = { type: RequestType.PromiseCancel; id: number };
62
>
type IRawEventListenRequest = { type: RequestType.EventListen; id: number; channelName: string; name: string; arg: any };
63
>
type IRawEventDisposeRequest = { type: RequestType.EventDispose; id: number };
64
>
type IRawRequest = IRawPromiseRequest | IRawPromiseCancelRequest | IRawEventListenRequest | IRawEventDisposeRequest;
65
>
66
>
const enum ResponseType {
67
>
Initialize = 200,
68
>
PromiseSuccess = 201,
69
>
PromiseError = 202,
70
>
PromiseErrorObj = 203,
71
>
EventFire = 204
72
>
}
73
>
74
function responseTypeToStr(type: ResponseType): string {
75
switch (type) {
76
case ResponseType.Initialize:
77
return `init`;
78
case ResponseType.PromiseSuccess:
79
return `reply:`;
80
case ResponseType.PromiseError:
81
case ResponseType.PromiseErrorObj:
82
return `replyErr:`;
83
case ResponseType.EventFire:
84
return `event:`;
85
}
86
}
88
>
type IRawInitializeResponse = { type: ResponseType.Initialize };
89
>
type IRawPromiseSuccessResponse = { type: ResponseType.PromiseSuccess; id: number; data: any };
90
>
type IRawPromiseErrorResponse = { type: ResponseType.PromiseError; id: number; data: { message: string; name: string; stack: string[] | undefined } };
91
>
type IRawPromiseErrorObjResponse = { type: ResponseType.PromiseErrorObj; id: number; data: any };
92
>
type IRawEventFireResponse = { type: ResponseType.EventFire; id: number; data: any };
93
>
type IRawResponse = IRawInitializeResponse | IRawPromiseSuccessResponse | IRawPromiseErrorResponse | IRawPromiseErrorObjResponse | IRawEventFireResponse;
94
>
95
>
interface IHandler {
96
>
(response: IRawResponse): void;
97
>
}
98
>
99
>
export interface IMessagePassingProtocol {
100
>
send(buffer: VSBuffer): void;
101
>
readonly onMessage: Event<VSBuffer>;
102
>
/**
103
>
* Wait for the write buffer (if applicable) to become empty.
104
>
*/
105
>
drain?(): Promise<void>;
106
>
}
107
>
108
>
enum State {
109
>
Uninitialized,
110
>
Idle
111
>
}
112
>
113
>
/**
114
>
* An `IChannelServer` hosts a collection of channels. You are
115
>
* able to register channels onto it, provided a channel name.
116
>
*/
117
>
export interface IChannelServer<TContext = string> {
118
>
registerChannel(channelName: string, channel: IServerChannel<TContext>): void;
119
>
}
120
>
121
>
/**
122
>
* An `IChannelClient` has access to a collection of channels. You
123
>
* are able to get those channels, given their channel name.
124
>
*/
125
>
export interface IChannelClient {
126
>
getChannel<T extends IChannel>(channelName: string): T;
127
>
}
128
>
129
>
export interface Client<TContext> {
130
>
readonly ctx: TContext;
131
>
}
132
>
133
>
export interface IConnectionHub<TContext> {
134
>
readonly connections: Connection<TContext>[];
135
>
readonly onDidAddConnection: Event<Connection<TContext>>;
136
>
readonly onDidRemoveConnection: Event<Connection<TContext>>;
137
>
}
138
>
139
>
/**
140
>
* An `IClientRouter` is responsible for routing calls to specific
141
>
* channels, in scenarios in which there are multiple possible
142
>
* channels (each from a separate client) to pick from.
143
>
*/
144
>
export interface IClientRouter<TContext = string> {
145
>
routeCall(hub: IConnectionHub<TContext>, command: string, arg?: any, cancellationToken?: CancellationToken): Promise<Client<TContext>>;
146
>
routeEvent(hub: IConnectionHub<TContext>, event: string, arg?: any): Promise<Client<TContext>>;
147
>
}
148
>
149
>
/**
150
>
* Similar to the `IChannelClient`, you can get channels from this
151
>
* collection of channels. The difference being that in the
152
>
* `IRoutingChannelClient`, there are multiple clients providing
153
>
* the same channel. You'll need to pass in an `IClientRouter` in
154
>
* order to pick the right one.
155
>
*/
156
>
export interface IRoutingChannelClient<TContext = string> {
157
>
getChannel<T extends IChannel>(channelName: string, router?: IClientRouter<TContext>): T;
158
>
}
159
>
160
>
interface IReader {
161
>
read(bytes: number): VSBuffer;
162
>
}
163
>
164
>
interface IWriter {
165
>
write(buffer: VSBuffer): void;
166
>
}
167
>
168
>
169
>
/**
170
>
* @see https://en.wikipedia.org/wiki/Variable-length_quantity
171
>
*/
173
>
let value = 0;
174
>
for (let n = 0; ; n += 7) {
175
>
const next = reader.read(1);
176
>
value |= (next.buffer[0] & 0b01111111) << n;
177
>
if (!(next.buffer[0] & 0b10000000)) {
178
>
return value;
179
>
}
180
>
}
181
}
183
>
const vqlZero = createOneByteBuffer(0);
184
>
185
>
/**
186
>
* @see https://en.wikipedia.org/wiki/Variable-length_quantity
187
>
*/
189
>
if (value === 0) {
191
>
return;
192
>
}
194
>
let len = 0;
195
>
for (let v2 = value; v2 !== 0; v2 = v2 >>> 7) {
196
>
len++;
197
>
}
198
>
199
>
const scratch = VSBuffer.alloc(len);
200
>
for (let i = 0; value !== 0; i++) {
201
>
scratch.buffer[i] = value & 0b01111111;
202
>
value = value >>> 7;
203
>
if (value > 0) {
204
>
scratch.buffer[i] |= 0b10000000;
205
>
}
206
>
}
207
>
208
>
writer.write(scratch);
209
>
}
211
>
export class BufferReader implements IReader {
212
>
213
>
private pos = 0;
214
>
215
>
constructor(private buffer: VSBuffer) { }
216
>
217
>
read(bytes: number): VSBuffer {
219
>
this.pos += result.byteLength;
220
>
return result;
221
>
}
223
>
224
>
export class BufferWriter implements IWriter, IDisposable {
226
>
private buffers: VSBuffer[] = [];
228
>
get buffer(): VSBuffer {
230
>
}
232
>
write(buffer: VSBuffer): void {
234
>
}
236
>
dispose(): void {
238
>
this.buffers.length = 0;
239
>
}
241
>
242
>
enum DataType {
243
>
Undefined = 0,
244
>
String = 1,
245
>
Buffer = 2,
246
>
VSBuffer = 3,
247
>
Array = 4,
248
>
Object = 5,
249
>
Int = 6
250
>
}
251
>
252
>
function createOneByteBuffer(value: number): VSBuffer {
253
>
const result = VSBuffer.alloc(1);
254
>
result.writeUInt8(value, 0);
255
>
return result;
256
>
}
257
>
258
>
const BufferPresets = {
259
>
Undefined: createOneByteBuffer(DataType.Undefined),
260
>
String: createOneByteBuffer(DataType.String),
261
>
Buffer: createOneByteBuffer(DataType.Buffer),
262
>
VSBuffer: createOneByteBuffer(DataType.VSBuffer),
263
>
Array: createOneByteBuffer(DataType.Array),
264
>
Object: createOneByteBuffer(DataType.Object),
265
>
Uint: createOneByteBuffer(DataType.Int),
266
>
};
267
>
268
>
export function serialize(writer: IWriter, data: any): void {
270
>
writer.write(BufferPresets.Undefined);
271
>
} else if (typeof data === 'string') {
272
>
const buffer = VSBuffer.fromString(data);
273
>
writer.write(BufferPresets.String);
274
>
writeInt32VQL(writer, buffer.byteLength);
275
>
writer.write(buffer);
276
>
} else if (VSBuffer.isNativeBuffer(data)) {
277
const buffer = VSBuffer.wrap(data);
278
writer.write(BufferPresets.Buffer);
279
writeInt32VQL(writer, buffer.byteLength);
280
writer.write(buffer);
283
>
writeInt32VQL(writer, data.byteLength);
284
>
writer.write(data);
286
>
writer.write(BufferPresets.Array);
287
>
writeInt32VQL(writer, data.length);
288
>
289
>
for (const el of data) {
290
>
serialize(writer, el);
291
>
}
292
>
} else if (typeof data === 'number' && (data | 0) === data) {
293
>
// write a vql if it's a number that we can do bitwise operations on
294
>
writer.write(BufferPresets.Uint);
295
>
writeInt32VQL(writer, data);
296
>
} else {
298
>
writer.write(BufferPresets.Object);
299
>
writeInt32VQL(writer, buffer.byteLength);
300
>
writer.write(buffer);
301
>
}
304
>
export function deserialize(reader: IReader): any {
306
>
307
>
switch (type) {
308
>
case DataType.Undefined: return undefined;
309
>
case DataType.String: return reader.read(readIntVQL(reader)).toString();
310
>
case DataType.Buffer: return reader.read(readIntVQL(reader)).buffer;
311
>
case DataType.VSBuffer: return reader.read(readIntVQL(reader));
312
>
case DataType.Array: {
313
>
const length = readIntVQL(reader);
314
>
const result: any[] = [];
315
>
316
>
for (let i = 0; i < length; i++) {
317
>
result.push(deserialize(reader));
318
>
}
319
>
320
>
return result;
321
>
}
322
>
case DataType.Object: return JSON.parse(reader.read(readIntVQL(reader)).toString());
323
>
case DataType.Int: return readIntVQL(reader);
324
>
}
325
>
}
327
>
interface PendingRequest {
328
>
request: IRawPromiseRequest | IRawEventListenRequest;
329
>
timeoutTimer: Timeout;
330
>
}
331
>
332
>
export class ChannelServer<TContext = string> implements IChannelServer<TContext>, IDisposable {
333
>
334
>
private channels = new Map<string, IServerChannel<TContext>>();
335
>
private activeRequests = new Map<number, IDisposable>();
336
>
private protocolListener: IDisposable | null;
337
>
338
>
// Requests might come in for channels which are not yet registered.
339
>
// They will timeout after `timeoutDelay`.
340
>
private pendingRequests = new Map<string, PendingRequest[]>();
341
>
342
>
constructor(private protocol: IMessagePassingProtocol, private ctx: TContext, private logger: IIPCLogger | null = null, private timeoutDelay = 1000) {
344
>
this.sendResponse({ type: ResponseType.Initialize });
345
>
}
347
>
registerChannel(channelName: string, channel: IServerChannel<TContext>): void {
349
>
350
>
// https://github.com/microsoft/vscode/issues/72531
351
>
setTimeout(() => this.flushPendingRequests(channelName), 0);
352
>
}
354
>
private sendResponse(response: IRawResponse): void {
356
>
case ResponseType.Initialize: {
357
>
const msgLength = this.send([response.type]);
358
>
this.logger?.logOutgoing(msgLength, 0, RequestInitiator.OtherSide, responseTypeToStr(response.type));
359
>
return;
360
>
}
361
>
362
>
case ResponseType.PromiseSuccess:
363
>
case ResponseType.PromiseError:
364
>
case ResponseType.EventFire:
365
>
case ResponseType.PromiseErrorObj: {
367
>
this.logger?.logOutgoing(msgLength, response.id, RequestInitiator.OtherSide, responseTypeToStr(response.type), response.data);
368
>
return;
369
>
}
371
>
}
373
>
private send(header: unknown, body: any = undefined): number {
375
>
try {
376
>
serialize(writer, header);
377
>
serialize(writer, body);
378
>
return this.sendBuffer(writer.buffer);
379
>
} finally {
380
>
writer.dispose();
381
>
}
382
>
}
384
>
private sendBuffer(message: VSBuffer): number {
386
>
this.protocol.send(message);
387
>
return message.byteLength;
388
>
} catch (err) {
389
// noop
390
return 0;
391
}
394
>
private onRawMessage(message: VSBuffer): void {
396
>
const header = deserialize(reader);
397
>
const body = deserialize(reader);
398
>
const type = header[0] as RequestType;
399
>
400
>
switch (type) {
401
>
case RequestType.Promise:
402
>
this.logger?.logIncoming(message.byteLength, header[1], RequestInitiator.OtherSide, `${requestTypeToStr(type)}: ${header[2]}.${header[3]}`, body);
ipc.ts ×11
403
>
return this.onPromise({ type, id: header[1], channelName: header[2], name: header[3], arg: body });
405
>
this.logger?.logIncoming(message.byteLength, header[1], RequestInitiator.OtherSide, `${requestTypeToStr(type)}: ${header[2]}.${header[3]}`, body);
ipc.ts ×8
406
>
return this.onEventListen({ type, id: header[1], channelName: header[2], name: header[3], arg: body });
408
>
this.logger?.logIncoming(message.byteLength, header[1], RequestInitiator.OtherSide, `${requestTypeToStr(type)}`);
ipc.ts ×3
409
>
return this.disposeActiveRequest({ type, id: header[1] });
411
>
this.logger?.logIncoming(message.byteLength, header[1], RequestInitiator.OtherSide, `${requestTypeToStr(type)}`);
ipc.ts ×1
412
>
return this.disposeActiveRequest({ type, id: header[1] });
414
>
}
416
>
private onPromise(request: IRawPromiseRequest): void {
418
>
419
>
if (!channel) {
420
this.collectPendingRequest(request);
421
return;
422
}
424
>
const cancellationTokenSource = new CancellationTokenSource();
425
>
let promise: Promise<any>;
426
>
427
>
try {
428
>
promise = channel.call(this.ctx, request.name, request.arg, cancellationTokenSource.token);
429
>
} catch (err) {
430
promise = Promise.reject(err);
431
}
433
>
const id = request.id;
434
>
435
>
promise.then(data => {
439
>
this.sendResponse({
440
>
id, data: {
441
>
message: err.message,
442
>
name: err.name,
443
>
stack: err.stack ? err.stack.split('\n') : undefined
444
>
}, type: ResponseType.PromiseError
445
>
});
446
>
} else {
447
this.sendResponse({ id, data: err, type: ResponseType.PromiseErrorObj });
448
}
450
>
disposable.dispose();
451
>
this.activeRequests.delete(request.id);
452
>
});
453
>
454
>
const disposable = toDisposable(() => cancellationTokenSource.cancel());
455
>
this.activeRequests.set(request.id, disposable);
456
>
}
458
>
private onEventListen(request: IRawEventListenRequest): void {
460
>
461
>
if (!channel) {
462
this.collectPendingRequest(request);
463
return;
464
}
466
>
const id = request.id;
467
>
const event = channel.listen(this.ctx, request.name, request.arg);
468
>
const disposable = event(data => this.sendResponse({ id, data, type: ResponseType.EventFire }));
469
>
470
>
this.activeRequests.set(request.id, disposable);
471
>
}
473
>
private disposeActiveRequest(request: IRawRequest): void {
475
>
476
>
if (disposable) {
477
>
disposable.dispose();
478
>
this.activeRequests.delete(request.id);
479
>
}
480
>
}
482
>
private collectPendingRequest(request: IRawPromiseRequest | IRawEventListenRequest): void {
483
let pendingRequests = this.pendingRequests.get(request.channelName);
484
485
if (!pendingRequests) {
486
pendingRequests = [];
487
this.pendingRequests.set(request.channelName, pendingRequests);
488
}
489
490
const timer = setTimeout(() => {
491
console.error(`Unknown channel: ${request.channelName}`);
492
493
if (request.type === RequestType.Promise) {
494
this.sendResponse({
495
id: request.id,
496
data: { name: 'Unknown channel', message: `Channel name '${request.channelName}' timed out after ${this.timeoutDelay}ms`, stack: undefined },
497
type: ResponseType.PromiseError
498
});
499
}
500
}, this.timeoutDelay);
501
502
pendingRequests.push({ request, timeoutTimer: timer });
503
}
505
>
private flushPendingRequests(channelName: string): void {
507
>
508
>
if (requests) {
509
for (const request of requests) {
510
clearTimeout(request.timeoutTimer);
511
512
switch (request.request.type) {
513
case RequestType.Promise: this.onPromise(request.request); break;
514
case RequestType.EventListen: this.onEventListen(request.request); break;
515
}
516
}
517
518
this.pendingRequests.delete(channelName);
519
}
522
>
public dispose(): void {
524
>
this.protocolListener.dispose();
525
>
this.protocolListener = null;
526
>
}
527
>
dispose(this.activeRequests.values());
528
>
this.activeRequests.clear();
529
>
}
531
>
532
>
export const enum RequestInitiator {
533
>
LocalSide = 0,
534
>
OtherSide = 1
535
>
}
536
>
537
>
export interface IIPCLogger {
538
>
logIncoming(msgLength: number, requestId: number, initiator: RequestInitiator, str: string, data?: any): void;
539
>
logOutgoing(msgLength: number, requestId: number, initiator: RequestInitiator, str: string, data?: any): void;
540
>
}
541
>
542
>
export class ChannelClient implements IChannelClient, IDisposable {
543
>
544
>
private isDisposed = false;
545
>
private state: State = State.Uninitialized;
546
>
private activeRequests = new Set<IDisposable>();
547
>
private handlers = new Map<number, IHandler>();
548
>
private lastRequestId = 0;
549
>
private protocolListener: IDisposable | null;
550
>
private logger: IIPCLogger | null;
551
>
552
>
private readonly _onDidInitialize = new Emitter<void>();
553
>
readonly onDidInitialize = this._onDidInitialize.event;
554
>
555
>
constructor(private protocol: IMessagePassingProtocol, logger: IIPCLogger | null = null) {
557
>
this.logger = logger;
558
>
}
560
>
getChannel<T extends IChannel>(channelName: string): T {
562
>
563
>
// eslint-disable-next-line local/code-no-dangerous-type-assertions
564
>
return {
565
>
call(command: string, arg?: any, cancellationToken?: CancellationToken) {
567
return Promise.reject(new CancellationError());
568
}
570
>
},
573
return Event.None;
574
}
576
>
}
578
>
}
580
>
private requestPromise(channelName: string, name: string, arg?: any, cancellationToken = CancellationToken.None): Promise<unknown> {
582
>
const type = RequestType.Promise;
583
>
const request: IRawRequest = { id, type, channelName, name, arg };
584
>
585
>
if (cancellationToken.isCancellationRequested) {
587
>
}
589
>
let disposable: IDisposable;
590
>
let disposableWithRequestCancel: IDisposable;
591
>
592
>
const result = new Promise((c, e) => {
593
>
if (cancellationToken.isCancellationRequested) {
594
return e(new CancellationError());
595
}
597
>
const doRequest = () => {
598
>
const handler: IHandler = response => {
600
>
case ResponseType.PromiseSuccess:
602
>
c(response.data);
603
>
break;
605
>
case ResponseType.PromiseError: {
607
>
const error = new Error(response.data.message);
608
>
error.stack = Array.isArray(response.data.stack) ? response.data.stack.join('\n') : response.data.stack;
609
>
error.name = response.data.name;
610
>
e(error);
611
>
break;
612
>
}
614
this.handlers.delete(id);
615
e(response.data);
616
break;
618
>
};
620
>
this.handlers.set(id, handler);
621
>
622
>
try {
623
>
this.sendRequest(request);
624
>
} catch (err) {
626
>
// request (e.g. an oversized argument). The handler was just
627
>
// registered but no request went out and it's only removed on a
628
>
// response, so without this it would leak (along with the rejected
629
>
// promise and error it retains). Clean up and reject.
630
>
this.handlers.delete(id);
631
>
e(err);
632
>
}
634
>
635
>
let uninitializedPromise: CancelablePromise<void> | null = null;
636
>
if (this.state === State.Idle) {
640
>
uninitializedPromise.then(() => {
641
>
uninitializedPromise = null;
642
>
doRequest();
643
>
});
644
>
}
646
>
const cancel = () => {
648
uninitializedPromise.cancel();
649
uninitializedPromise = null;
651
>
this.sendRequest({ id, type: RequestType.PromiseCancel });
652
>
}
653
>
654
>
e(new CancellationError());
655
>
};
657
>
disposable = cancellationToken.onCancellationRequested(cancel);
658
>
disposableWithRequestCancel = {
659
>
dispose: createSingleCallFunction(() => {
660
cancel();
661
disposable.dispose();
663
>
};
664
>
665
>
this.activeRequests.add(disposableWithRequestCancel);
666
>
});
667
>
668
>
return result.finally(() => {
669
>
disposable?.dispose(); // Seen as undefined in tests.
670
>
this.activeRequests.delete(disposableWithRequestCancel);
671
>
});
674
>
private requestEvent(channelName: string, name: string, arg?: any): Event<any> {
676
>
const type = RequestType.EventListen;
677
>
const request: IRawRequest = { id, type, channelName, name, arg };
678
>
679
>
let uninitializedPromise: CancelablePromise<void> | null = null;
680
>
681
>
const emitter = new Emitter<any>({
682
>
onWillAddFirstListener: () => {
683
>
const handler: IHandler = (res: IRawResponse) => emitter.fire((res as IRawEventFireResponse).data);
684
>
this.handlers.set(id, handler);
685
>
const doRequest = () => {
686
>
this.activeRequests.add(emitter);
687
>
this.sendRequest(request);
688
>
};
689
>
if (this.state === State.Idle) {
690
>
doRequest();
691
>
} else {
693
>
uninitializedPromise.then(() => {
694
>
uninitializedPromise = null;
695
>
doRequest();
696
>
});
697
>
}
699
>
onDidRemoveLastListener: () => {
700
>
if (uninitializedPromise) {
701
uninitializedPromise.cancel();
702
uninitializedPromise = null;
704
>
this.activeRequests.delete(emitter);
705
>
this.sendRequest({ id, type: RequestType.EventDispose });
706
>
}
707
>
this.handlers.delete(id);
708
>
}
709
>
});
710
>
711
>
return emitter.event;
712
>
}
714
>
private sendRequest(request: IRawRequest): void {
716
>
case RequestType.Promise:
717
>
case RequestType.EventListen: {
718
>
const msgLength = this.send([request.type, request.id, request.channelName, request.name], request.arg);
719
>
this.logger?.logOutgoing(msgLength, request.id, RequestInitiator.LocalSide, `${requestTypeToStr(request.type)}: ${request.channelName}.${request.name}`, request.arg);
720
>
return;
721
>
}
722
>
723
>
case RequestType.PromiseCancel:
724
>
case RequestType.EventDispose: {
726
>
this.logger?.logOutgoing(msgLength, request.id, RequestInitiator.LocalSide, requestTypeToStr(request.type));
727
>
return;
728
>
}
730
>
}
732
>
private send(header: unknown, body: any = undefined): number {
734
>
try {
735
>
serialize(writer, header);
736
>
serialize(writer, body);
737
>
return this.sendBuffer(writer.buffer);
738
>
} finally {
739
>
writer.dispose();
740
>
}
741
>
}
743
>
private sendBuffer(message: VSBuffer): number {
745
>
this.protocol.send(message);
746
>
return message.byteLength;
747
>
} catch (err) {
748
// noop
749
return 0;
750
}
753
>
private onBuffer(message: VSBuffer): void {
755
>
const header = deserialize(reader);
756
>
const body = deserialize(reader);
757
>
const type: ResponseType = header[0];
758
>
759
>
switch (type) {
760
>
case ResponseType.Initialize:
761
>
this.logger?.logIncoming(message.byteLength, 0, RequestInitiator.LocalSide, responseTypeToStr(type));
762
>
return this.onResponse({ type: header[0] });
763
>
764
>
case ResponseType.PromiseSuccess:
765
>
case ResponseType.PromiseError:
766
>
case ResponseType.EventFire:
767
>
case ResponseType.PromiseErrorObj:
768
>
this.logger?.logIncoming(message.byteLength, header[1], RequestInitiator.LocalSide, responseTypeToStr(type), body);
ipc.ts ×5
769
>
return this.onResponse({ type: header[0], id: header[1], data: body });
771
>
}
773
>
private onResponse(response: IRawResponse): void {
775
>
this.state = State.Idle;
776
>
this._onDidInitialize.fire();
777
>
return;
778
>
}
780
>
const handler = this.handlers.get(response.id);
781
>
782
>
handler?.(response);
785
>
@memoize
786
>
get onDidInitializePromise(): Promise<void> {
788
>
}
790
>
private whenInitialized(): Promise<void> {
792
return Promise.resolve();
794
>
return this.onDidInitializePromise;
795
>
}
796
>
}
798
>
dispose(): void {
800
>
if (this.protocolListener) {
801
>
this.protocolListener.dispose();
802
>
this.protocolListener = null;
803
>
}
804
>
dispose(this.activeRequests.values());
805
>
this.activeRequests.clear();
806
>
this._onDidInitialize.dispose();
807
>
}
809
>
810
>
export interface ClientConnectionEvent {
811
>
protocol: IMessagePassingProtocol;
812
>
readonly onDidClientDisconnect: Event<void>;
813
>
}
814
>
815
>
interface Connection<TContext> extends Client<TContext> {
816
>
readonly channelServer: ChannelServer<TContext>;
817
>
readonly channelClient: ChannelClient;
818
>
}
819
>
820
>
/**
821
>
* An `IPCServer` is both a channel server and a routing channel
822
>
* client.
823
>
*
824
>
* As the owner of a protocol, you should extend both this
825
>
* and the `IPCClient` classes to get IPC implementations
826
>
* for your protocol.
827
>
*/
828
>
export class IPCServer<TContext = string> implements IChannelServer<TContext>, IRoutingChannelClient<TContext>, IConnectionHub<TContext>, IDisposable {
829
>
830
>
private channels = new Map<string, IServerChannel<TContext>>();
831
>
private _connections = new Set<Connection<TContext>>();
832
>
833
>
private readonly _onDidAddConnection = new Emitter<Connection<TContext>>();
834
>
readonly onDidAddConnection: Event<Connection<TContext>> = this._onDidAddConnection.event;
835
>
836
>
private readonly _onDidRemoveConnection = new Emitter<Connection<TContext>>();
837
>
readonly onDidRemoveConnection: Event<Connection<TContext>> = this._onDidRemoveConnection.event;
838
>
839
>
private readonly disposables = new DisposableStore();
840
>
841
>
get connections(): Connection<TContext>[] {
842
>
const result: Connection<TContext>[] = [];
843
>
this._connections.forEach(ctx => result.push(ctx));
844
>
return result;
845
>
}
846
>
847
>
constructor(onDidClientConnect: Event<ClientConnectionEvent>, ipcLogger?: IIPCLogger | null, timeoutDelay?: number) {
849
>
const onFirstMessage = Event.once(protocol.onMessage);
850
>
851
>
const connectionDisposables = new DisposableStore();
852
>
853
>
const onFirstMessageDisposable = onFirstMessage(msg => {
854
>
const reader = new BufferReader(msg);
855
>
const ctx = deserialize(reader) as TContext;
856
>
857
>
const channelServer = new ChannelServer(protocol, ctx, ipcLogger, timeoutDelay);
858
>
const channelClient = new ChannelClient(protocol, ipcLogger);
859
>
860
>
this.channels.forEach((channel, name) => channelServer.registerChannel(name, channel));
861
>
862
>
const connection: Connection<TContext> = { channelServer, channelClient, ctx };
863
>
this._connections.add(connection);
864
>
this._onDidAddConnection.fire(connection);
865
>
866
>
connectionDisposables.add(onDidClientDisconnect(() => {
868
>
channelClient.dispose();
869
>
this._connections.delete(connection);
870
>
this._onDidRemoveConnection.fire(connection);
871
>
this.disposables.delete(connectionDisposables);
872
>
connectionDisposables.dispose();
874
>
});
875
>
876
>
connectionDisposables.add(onFirstMessageDisposable);
877
>
this.disposables.add(connectionDisposables);
878
>
}));
879
>
}
881
>
/**
882
>
* Get a channel from a remote client. When passed a router,
883
>
* one can specify which client it wants to call and listen to/from.
884
>
* Otherwise, when calling without a router, a random client will
885
>
* be selected and when listening without a router, every client
886
>
* will be listened to.
887
>
*/
888
>
getChannel<T extends IChannel>(channelName: string, router: IClientRouter<TContext>): T;
889
>
getChannel<T extends IChannel>(channelName: string, clientFilter: (client: Client<TContext>) => boolean): T;
890
>
getChannel<T extends IChannel>(channelName: string, routerOrClientFilter: IClientRouter<TContext> | ((client: Client<TContext>) => boolean)): T {
892
>
893
>
// eslint-disable-next-line local/code-no-dangerous-type-assertions
894
>
return {
895
>
call(command: string, arg?: any, cancellationToken?: CancellationToken): Promise<T> {
896
let connectionPromise: Promise<Client<TContext>>;
897
898
if (isFunction(routerOrClientFilter)) {
899
// when no router is provided, we go random client picking
900
const connection = getRandomElement(that.connections.filter(routerOrClientFilter));
901
902
connectionPromise = connection
903
// if we found a client, let's call on it
904
? Promise.resolve(connection)
905
// else, let's wait for a client to come along
906
: Event.toPromise(Event.filter(that.onDidAddConnection, routerOrClientFilter));
907
} else {
908
connectionPromise = routerOrClientFilter.routeCall(that, command, arg);
909
}
910
911
const channelPromise = connectionPromise
912
.then(connection => (connection as Connection<TContext>).channelClient.getChannel(channelName));
913
914
return getDelayedChannel(channelPromise)
915
.call(command, arg, cancellationToken);
916
},
918
>
if (isFunction(routerOrClientFilter)) {
919
>
return that.getMulticastEvent(channelName, routerOrClientFilter, event, arg);
920
>
}
921
922
const channelPromise = routerOrClientFilter.routeEvent(that, event, arg)
923
.then(connection => (connection as Connection<TContext>).channelClient.getChannel(channelName));
924
925
return getDelayedChannel(channelPromise)
926
.listen(event, arg);
928
>
} as T;
929
>
}
931
>
private getMulticastEvent<T extends IChannel>(channelName: string, clientFilter: (client: Client<TContext>) => boolean, eventName: string, arg: any): Event<T> {
933
>
let disposables: DisposableStore | undefined;
934
>
935
>
// Create an emitter which hooks up to all clients
936
>
// as soon as first listener is added. It also
937
>
// disconnects from all clients as soon as the last listener
938
>
// is removed.
939
>
const emitter = new Emitter<T>({
940
>
onWillAddFirstListener: () => {
941
>
disposables = new DisposableStore();
942
>
943
>
// The event multiplexer is useful since the active
944
>
// client list is dynamic. We need to hook up and disconnection
945
>
// to/from clients as they come and go.
946
>
const eventMultiplexer = new EventMultiplexer<T>();
947
>
const map = new Map<Connection<TContext>, IDisposable>();
948
>
949
>
const onDidAddConnection = (connection: Connection<TContext>) => {
950
>
const channel = connection.channelClient.getChannel(channelName);
951
>
const event = channel.listen<T>(eventName, arg);
952
>
const disposable = eventMultiplexer.add(event);
953
>
954
>
map.set(connection, disposable);
955
>
};
956
>
957
>
const onDidRemoveConnection = (connection: Connection<TContext>) => {
958
>
const disposable = map.get(connection);
959
>
960
>
if (!disposable) {
961
return;
962
}
964
>
disposable.dispose();
965
>
map.delete(connection);
966
>
};
967
>
968
>
that.connections.filter(clientFilter).forEach(onDidAddConnection);
969
>
Event.filter(that.onDidAddConnection, clientFilter)(onDidAddConnection, undefined, disposables);
970
>
that.onDidRemoveConnection(onDidRemoveConnection, undefined, disposables);
971
>
eventMultiplexer.event(emitter.fire, emitter, disposables);
972
>
973
>
disposables.add(eventMultiplexer);
974
>
},
975
>
onDidRemoveLastListener: () => {
976
>
disposables?.dispose();
977
>
disposables = undefined;
978
>
}
979
>
});
980
>
that.disposables.add(emitter);
981
>
982
>
return emitter.event;
983
>
}
985
>
registerChannel(channelName: string, channel: IServerChannel<TContext>): void {
987
>
988
>
for (const connection of this._connections) {
989
connection.channelServer.registerChannel(channelName, channel);
990
}
993
>
dispose(): void {
995
>
996
>
for (const connection of this._connections) {
998
>
connection.channelServer.dispose();
999
>
}
1001
>
this._connections.clear();
1002
>
this.channels.clear();
1003
>
this._onDidAddConnection.dispose();
1004
>
this._onDidRemoveConnection.dispose();
1005
>
}
1007
>
1008
>
/**
1009
>
* An `IPCClient` is both a channel client and a channel server.
1010
>
*
1011
>
* As the owner of a protocol, you should extend both this
1012
>
* and the `IPCServer` classes to get IPC implementations
1013
>
* for your protocol.
1014
>
*/
1015
>
export class IPCClient<TContext = string> implements IChannelClient, IChannelServer<TContext>, IDisposable {
1016
>
1017
>
private channelClient: ChannelClient;
1018
>
private channelServer: ChannelServer<TContext>;
1019
>
1020
>
constructor(protocol: IMessagePassingProtocol, ctx: TContext, ipcLogger: IIPCLogger | null = null) {
1022
>
try {
1023
>
serialize(writer, ctx);
1024
>
protocol.send(writer.buffer);
1025
>
} finally {
1026
>
writer.dispose();
1027
>
}
1028
>
1029
>
this.channelClient = new ChannelClient(protocol, ipcLogger);
1030
>
this.channelServer = new ChannelServer(protocol, ctx, ipcLogger);
1031
>
}
1033
>
getChannel<T extends IChannel>(channelName: string): T {
1035
>
}
1037
>
registerChannel(channelName: string, channel: IServerChannel<TContext>): void {
1039
>
}
1041
>
dispose(): void {
1043
>
this.channelServer.dispose();
1044
>
}
1046
>
1047
>
export function getDelayedChannel<T extends IChannel>(promise: Promise<T>): T {
1048
// eslint-disable-next-line local/code-no-dangerous-type-assertions
1049
return {
1050
call(command: string, arg?: any, cancellationToken?: CancellationToken): Promise<T> {
1051
return promise.then(c => c.call<T>(command, arg, cancellationToken));
1052
},
1053
1054
listen<T>(event: string, arg?: any): Event<T> {
1055
const relay = new Relay<any>();
1056
promise.then(c => relay.input = c.listen(event, arg));
1057
return relay.event;
1058
}
1059
} as T;
1060
}
1062
>
export function getNextTickChannel<T extends IChannel>(channel: T): T {
1063
let didTick = false;
1064
1065
// eslint-disable-next-line local/code-no-dangerous-type-assertions
1066
return {
1067
call<T>(command: string, arg?: any, cancellationToken?: CancellationToken): Promise<T> {
1068
if (didTick) {
1069
return channel.call(command, arg, cancellationToken);
1070
}
1071
1072
return timeout(0)
1073
.then(() => didTick = true)
1074
.then(() => channel.call<T>(command, arg, cancellationToken));
1075
},
1076
listen<T>(event: string, arg?: any): Event<T> {
1077
if (didTick) {
1078
return channel.listen<T>(event, arg);
1079
}
1080
1081
const relay = new Relay<T>();
1082
1083
timeout(0)
1084
.then(() => didTick = true)
1085
.then(() => relay.input = channel.listen<T>(event, arg));
1086
1087
return relay.event;
1088
}
1089
} as T;
1090
}
1092
>
export class StaticRouter<TContext = string> implements IClientRouter<TContext> {
1093
>
1094
>
constructor(private fn: (ctx: TContext) => boolean | Promise<boolean>) { }
1095
>
1096
>
routeCall(hub: IConnectionHub<TContext>): Promise<Client<TContext>> {
1097
return this.route(hub);
1098
}
1100
>
routeEvent(hub: IConnectionHub<TContext>): Promise<Client<TContext>> {
1101
return this.route(hub);
1102
}
1104
>
private async route(hub: IConnectionHub<TContext>): Promise<Client<TContext>> {
1105
for (const connection of hub.connections) {
1106
if (await Promise.resolve(this.fn(connection.ctx))) {
1107
return Promise.resolve(connection);
1108
}
1109
}
1110
1111
await Event.toPromise(hub.onDidAddConnection);
1112
return await this.route(hub);
1113
}
1115
>
1116
>
/**
1117
>
* Use ProxyChannels to automatically wrapping and unwrapping
1118
>
* services to/from IPC channels, instead of manually wrapping
1119
>
* each service method and event.
1120
>
*
1121
>
* Restrictions:
1122
>
* - If marshalling is enabled, only `URI` and `RegExp` is converted
1123
>
* automatically for you
1124
>
* - Events must follow the naming convention `onUpperCase`
1125
>
* - `CancellationToken` is currently not supported
1126
>
* - If a context is provided, you can use `AddFirstParameterToFunctions`
1127
>
* utility to signal this in the receiving side type
1128
>
*/
1129
>
export namespace ProxyChannel {
1130
>
1131
>
export interface IProxyOptions {
1132
>
1133
>
/**
1134
>
* Disables automatic marshalling of `URI`.
1135
>
* If marshalling is disabled, `UriComponents`
1136
>
* must be used instead.
1137
>
*/
1138
>
disableMarshalling?: boolean;
1139
>
}
1140
>
1141
>
export interface ICreateServiceChannelOptions extends IProxyOptions { }
1142
>
1143
>
export function fromService<TContext>(service: unknown, disposables: DisposableStore, options?: ICreateServiceChannelOptions): IServerChannel<TContext> {
1145
>
const disableMarshalling = options?.disableMarshalling;
1146
>
1147
>
// Buffer any event that should be supported by
1148
>
// iterating over all property keys and finding them
1149
>
// However, this will not work for services that
1150
>
// are lazy and use a Proxy within. For that we
1151
>
// still need to check later (see below).
1152
>
const mapEventNameToEvent = new Map<string, Event<unknown>>();
1153
>
for (const key in handler) {
1154
>
if (propertyIsEvent(key)) {
1155
>
mapEventNameToEvent.set(key, Event.buffer(handler[key] as Event<unknown>, key, true, undefined, disposables));
1156
>
}
1157
>
}
1158
>
1159
>
return new class implements IServerChannel {
1160
>
1161
>
listen<T>(_: unknown, event: string, arg: any): Event<T> {
1163
>
if (eventImpl) {
1164
>
return eventImpl as Event<T>;
1165
>
}
1166
1167
const target = handler[event];
1168
if (typeof target === 'function') {
1169
if (propertyIsDynamicEvent(event)) {
1170
return target.call(handler, arg);
1171
}
1172
1173
if (propertyIsEvent(event)) {
1174
mapEventNameToEvent.set(event, Event.buffer(handler[event] as Event<unknown>, event, true, undefined, disposables));
1175
1176
return mapEventNameToEvent.get(event) as Event<T>;
1177
}
1178
}
1179
1180
throw new ErrorNoTelemetry(`Event not found: ${event}`);
1183
>
call(_: unknown, command: string, args?: any[]): Promise<any> {
1185
>
if (typeof target === 'function') {
1186
>
1187
>
// Revive unless marshalling disabled
1188
>
if (!disableMarshalling && Array.isArray(args)) {
1189
>
for (let i = 0; i < args.length; i++) {
1191
>
}
1193
>
1194
>
let res = target.apply(handler, args);
1195
>
if (!(res instanceof Promise)) {
1196
res = Promise.resolve(res);
1197
}
1199
>
}
1200
1201
throw new ErrorNoTelemetry(`Method not found: ${command}`);
1204
>
}
1206
>
export interface ICreateProxyServiceOptions extends IProxyOptions {
1207
>
1208
>
/**
1209
>
* If provided, will add the value of `context`
1210
>
* to each method call to the target.
1211
>
*/
1212
>
context?: unknown;
1213
>
1214
>
/**
1215
>
* If provided, will not proxy any of the properties
1216
>
* that are part of the Map but rather return that value.
1217
>
*/
1218
>
properties?: Map<string, unknown>;
1219
>
}
1220
>
1221
>
export function toService<T extends object>(channel: IChannel, options?: ICreateProxyServiceOptions): T {
1223
>
1224
>
return new Proxy({}, {
1225
>
get(_target: T, propKey: PropertyKey) {
1226
>
if (typeof propKey === 'string') {
1227
>
1228
>
// Check for predefined values
1229
>
if (options?.properties?.has(propKey)) {
1230
return options.properties.get(propKey);
1231
}
1233
>
// Dynamic Event
1234
>
if (propertyIsDynamicEvent(propKey)) {
1235
return function (arg: unknown) {
1236
return channel.listen(propKey, arg);
1237
};
1238
}
1240
>
// Event
1241
>
if (propertyIsEvent(propKey)) {
1243
>
}
1245
>
// Function
1246
>
return async function (...args: unknown[]) {
1247
>
1248
>
// Add context if any
1249
>
let methodArgs: unknown[];
1250
>
if (options && !isUndefinedOrNull(options.context)) {
1254
>
}
1256
>
const result = await channel.call(propKey, methodArgs);
1258
>
// Revive unless marshalling disabled
1259
>
if (!disableMarshalling) {
1260
>
return revive(result);
1261
>
}
1262
1263
return result;
1265
>
}
1266
1267
throw new ErrorNoTelemetry(`Property not found: ${String(propKey)}`);
1269
>
}) as T;
1270
>
}
1272
>
function propertyIsEvent(name: string): boolean {
1274
>
return name[0] === 'o' && name[1] === 'n' && strings.isUpperAsciiLetter(name.charCodeAt(2));
1275
>
}
1277
>
function propertyIsDynamicEvent(name: string): boolean {
1278
>
// Assume a property is a dynamic event (a method that returns an event) if it has a form of "onDynamicSomething"
ipc.ts ×6
1279
>
return /^onDynamic/.test(name) && strings.isUpperAsciiLetter(name.charCodeAt(9));
1280
>
}
1282
>
1283
>
const colorTables = [
1284
>
['#2977B1', '#FC802D', '#34A13A', '#D3282F', '#9366BA'],
1285
>
['#8B564C', '#E177C0', '#7F7F7F', '#BBBE3D', '#2EBECD']
1286
>
];
1287
>
1288
function prettyWithoutArrays(data: unknown): any {
1289
if (Array.isArray(data)) {
1290
return data;
1291
}
1292
if (data && typeof data === 'object' && typeof data.toString === 'function') {
1293
const result = data.toString();
1294
if (result !== '[object Object]') {
1295
return result;
1296
}
1297
}
1298
return data;
1299
}
1301
function pretty(data: unknown): any {
1302
if (Array.isArray(data)) {
1303
return data.map(prettyWithoutArrays);
1304
}
1305
return prettyWithoutArrays(data);
1306
}
1308
function logWithColors(direction: string, totalLength: number, msgLength: number, req: number, initiator: RequestInitiator, str: string, data: any): void {
1309
data = pretty(data);
1310
1311
const colorTable = colorTables[initiator];
1312
const color = colorTable[req % colorTable.length];
1313
let args = [`%c[${direction}]%c[${String(totalLength).padStart(7, ' ')}]%c[len: ${String(msgLength).padStart(5, ' ')}]%c${String(req).padStart(5, ' ')} - ${str}`, 'color: darkgreen', 'color: grey', 'color: grey', `color: ${color}`];
1314
if (/\($/.test(str)) {
1315
args = args.concat(data);
1316
args.push(')');
1317
} else {
1318
args.push(data);
1319
}
1320
console.log.apply(console, args as [string, ...string[]]);
1321
}
1323
>
export class IPCLogger implements IIPCLogger {
1324
>
private _totalIncoming = 0;
1325
>
private _totalOutgoing = 0;
1326
>
1327
>
constructor(
1328
private readonly _outgoingPrefix: string,
1329
private readonly _incomingPrefix: string,
1330
) { }
1332
>
public logOutgoing(msgLength: number, requestId: number, initiator: RequestInitiator, str: string, data?: any): void {
1333
this._totalOutgoing += msgLength;
1334
logWithColors(this._outgoingPrefix, this._totalOutgoing, msgLength, requestId, initiator, str, data);
1335
}
1337
>
public logIncoming(msgLength: number, requestId: number, initiator: RequestInitiator, str: string, data?: any): void {
1338
this._totalIncoming += msgLength;
1339
logWithColors(this._incomingPrefix, this._totalIncoming, msgLength, requestId, initiator, str, data);
1340
}