Merge from vscode 05fc61ffb1aee9fd19173c32113daed079f9b7bd (#5074)

* Merge from vscode 05fc61ffb1aee9fd19173c32113daed079f9b7bd

* fix tests

src/vs/base/common/buffer.ts

@@ -11,7 +11,7 @@ let textDecoder: TextDecoder | null;
 
 export class VSBuffer {
 
-	public static alloc(byteLength: number): VSBuffer {
+	static alloc(byteLength: number): VSBuffer {
 		if (hasBuffer) {
 			return new VSBuffer(Buffer.allocUnsafe(byteLength));
 		} else {
@@ -19,7 +19,7 @@ export class VSBuffer {
 		}
 	}
 
-	public static wrap(actual: Uint8Array): VSBuffer {
+	static wrap(actual: Uint8Array): VSBuffer {
 		if (hasBuffer && !(Buffer.isBuffer(actual))) {
 			// https://nodejs.org/dist/latest-v10.x/docs/api/buffer.html#buffer_class_method_buffer_from_arraybuffer_byteoffset_length
 			// Create a zero-copy Buffer wrapper around the ArrayBuffer pointed to by the Uint8Array
@@ -28,7 +28,7 @@ export class VSBuffer {
 		return new VSBuffer(actual);
 	}
 
-	public static fromString(source: string): VSBuffer {
+	static fromString(source: string): VSBuffer {
 		if (hasBuffer) {
 			return new VSBuffer(Buffer.from(source));
 		} else {
@@ -39,7 +39,7 @@ export class VSBuffer {
 		}
 	}
 
-	public static concat(buffers: VSBuffer[], totalLength?: number): VSBuffer {
+	static concat(buffers: VSBuffer[], totalLength?: number): VSBuffer {
 		if (typeof totalLength === 'undefined') {
 			totalLength = 0;
 			for (let i = 0, len = buffers.length; i < len; i++) {
@@ -58,15 +58,15 @@ export class VSBuffer {
 		return ret;
 	}
 
-	public readonly buffer: Uint8Array;
-	public readonly byteLength: number;
+	readonly buffer: Uint8Array;
+	readonly byteLength: number;
 
 	private constructor(buffer: Uint8Array) {
 		this.buffer = buffer;
 		this.byteLength = this.buffer.byteLength;
 	}
 
-	public toString(): string {
+	toString(): string {
 		if (hasBuffer) {
 			return this.buffer.toString();
 		} else {
@@ -77,33 +77,32 @@ export class VSBuffer {
 		}
 	}
 
-	public slice(start?: number, end?: number): VSBuffer {
+	slice(start?: number, end?: number): VSBuffer {
 		return new VSBuffer(this.buffer.slice(start, end));
 	}
 
-	public set(array: VSBuffer, offset?: number): void {
+	set(array: VSBuffer, offset?: number): void {
 		this.buffer.set(array.buffer, offset);
 	}
 
-	public readUint32BE(offset: number): number {
-		return readUint32BE(this.buffer, offset);
+	readUInt32BE(offset: number): number {
+		return readUInt32BE(this.buffer, offset);
 	}
 
-	public writeUint32BE(value: number, offset: number): void {
-		writeUint32BE(this.buffer, value, offset);
+	writeUInt32BE(value: number, offset: number): void {
+		writeUInt32BE(this.buffer, value, offset);
 	}
 
-	public readUint8(offset: number): number {
-		return readUint8(this.buffer, offset);
+	readUInt8(offset: number): number {
+		return readUInt8(this.buffer, offset);
 	}
 
-	public writeUint8(value: number, offset: number): void {
-		writeUint8(this.buffer, value, offset);
+	writeUInt8(value: number, offset: number): void {
+		writeUInt8(this.buffer, value, offset);
 	}
-
 }
 
-function readUint32BE(source: Uint8Array, offset: number): number {
+function readUInt32BE(source: Uint8Array, offset: number): number {
 	return (
 		source[offset] * 2 ** 24
 		+ source[offset + 1] * 2 ** 16
@@ -112,7 +111,7 @@ function readUint32BE(source: Uint8Array, offset: number): number {
 	);
 }
 
-function writeUint32BE(destination: Uint8Array, value: number, offset: number): void {
+function writeUInt32BE(destination: Uint8Array, value: number, offset: number): void {
 	destination[offset + 3] = value;
 	value = value >>> 8;
 	destination[offset + 2] = value;
@@ -122,11 +121,11 @@ function writeUint32BE(destination: Uint8Array, value: number, offset: number):
 	destination[offset] = value;
 }
 
-function readUint8(source: Uint8Array, offset: number): number {
+function readUInt8(source: Uint8Array, offset: number): number {
 	return source[offset];
 }
 
-function writeUint8(destination: Uint8Array, value: number, offset: number): void {
+function writeUInt8(destination: Uint8Array, value: number, offset: number): void {
 	destination[offset] = value;
 }
 
@@ -139,6 +138,47 @@ export interface VSBufferReadable {
 	read(): VSBuffer | null;
 }
 
+/**
+ * A buffer readable stream emits data to listeners. The stream
+ * will only start emitting when the first data listener has
+ * been added or the resume() method has been called.
+ */
+export interface VSBufferReadableStream {
+
+	/**
+	 * The 'data' event is emitted whenever the stream is
+	 * relinquishing ownership of a chunk of data to a consumer.
+	 */
+	on(event: 'data', callback: (chunk: VSBuffer) => void): void;
+
+	/**
+	 * Emitted when any error occurs.
+	 */
+	on(event: 'error', callback: (err: any) => void): void;
+
+	/**
+	 * The 'end' event is emitted when there is no more data
+	 * to be consumed from the stream. The 'end' event will
+	 * not be emitted unless the data is completely consumed.
+	 */
+	on(event: 'end', callback: () => void): void;
+
+	/**
+	 * Stops emitting any events until resume() is called.
+	 */
+	pause(): void;
+
+	/**
+	 * Starts emitting events again after pause() was called.
+	 */
+	resume(): void;
+
+	/**
+	 * Destroys the stream and stops emitting any event.
+	 */
+	destroy(): void;
+}
+
 /**
  * Helper to fully read a VSBuffer readable into a single buffer.
  */
@@ -158,6 +198,7 @@ export function readableToBuffer(readable: VSBufferReadable): VSBuffer {
  */
 export function bufferToReadable(buffer: VSBuffer): VSBufferReadable {
 	let done = false;
+
 	return {
 		read: () => {
 			if (done) {
@@ -169,4 +210,231 @@ export function bufferToReadable(buffer: VSBuffer): VSBufferReadable {
 			return buffer;
 		}
 	};
+}
+
+/**
+ * Helper to fully read a VSBuffer stream into a single buffer.
+ */
+export function streamToBuffer(stream: VSBufferReadableStream): Promise<VSBuffer> {
+	return new Promise((resolve, reject) => {
+		const chunks: VSBuffer[] = [];
+
+		stream.on('data', chunk => chunks.push(chunk));
+		stream.on('error', error => reject(error));
+		stream.on('end', () => resolve(VSBuffer.concat(chunks)));
+	});
+}
+
+/**
+ * Helper to create a VSBufferStream from an existing VSBuffer.
+ */
+export function bufferToStream(buffer: VSBuffer): VSBufferReadableStream {
+	const stream = writeableBufferStream();
+
+	stream.end(buffer);
+
+	return stream;
+}
+
+/**
+ * Helper to create a VSBufferStream that can be pushed
+ * buffers to. Will only start to emit data when a listener
+ * is added.
+ */
+export function writeableBufferStream(): VSBufferWriteableStream {
+	return new VSBufferWriteableStreamImpl();
+}
+
+export interface VSBufferWriteableStream extends VSBufferReadableStream {
+	write(chunk: VSBuffer): void;
+	error(error: Error): void;
+	end(result?: VSBuffer | Error): void;
+}
+
+class VSBufferWriteableStreamImpl implements VSBufferWriteableStream {
+
+	private readonly state = {
+		flowing: false,
+		ended: false,
+		destroyed: false
+	};
+
+	private readonly buffer = {
+		data: [] as VSBuffer[],
+		error: [] as Error[]
+	};
+
+	private readonly listeners = {
+		data: [] as { (chunk: VSBuffer): void }[],
+		error: [] as { (error: Error): void }[],
+		end: [] as { (): void }[]
+	};
+
+	pause(): void {
+		if (this.state.destroyed) {
+			return;
+		}
+
+		this.state.flowing = false;
+	}
+
+	resume(): void {
+		if (this.state.destroyed) {
+			return;
+		}
+
+		if (!this.state.flowing) {
+			this.state.flowing = true;
+
+			// emit buffered events
+			this.flowData();
+			this.flowErrors();
+			this.flowEnd();
+		}
+	}
+
+	write(chunk: VSBuffer): void {
+		if (this.state.destroyed) {
+			return;
+		}
+
+		// flowing: directly send the data to listeners
+		if (this.state.flowing) {
+			this.listeners.data.forEach(listener => listener(chunk));
+		}
+
+		// not yet flowing: buffer data until flowing
+		else {
+			this.buffer.data.push(chunk);
+		}
+	}
+
+	error(error: Error): void {
+		if (this.state.destroyed) {
+			return;
+		}
+
+		// flowing: directly send the error to listeners
+		if (this.state.flowing) {
+			this.listeners.error.forEach(listener => listener(error));
+		}
+
+		// not yet flowing: buffer errors until flowing
+		else {
+			this.buffer.error.push(error);
+		}
+	}
+
+	end(result?: VSBuffer | Error): void {
+		if (this.state.destroyed) {
+			return;
+		}
+
+		// end with data or error if provided
+		if (result instanceof Error) {
+			this.error(result);
+		} else if (result) {
+			this.write(result);
+		}
+
+		// flowing: send end event to listeners
+		if (this.state.flowing) {
+			this.listeners.end.forEach(listener => listener());
+
+			this.destroy();
+		}
+
+		// not yet flowing: remember state
+		else {
+			this.state.ended = true;
+		}
+	}
+
+	on(event: 'data', callback: (chunk: VSBuffer) => void): void;
+	on(event: 'error', callback: (err: any) => void): void;
+	on(event: 'end', callback: () => void): void;
+	on(event: 'data' | 'error' | 'end', callback: (arg0?: any) => void): void {
+		if (this.state.destroyed) {
+			return;
+		}
+
+		switch (event) {
+			case 'data':
+				this.listeners.data.push(callback);
+
+				// switch into flowing mode as soon as the first 'data'
+				// listener is added and we are not yet in flowing mode
+				this.resume();
+
+				break;
+
+			case 'end':
+				this.listeners.end.push(callback);
+
+				// emit 'end' event directly if we are flowing
+				// and the end has already been reached
+				//
+				// finish() when it went through
+				if (this.state.flowing && this.flowEnd()) {
+					this.destroy();
+				}
+
+				break;
+
+			case 'error':
+				this.listeners.error.push(callback);
+
+				// emit buffered 'error' events unless done already
+				// now that we know that we have at least one listener
+				if (this.state.flowing) {
+					this.flowErrors();
+				}
+
+				break;
+		}
+	}
+
+	private flowData(): void {
+		if (this.buffer.data.length > 0) {
+			const fullDataBuffer = VSBuffer.concat(this.buffer.data);
+
+			this.listeners.data.forEach(listener => listener(fullDataBuffer));
+
+			this.buffer.data.length = 0;
+		}
+	}
+
+	private flowErrors(): void {
+		if (this.listeners.error.length > 0) {
+			for (const error of this.buffer.error) {
+				this.listeners.error.forEach(listener => listener(error));
+			}
+
+			this.buffer.error.length = 0;
+		}
+	}
+
+	private flowEnd(): boolean {
+		if (this.state.ended) {
+			this.listeners.end.forEach(listener => listener());
+
+			return this.listeners.end.length > 0;
+		}
+
+		return false;
+	}
+
+	destroy(): void {
+		if (!this.state.destroyed) {
+			this.state.destroyed = true;
+			this.state.ended = true;
+
+			this.buffer.data.length = 0;
+			this.buffer.error.length = 0;
+
+			this.listeners.data.length = 0;
+			this.listeners.error.length = 0;
+			this.listeners.end.length = 0;
+		}
+	}
 }