226 lines
8.0 KiB
JavaScript
226 lines
8.0 KiB
JavaScript
|
/* --------------------------------------------------------------------------------------------
|
||
|
|
* Copyright (c) Microsoft Corporation. All rights reserved.
|
||
|
|
* Licensed under the MIT License. See License.txt in the project root for license information.
|
||
|
|
* ------------------------------------------------------------------------------------------ */
|
||
|
|
'use strict';
|
||
|
|
Object.defineProperty(exports, "__esModule", { value: true });
|
||
|
|
const events_1 = require("./events");
|
||
|
|
const Is = require("./is");
|
||
|
|
let DefaultSize = 8192;
|
||
|
|
let CR = Buffer.from('\r', 'ascii')[0];
|
||
|
|
let LF = Buffer.from('\n', 'ascii')[0];
|
||
|
|
let CRLF = '\r\n';
|
||
|
|
class MessageBuffer {
|
||
|
|
constructor(encoding = 'utf8') {
|
||
|
|
this.encoding = encoding;
|
||
|
|
this.index = 0;
|
||
|
|
this.buffer = Buffer.allocUnsafe(DefaultSize);
|
||
|
|
}
|
||
|
|
append(chunk) {
|
||
|
|
var toAppend = chunk;
|
||
|
|
if (typeof (chunk) === 'string') {
|
||
|
|
var str = chunk;
|
||
|
|
var bufferLen = Buffer.byteLength(str, this.encoding);
|
||
|
|
toAppend = Buffer.allocUnsafe(bufferLen);
|
||
|
|
toAppend.write(str, 0, bufferLen, this.encoding);
|
||
|
|
}
|
||
|
|
if (this.buffer.length - this.index >= toAppend.length) {
|
||
|
|
toAppend.copy(this.buffer, this.index, 0, toAppend.length);
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
var newSize = (Math.ceil((this.index + toAppend.length) / DefaultSize) + 1) * DefaultSize;
|
||
|
|
if (this.index === 0) {
|
||
|
|
this.buffer = Buffer.allocUnsafe(newSize);
|
||
|
|
toAppend.copy(this.buffer, 0, 0, toAppend.length);
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
this.buffer = Buffer.concat([this.buffer.slice(0, this.index), toAppend], newSize);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
this.index += toAppend.length;
|
||
|
|
}
|
||
|
|
tryReadHeaders() {
|
||
|
|
let result = undefined;
|
||
|
|
let current = 0;
|
||
|
|
while (current + 3 < this.index && (this.buffer[current] !== CR || this.buffer[current + 1] !== LF || this.buffer[current + 2] !== CR || this.buffer[current + 3] !== LF)) {
|
||
|
|
current++;
|
||
|
|
}
|
||
|
|
// No header / body separator found (e.g CRLFCRLF)
|
||
|
|
if (current + 3 >= this.index) {
|
||
|
|
return result;
|
||
|
|
}
|
||
|
|
result = Object.create(null);
|
||
|
|
let headers = this.buffer.toString('ascii', 0, current).split(CRLF);
|
||
|
|
headers.forEach((header) => {
|
||
|
|
let index = header.indexOf(':');
|
||
|
|
if (index === -1) {
|
||
|
|
throw new Error('Message header must separate key and value using :');
|
||
|
|
}
|
||
|
|
let key = header.substr(0, index);
|
||
|
|
let value = header.substr(index + 1).trim();
|
||
|
|
result[key] = value;
|
||
|
|
});
|
||
|
|
let nextStart = current + 4;
|
||
|
|
this.buffer = this.buffer.slice(nextStart);
|
||
|
|
this.index = this.index - nextStart;
|
||
|
|
return result;
|
||
|
|
}
|
||
|
|
tryReadContent(length) {
|
||
|
|
if (this.index < length) {
|
||
|
|
return null;
|
||
|
|
}
|
||
|
|
let result = this.buffer.toString(this.encoding, 0, length);
|
||
|
|
let nextStart = length;
|
||
|
|
this.buffer.copy(this.buffer, 0, nextStart);
|
||
|
|
this.index = this.index - nextStart;
|
||
|
|
return result;
|
||
|
|
}
|
||
|
|
get numberOfBytes() {
|
||
|
|
return this.index;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
var MessageReader;
|
||
|
|
(function (MessageReader) {
|
||
|
|
function is(value) {
|
||
|
|
let candidate = value;
|
||
|
|
return candidate && Is.func(candidate.listen) && Is.func(candidate.dispose) &&
|
||
|
|
Is.func(candidate.onError) && Is.func(candidate.onClose) && Is.func(candidate.onPartialMessage);
|
||
|
|
}
|
||
|
|
MessageReader.is = is;
|
||
|
|
})(MessageReader = exports.MessageReader || (exports.MessageReader = {}));
|
||
|
|
class AbstractMessageReader {
|
||
|
|
constructor() {
|
||
|
|
this.errorEmitter = new events_1.Emitter();
|
||
|
|
this.closeEmitter = new events_1.Emitter();
|
||
|
|
this.partialMessageEmitter = new events_1.Emitter();
|
||
|
|
}
|
||
|
|
dispose() {
|
||
|
|
this.errorEmitter.dispose();
|
||
|
|
this.closeEmitter.dispose();
|
||
|
|
}
|
||
|
|
get onError() {
|
||
|
|
return this.errorEmitter.event;
|
||
|
|
}
|
||
|
|
fireError(error) {
|
||
|
|
this.errorEmitter.fire(this.asError(error));
|
||
|
|
}
|
||
|
|
get onClose() {
|
||
|
|
return this.closeEmitter.event;
|
||
|
|
}
|
||
|
|
fireClose() {
|
||
|
|
this.closeEmitter.fire(undefined);
|
||
|
|
}
|
||
|
|
get onPartialMessage() {
|
||
|
|
return this.partialMessageEmitter.event;
|
||
|
|
}
|
||
|
|
firePartialMessage(info) {
|
||
|
|
this.partialMessageEmitter.fire(info);
|
||
|
|
}
|
||
|
|
asError(error) {
|
||
|
|
if (error instanceof Error) {
|
||
|
|
return error;
|
||
|
|
}
|
||
|
|
else {
|
||
|
|
return new Error(`Reader recevied error. Reason: ${Is.string(error.message) ? error.message : 'unknown'}`);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
exports.AbstractMessageReader = AbstractMessageReader;
|
||
|
|
class StreamMessageReader extends AbstractMessageReader {
|
||
|
|
constructor(readable, encoding = 'utf8') {
|
||
|
|
super();
|
||
|
|
this.readable = readable;
|
||
|
|
this.buffer = new MessageBuffer(encoding);
|
||
|
|
this._partialMessageTimeout = 10000;
|
||
|
|
}
|
||
|
|
set partialMessageTimeout(timeout) {
|
||
|
|
this._partialMessageTimeout = timeout;
|
||
|
|
}
|
||
|
|
get partialMessageTimeout() {
|
||
|
|
return this._partialMessageTimeout;
|
||
|
|
}
|
||
|
|
listen(callback) {
|
||
|
|
this.nextMessageLength = -1;
|
||
|
|
this.messageToken = 0;
|
||
|
|
this.partialMessageTimer = undefined;
|
||
|
|
this.callback = callback;
|
||
|
|
this.readable.on('data', (data) => {
|
||
|
|
this.onData(data);
|
||
|
|
});
|
||
|
|
this.readable.on('error', (error) => this.fireError(error));
|
||
|
|
this.readable.on('close', () => this.fireClose());
|
||
|
|
}
|
||
|
|
onData(data) {
|
||
|
|
this.buffer.append(data);
|
||
|
|
while (true) {
|
||
|
|
if (this.nextMessageLength === -1) {
|
||
|
|
let headers = this.buffer.tryReadHeaders();
|
||
|
|
if (!headers) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
let contentLength = headers['Content-Length'];
|
||
|
|
if (!contentLength) {
|
||
|
|
throw new Error('Header must provide a Content-Length property.');
|
||
|
|
}
|
||
|
|
let length = parseInt(contentLength);
|
||
|
|
if (isNaN(length)) {
|
||
|
|
throw new Error('Content-Length value must be a number.');
|
||
|
|
}
|
||
|
|
this.nextMessageLength = length;
|
||
|
|
// Take the encoding form the header. For compatibility
|
||
|
|
// treat both utf-8 and utf8 as node utf8
|
||
|
|
}
|
||
|
|
var msg = this.buffer.tryReadContent(this.nextMessageLength);
|
||
|
|
if (msg === null) {
|
||
|
|
/** We haven't recevied the full message yet. */
|
||
|
|
this.setPartialMessageTimer();
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
this.clearPartialMessageTimer();
|
||
|
|
this.nextMessageLength = -1;
|
||
|
|
this.messageToken++;
|
||
|
|
var json = JSON.parse(msg);
|
||
|
|
this.callback(json);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
clearPartialMessageTimer() {
|
||
|
|
if (this.partialMessageTimer) {
|
||
|
|
clearTimeout(this.partialMessageTimer);
|
||
|
|
this.partialMessageTimer = undefined;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
setPartialMessageTimer() {
|
||
|
|
this.clearPartialMessageTimer();
|
||
|
|
if (this._partialMessageTimeout <= 0) {
|
||
|
|
return;
|
||
|
|
}
|
||
|
|
this.partialMessageTimer = setTimeout((token, timeout) => {
|
||
|
|
this.partialMessageTimer = undefined;
|
||
|
|
if (token === this.messageToken) {
|
||
|
|
this.firePartialMessage({ messageToken: token, waitingTime: timeout });
|
||
|
|
this.setPartialMessageTimer();
|
||
|
|
}
|
||
|
|
}, this._partialMessageTimeout, this.messageToken, this._partialMessageTimeout);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
exports.StreamMessageReader = StreamMessageReader;
|
||
|
|
class IPCMessageReader extends AbstractMessageReader {
|
||
|
|
constructor(process) {
|
||
|
|
super();
|
||
|
|
this.process = process;
|
||
|
|
let eventEmitter = this.process;
|
||
|
|
eventEmitter.on('error', (error) => this.fireError(error));
|
||
|
|
eventEmitter.on('close', () => this.fireClose());
|
||
|
|
}
|
||
|
|
listen(callback) {
|
||
|
|
this.process.on('message', callback);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
exports.IPCMessageReader = IPCMessageReader;
|
||
|
|
class SocketMessageReader extends StreamMessageReader {
|
||
|
|
constructor(socket, encoding = 'utf-8') {
|
||
|
|
super(socket, encoding);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
exports.SocketMessageReader = SocketMessageReader;
|