From 7aa82de380d6966ac13c4634fb961f2ef4acf2c9 Mon Sep 17 00:00:00 2001
From: Johannes Rieken <johannes.rieken@gmail.com>
Date: Mon, 23 Jan 2017 08:36:54 +0100
Subject: [PATCH] add simple protocol test

---
 src/vs/base/parts/ipc/node/ipc.net.ts         | 25 +++--
 .../base/parts/ipc/test/node/ipc.net.test.ts  | 91 +++++++++++++++++++
 2 files changed, 103 insertions(+), 13 deletions(-)
 create mode 100644 src/vs/base/parts/ipc/test/node/ipc.net.test.ts

diff --git a/src/vs/base/parts/ipc/node/ipc.net.ts b/src/vs/base/parts/ipc/node/ipc.net.ts
index 251d5a7a0c5..6877257ee32 100644
--- a/src/vs/base/parts/ipc/node/ipc.net.ts
+++ b/src/vs/base/parts/ipc/node/ipc.net.ts
@@ -6,6 +6,7 @@
 'use strict';
 
 import { Socket, Server as NetServer, createConnection, createServer } from 'net';
+import { Duplex } from 'stream';
 import { TPromise } from 'vs/base/common/winjs.base';
 import Event, { Emitter, once, mapEvent } from 'vs/base/common/event';
 import { fromEventEmitter } from 'vs/base/node/event';
@@ -31,19 +32,19 @@ function bufferIndexOf(buffer: Buffer, value: number, start = 0) {
 	return start;
 }
 
-class Protocol implements IMessagePassingProtocol {
+export class Protocol implements IMessagePassingProtocol {
 
 	private static Boundary = new Buffer([0]);
 
-	private _onMessage: Event<any>;
-	get onMessage(): Event<any> { return this._onMessage; }
 
-	constructor(private socket: Socket) {
+	private _onMessage = new Emitter<any>();
+
+	readonly onMessage: Event<any> = this._onMessage.event;
+
+	constructor(private stream: Duplex) {
 		let buffer = null;
-		const emitter = new Emitter<any>();
-		const onRawData = fromEventEmitter(socket, 'data', data => data);
 
-		onRawData((data: Buffer) => {
+		stream.on('data', (data: Buffer) => {
 			let lastIndex = 0;
 			let index = 0;
 
@@ -51,10 +52,10 @@ class Protocol implements IMessagePassingProtocol {
 				const dataToParse = data.slice(lastIndex, index);
 
 				if (buffer) {
-					emitter.fire(JSON.parse(Buffer.concat([buffer, dataToParse]).toString('utf8')));
+					this._onMessage.fire(JSON.parse(Buffer.concat([buffer, dataToParse]).toString('utf8')));
 					buffer = null;
 				} else {
-					emitter.fire(JSON.parse(dataToParse.toString('utf8')));
+					this._onMessage.fire(JSON.parse(dataToParse.toString('utf8')));
 				}
 
 				lastIndex = index + 1;
@@ -70,14 +71,12 @@ class Protocol implements IMessagePassingProtocol {
 				}
 			}
 		});
-
-		this._onMessage = emitter.event;
 	}
 
 	public send(message: any): void {
 		try {
-			this.socket.write(JSON.stringify(message));
-			this.socket.write(Protocol.Boundary);
+			this.stream.write(JSON.stringify(message));
+			this.stream.write(Protocol.Boundary);
 		} catch (e) {
 			// noop
 		}
diff --git a/src/vs/base/parts/ipc/test/node/ipc.net.test.ts b/src/vs/base/parts/ipc/test/node/ipc.net.test.ts
new file mode 100644
index 00000000000..e0b797b94f4
--- /dev/null
+++ b/src/vs/base/parts/ipc/test/node/ipc.net.test.ts
@@ -0,0 +1,91 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *  Licensed under the MIT License. See License.txt in the project root for license information.
+ *--------------------------------------------------------------------------------------------*/
+
+'use strict';
+
+import * as assert from 'assert';
+import { TPromise } from 'vs/base/common/winjs.base';
+import { Duplex } from 'stream';
+import { Protocol } from 'vs/base/parts/ipc/node/ipc.net';
+
+
+let _buffer: Buffer[] = [];
+
+const testDuplex = new Duplex(<any>{
+	read(size) {
+		const chunks: Buffer[] = [];
+		for (const chunk of _buffer) {
+			chunks.push(chunk);
+			size -= chunk.length;
+			if (size <= 0) {
+				break;
+			}
+		}
+		_buffer = _buffer.slice(chunks.length);
+		this.push(Buffer.concat(chunks));
+	},
+	write(chunk, encoding, callback) {
+		_buffer.push(chunk);
+		callback();
+	}
+});
+
+class TestDuplex extends Duplex {
+
+	private _buffer: Buffer[] = [];
+
+	constructor(options) {
+		super(options);
+	}
+
+	_write(chunk, encoding, callback) {
+		this._buffer.push(chunk);
+		callback();
+	}
+
+	_read(size) {
+		const chunks: Buffer[] = [];
+		for (const chunk of this._buffer) {
+			chunks.push(chunk);
+			size -= chunk.length;
+			if (size <= 0) {
+				break;
+			}
+		}
+		this._buffer = this._buffer.slice(chunks.length);
+		this.push(Buffer.from(chunks));
+	}
+}
+
+
+suite('IPC, Socket Protocol', () => {
+
+	test('read/write', () => {
+
+		const stream = testDuplex;
+
+		const a = new Protocol(stream);
+		const b = new Protocol(stream);
+
+		a.send('foobarfarboo');
+
+		const p1 = new TPromise(resolve => {
+			b.onMessage(data => {
+				assert.equal(data, 'foobarfarboo');
+				resolve(null);
+			});
+		});
+		a.send('message2');
+
+		const p2 = new TPromise(resolve => {
+			b.onMessage(data => {
+				assert.equal(data, 'message2');
+				resolve(null);
+			});
+		});
+
+		return TPromise.join([p1, p2]);
+	});
+});
-- 
GitLab