src/vs/base/common/stream.ts
790 LOC · 774 covered · 16 uncovered · 149 ranges · 13454 concepts · 69 introducers · 6883 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
>
/*---------------------------------------------------------------------------------------------
buffer.ts ×42
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 { CancellationToken } from './cancellation.js';
7
>
import { onUnexpectedError } from './errors.js';
8
>
import { DisposableStore, toDisposable } from './lifecycle.js';
9
>
10
>
/**
11
>
* The payload that flows in readable stream events.
12
>
*/
13
>
export type ReadableStreamEventPayload<T> = T | Error | 'end';
14
>
15
>
export interface ReadableStreamEvents<T> {
16
>
17
>
/**
18
>
* The 'data' event is emitted whenever the stream is
19
>
* relinquishing ownership of a chunk of data to a consumer.
20
>
*
21
>
* NOTE: PLEASE UNDERSTAND THAT ADDING A DATA LISTENER CAN
22
>
* TURN THE STREAM INTO FLOWING MODE. IT IS THEREFOR THE
23
>
* LAST LISTENER THAT SHOULD BE ADDED AND NOT THE FIRST
24
>
*
25
>
* Use `listenStream` as a helper method to listen to
26
>
* stream events in the right order.
27
>
*/
28
>
on(event: 'data', callback: (data: T) => void): void;
29
>
30
>
/**
31
>
* Emitted when any error occurs.
32
>
*/
33
>
on(event: 'error', callback: (err: Error) => void): void;
34
>
35
>
/**
36
>
* The 'end' event is emitted when there is no more data
37
>
* to be consumed from the stream. The 'end' event will
38
>
* not be emitted unless the data is completely consumed.
39
>
*/
40
>
on(event: 'end', callback: () => void): void;
41
>
}
42
>
43
>
/**
44
>
* A interface that emulates the API shape of a node.js readable
45
>
* stream for use in native and web environments.
46
>
*/
47
>
export interface ReadableStream<T> extends ReadableStreamEvents<T> {
48
>
49
>
/**
50
>
* Stops emitting any events until resume() is called.
51
>
*/
52
>
pause(): void;
53
>
54
>
/**
55
>
* Starts emitting events again after pause() was called.
56
>
*/
57
>
resume(): void;
58
>
59
>
/**
60
>
* Destroys the stream and stops emitting any event.
61
>
*/
62
>
destroy(): void;
63
>
64
>
/**
65
>
* Allows to remove a listener that was previously added.
66
>
*/
67
>
removeListener(event: string, callback: Function): void;
68
>
}
69
>
70
>
/**
71
>
* A interface that emulates the API shape of a node.js readable
72
>
* for use in native and web environments.
73
>
*/
74
>
export interface Readable<T> {
75
>
76
>
/**
77
>
* Read data from the underlying source. Will return
78
>
* null to indicate that no more data can be read.
79
>
*/
80
>
read(): T | null;
81
>
}
82
>
83
>
export function isReadable<T>(obj: unknown): obj is Readable<T> {
85
>
if (!candidate) {
86
>
return false;
87
>
}
88
>
89
>
return typeof candidate.read === 'function';
90
>
}
92
>
/**
93
>
* A interface that emulates the API shape of a node.js writeable
94
>
* stream for use in native and web environments.
95
>
*/
96
>
export interface WriteableStream<T> extends ReadableStream<T> {
97
>
98
>
/**
99
>
* Writing data to the stream will trigger the on('data')
100
>
* event listener if the stream is flowing and buffer the
101
>
* data otherwise until the stream is flowing.
102
>
*
103
>
* If a `highWaterMark` is configured and writing to the
104
>
* stream reaches this mark, a promise will be returned
105
>
* that should be awaited on before writing more data.
106
>
* Otherwise there is a risk of buffering a large number
107
>
* of data chunks without consumer.
108
>
*/
109
>
write(data: T): void | Promise<void>;
110
>
111
>
/**
112
>
* Signals an error to the consumer of the stream via the
113
>
* on('error') handler if the stream is flowing.
114
>
*
115
>
* NOTE: call `end` to signal that the stream has ended,
116
>
* this DOES NOT happen automatically from `error`.
117
>
*/
118
>
error(error: Error): void;
119
>
120
>
/**
121
>
* Signals the end of the stream to the consumer. If the
122
>
* result is provided, will trigger the on('data') event
123
>
* listener if the stream is flowing and buffer the data
124
>
* otherwise until the stream is flowing.
125
>
*/
126
>
end(result?: T): void;
127
>
}
128
>
129
>
/**
130
>
* A stream that has a buffer already read. Returns the original stream
131
>
* that was read as well as the chunks that got read.
132
>
*
133
>
* The `ended` flag indicates if the stream has been fully consumed.
134
>
*/
135
>
export interface ReadableBufferedStream<T> {
136
>
137
>
/**
138
>
* The original stream that is being read.
139
>
*/
140
>
stream: ReadableStream<T>;
141
>
142
>
/**
143
>
* An array of chunks already read from this stream.
144
>
*/
145
>
buffer: T[];
146
>
147
>
/**
148
>
* Signals if the stream has ended or not. If not, consumers
149
>
* should continue to read from the stream until consumed.
150
>
*/
151
>
ended: boolean;
152
>
}
153
>
154
>
export function isReadableStream<T>(obj: unknown): obj is ReadableStream<T> {
156
>
if (!candidate) {
157
>
return false;
158
>
}
159
>
160
>
return [candidate.on, candidate.pause, candidate.resume, candidate.destroy].every(fn => typeof fn === 'function');
161
>
}
163
>
export function isReadableBufferedStream<T>(obj: unknown): obj is ReadableBufferedStream<T> {
165
>
if (!candidate) {
166
return false;
167
}
169
>
return isReadableStream(candidate.stream) && Array.isArray(candidate.buffer) && typeof candidate.ended === 'boolean';
170
>
}
172
>
export interface IReducer<T, R = T> {
173
>
(data: T[]): R;
174
>
}
175
>
176
>
export interface IDataTransformer<Original, Transformed> {
177
>
(data: Original): Transformed;
178
>
}
179
>
180
>
export interface IErrorTransformer {
181
>
(error: Error): Error;
182
>
}
183
>
184
>
export interface ITransformer<Original, Transformed> {
185
>
data: IDataTransformer<Original, Transformed>;
186
>
error?: IErrorTransformer;
187
>
}
188
>
189
>
export function newWriteableStream<T>(reducer: IReducer<T> | null, options?: WriteableStreamOptions): WriteableStream<T> {
191
>
}
193
>
export interface WriteableStreamOptions {
194
>
195
>
/**
196
>
* The number of objects to buffer before WriteableStream#write()
197
>
* signals back that the buffer is full. Can be used to reduce
198
>
* the memory pressure when the stream is not flowing.
199
>
*/
200
>
highWaterMark?: number;
201
>
}
202
>
203
>
class WriteableStreamImpl<T> implements WriteableStream<T> {
204
>
205
>
private readonly state = {
206
>
flowing: false,
207
>
ended: false,
208
>
destroyed: false
209
>
};
210
>
211
>
private readonly buffer = {
212
>
data: [] as T[],
213
>
error: [] as Error[]
214
>
};
215
>
216
>
private readonly listeners = {
217
>
data: [] as { (data: T): void }[],
218
>
error: [] as { (error: Error): void }[],
219
>
end: [] as { (): void }[]
220
>
};
221
>
222
>
private readonly pendingWritePromises: Function[] = [];
223
>
224
>
/**
225
>
* @param reducer a function that reduces the buffered data into a single object;
226
>
* because some objects can be complex and non-reducible, we also
227
>
* allow passing the explicit `null` value to skip the reduce step
228
>
* @param options stream options
229
>
*/
230
>
constructor(private reducer: IReducer<T> | null, private options?: WriteableStreamOptions) { }
231
>
232
>
pause(): void {
234
return;
235
}
237
>
this.state.flowing = false;
238
>
}
240
>
resume(): void {
242
return;
243
}
245
>
if (!this.state.flowing) {
246
>
this.state.flowing = true;
247
>
248
>
// emit buffered events
249
>
this.flowData();
250
>
this.flowErrors();
251
>
this.flowEnd();
252
>
}
253
>
}
255
>
write(data: T): void | Promise<void> {
258
>
}
260
>
// flowing: directly send the data to listeners
261
>
if (this.state.flowing) {
263
>
}
265
>
// not yet flowing: buffer data until flowing
266
>
else {
267
>
this.buffer.data.push(data);
268
>
269
>
// highWaterMark: if configured, signal back when buffer reached limits
270
>
if (typeof this.options?.highWaterMark === 'number' && this.buffer.data.length > this.options.highWaterMark) {
272
>
}
276
>
error(error: Error): void {
279
>
}
281
>
// flowing: directly send the error to listeners
282
>
if (this.state.flowing) {
284
>
}
286
>
// not yet flowing: buffer errors until flowing
287
>
else {
288
>
this.buffer.error.push(error);
289
>
}
292
>
end(result?: T): void {
295
>
}
297
>
// end with data if provided
298
>
if (typeof result !== 'undefined') {
300
>
}
302
>
// flowing: send end event to listeners
303
>
if (this.state.flowing) {
305
>
306
>
this.destroy();
307
>
}
309
>
// not yet flowing: remember state
310
>
else {
311
>
this.state.ended = true;
312
>
}
315
>
private emitData(data: T): void {
316
>
this.listeners.data.slice(0).forEach(listener => listener(data)); // slice to avoid listener mutation from delivering event
stream.ts ×1
317
>
}
319
>
private emitError(error: Error): void {
321
onUnexpectedError(error); // nobody listened to this error so we log it as unexpected
323
>
this.listeners.error.slice(0).forEach(listener => listener(error)); // slice to avoid listener mutation from delivering event
324
>
}
325
>
}
327
>
private emitEnd(): void {
328
>
this.listeners.end.slice(0).forEach(listener => listener()); // slice to avoid listener mutation from delivering event
stream.ts ×1
329
>
}
331
>
on(event: 'data', callback: (data: T) => void): void;
332
>
on(event: 'error', callback: (err: Error) => void): void;
333
>
on(event: 'end', callback: () => void): void;
334
>
on(event: 'data' | 'error' | 'end', callback: ((data: T) => void) | ((err: Error) => void) | (() => void)): void {
337
>
}
339
>
switch (event) {
340
>
case 'data':
341
>
this.listeners.data.push(callback as (data: T) => void);
342
>
343
>
// switch into flowing mode as soon as the first 'data'
344
>
// listener is added and we are not yet in flowing mode
345
>
this.resume();
346
>
347
>
break;
348
>
349
>
case 'end':
351
>
352
>
// emit 'end' event directly if we are flowing
353
>
// and the end has already been reached
354
>
//
355
>
// finish() when it went through
356
>
if (this.state.flowing && this.flowEnd()) {
358
>
}
360
>
break;
362
>
case 'error':
364
>
365
>
// emit buffered 'error' events unless done already
366
>
// now that we know that we have at least one listener
367
>
if (this.state.flowing) {
369
>
}
371
>
break;
373
>
}
375
>
removeListener(event: string, callback: Function): void {
377
return;
378
}
380
>
let listeners: unknown[] | undefined = undefined;
381
>
382
>
switch (event) {
383
>
case 'data':
385
>
break;
387
>
case 'end':
389
>
break;
391
>
case 'error':
393
>
break;
395
>
396
>
if (listeners) {
397
>
const index = listeners.indexOf(callback);
398
>
if (index >= 0) {
399
>
listeners.splice(index, 1);
400
>
}
401
>
}
402
>
}
404
>
private flowData(): void {
406
>
if (this.buffer.data.length === 0) {
408
>
}
410
>
// if buffer data can be reduced into a single object,
411
>
// emit the reduced data
412
>
if (typeof this.reducer === 'function') {
414
>
415
>
this.emitData(fullDataBuffer);
418
>
for (const data of this.buffer.data) {
419
>
this.emitData(data);
420
>
}
421
>
}
423
>
this.buffer.data.length = 0;
424
>
425
>
// when the buffer is empty, resolve all pending writers
426
>
const pendingWritePromises = [...this.pendingWritePromises];
427
>
this.pendingWritePromises.length = 0;
428
>
pendingWritePromises.forEach(pendingWritePromise => pendingWritePromise());
431
>
private flowErrors(): void {
435
>
}
437
>
this.buffer.error.length = 0;
438
>
}
441
>
private flowEnd(): boolean {
444
>
445
>
return this.listeners.end.length > 0;
446
>
}
448
>
return false;
451
>
destroy(): void {
453
>
this.state.destroyed = true;
454
>
this.state.ended = true;
455
>
456
>
this.buffer.data.length = 0;
457
>
this.buffer.error.length = 0;
458
>
459
>
this.listeners.data.length = 0;
460
>
this.listeners.error.length = 0;
461
>
this.listeners.end.length = 0;
462
>
463
>
this.pendingWritePromises.length = 0;
464
>
}
465
>
}
467
>
468
>
/**
469
>
* Helper to fully read a T readable into a T.
470
>
*/
471
>
export function consumeReadable<T>(readable: Readable<T>, reducer: IReducer<T>): T {
473
>
474
>
let chunk: T | null;
475
>
while ((chunk = readable.read()) !== null) {
477
>
}
479
>
return reducer(chunks);
480
>
}
482
>
/**
483
>
* Helper to read a T readable up to a maximum of chunks. If the limit is
484
>
* reached, will return a readable instead to ensure all data can still
485
>
* be read.
486
>
*/
487
>
export function peekReadable<T>(readable: Readable<T>, reducer: IReducer<T>, maxChunks: number): T | Readable<T> {
489
>
490
>
let chunk: T | null | undefined = undefined;
491
>
while ((chunk = readable.read()) !== null && chunks.length < maxChunks) {
492
>
chunks.push(chunk);
493
>
}
494
>
495
>
// If the last chunk is null, it means we reached the end of
496
>
// the readable and return all the data at once
497
>
if (chunk === null && chunks.length > 0) {
498
>
return reducer(chunks);
499
>
}
500
>
501
>
// Otherwise, we still have a chunk, it means we reached the maxChunks
502
>
// value and as such we return a new Readable that first returns
503
>
// the existing read chunks and then continues with reading from
504
>
// the underlying readable.
505
>
return {
506
>
read: () => {
507
>
508
>
// First consume chunks from our array
509
>
if (chunks.length > 0) {
510
>
return chunks.shift()!;
511
>
}
512
>
513
>
// Then ensure to return our last read chunk
514
>
if (typeof chunk !== 'undefined') {
515
>
const lastReadChunk = chunk;
516
>
517
>
// explicitly use undefined here to indicate that we consumed
518
>
// the chunk, which could have either been null or valued.
519
>
chunk = undefined;
520
>
521
>
return lastReadChunk;
522
>
}
523
>
524
>
// Finally delegate back to the Readable
525
>
return readable.read();
526
>
}
527
>
};
528
>
}
530
>
/**
531
>
* Helper to fully read a T stream into a T or consuming
532
>
* a stream fully, awaiting all the events without caring
533
>
* about the data.
534
>
*/
535
>
export function consumeStream<T, R = T>(stream: ReadableStreamEvents<T>, reducer: IReducer<T, R>): Promise<R>;
536
>
export function consumeStream(stream: ReadableStreamEvents<unknown>): Promise<undefined>;
537
>
export function consumeStream<T, R = T>(stream: ReadableStreamEvents<T>, reducer?: IReducer<T, R>): Promise<R | undefined> {
539
>
const chunks: T[] = [];
540
>
541
>
listenStream(stream, {
542
>
onData: chunk => {
545
>
}
552
>
}
559
>
}
562
>
});
563
>
}
565
>
export interface IStreamListener<T> {
566
>
567
>
/**
568
>
* The 'data' event is emitted whenever the stream is
569
>
* relinquishing ownership of a chunk of data to a consumer.
570
>
*/
571
>
onData(data: T): void;
572
>
573
>
/**
574
>
* Emitted when any error occurs.
575
>
*/
576
>
onError(err: Error): void;
577
>
578
>
/**
579
>
* The 'end' event is emitted when there is no more data
580
>
* to be consumed from the stream. The 'end' event will
581
>
* not be emitted unless the data is completely consumed.
582
>
*/
583
>
onEnd(): void;
584
>
}
585
>
586
>
/**
587
>
* Helper to listen to all events of a T stream in proper order.
588
>
*/
589
>
export function listenStream<T>(stream: ReadableStreamEvents<T>, listener: IStreamListener<T>, token?: CancellationToken): void {
591
>
stream.on('error', error => {
594
>
}
596
>
597
>
stream.on('end', () => {
600
>
}
602
>
603
>
// Adding the `data` listener will turn the stream
604
>
// into flowing mode. As such it is important to
605
>
// add this listener last (DO NOT CHANGE!)
606
>
stream.on('data', data => {
609
>
}
611
>
}
613
>
/**
614
>
* Helper to peek up to `maxChunks` into a stream. The return type signals if
615
>
* the stream has ended or not. If not, caller needs to add a `data` listener
616
>
* to continue reading.
617
>
*/
618
>
export function peekStream<T>(stream: ReadableStream<T>, maxChunks: number): Promise<ReadableBufferedStream<T>> {
620
>
const streamListeners = new DisposableStore();
621
>
const buffer: T[] = [];
622
>
623
>
// Data Listener
624
>
const dataListener = (chunk: T) => {
626
>
// Add to buffer
627
>
buffer.push(chunk);
628
>
629
>
// We reached maxChunks and thus need to return
630
>
if (buffer.length > maxChunks) {
632
>
// Dispose any listeners and ensure to pause the
633
>
// stream so that it can be consumed again by caller
634
>
streamListeners.dispose();
635
>
stream.pause();
636
>
637
>
return resolve({ stream, buffer, ended: false });
638
>
}
641
>
// Error Listener
642
>
const errorListener = (error: Error) => {
644
>
645
>
return reject(error);
646
>
};
648
>
// End Listener
649
>
const endListener = () => {
651
>
652
>
return resolve({ stream, buffer, ended: true });
653
>
};
655
>
streamListeners.add(toDisposable(() => stream.removeListener('error', errorListener)));
656
>
stream.on('error', errorListener);
657
>
658
>
streamListeners.add(toDisposable(() => stream.removeListener('end', endListener)));
659
>
stream.on('end', endListener);
660
>
661
>
// Important: leave the `data` listener last because
662
>
// this can turn the stream into flowing mode and we
663
>
// want `error` events to be received as well.
664
>
streamListeners.add(toDisposable(() => stream.removeListener('data', dataListener)));
665
>
stream.on('data', dataListener);
666
>
});
667
>
}
669
>
/**
670
>
* Helper to create a readable stream from an existing T.
671
>
*/
672
>
export function toStream<T>(t: T, reducer: IReducer<T>): ReadableStream<T> {
674
>
675
>
stream.end(t);
676
>
677
>
return stream;
678
>
}
680
>
/**
681
>
* Helper to create an empty stream
682
>
*/
683
>
export function emptyStream(): ReadableStream<never> {
684
const stream = newWriteableStream<never>(() => { throw new Error('not supported'); });
685
stream.end();
686
687
return stream;
688
}
690
>
/**
691
>
* Helper to convert a T into a Readable<T>.
692
>
*/
693
>
export function toReadable<T>(t: T): Readable<T> {
695
>
696
>
return {
697
>
read: () => {
699
>
return null;
700
>
}
701
>
702
>
consumed = true;
703
>
704
>
return t;
705
>
}
707
>
}
709
>
/**
710
>
* Helper to transform a readable stream into another stream.
711
>
*/
712
>
export function transform<Original, Transformed>(stream: ReadableStreamEvents<Original>, transformer: ITransformer<Original, Transformed>, reducer: IReducer<Transformed>): ReadableStream<Transformed> {
714
>
715
>
listenStream(stream, {
716
>
onData: data => target.write(transformer.data(data)),
717
>
onError: error => target.error(transformer.error ? transformer.error(error) : error),
718
>
onEnd: () => target.end()
719
>
});
720
>
721
>
return target;
722
>
}
724
>
/**
725
>
* Helper to take an existing readable that will
726
>
* have a prefix injected to the beginning.
727
>
*/
728
>
export function prefixedReadable<T>(prefix: T, readable: Readable<T>, reducer: IReducer<T>): Readable<T> {
730
>
731
>
return {
732
>
read: () => {
733
>
const chunk = readable.read();
734
>
735
>
// Handle prefix only once
736
>
if (!prefixHandled) {
737
>
prefixHandled = true;
738
>
739
>
// If we have also a read-result, make
740
>
// sure to reduce it to a single result
741
>
if (chunk !== null) {
742
>
return reducer([prefix, chunk]);
743
>
}
744
>
745
>
// Otherwise, just return prefix directly
746
>
return prefix;
747
>
}
748
>
749
>
return chunk;
750
>
}
751
>
};
752
>
}
754
>
/**
755
>
* Helper to take an existing stream that will
756
>
* have a prefix injected to the beginning.
757
>
*/
758
>
export function prefixedStream<T>(prefix: T, stream: ReadableStream<T>, reducer: IReducer<T>): ReadableStream<T> {
760
>
761
>
const target = newWriteableStream<T>(reducer);
762
>
763
>
listenStream(stream, {
764
>
onData: data => {
765
>
766
>
// Handle prefix only once
767
>
if (!prefixHandled) {
768
>
prefixHandled = true;
769
>
770
>
return target.write(reducer([prefix, data]));
771
>
}
772
773
return target.write(data);
775
>
onError: error => target.error(error),
776
>
onEnd: () => {
777
>
778
>
// Handle prefix only once
779
>
if (!prefixHandled) {
780
>
prefixHandled = true;
781
>
782
>
target.write(prefix);
783
>
}
784
>
785
>
target.end();
786
>
}
787
>
});
788
>
789
>
return target;
790
>
}