mirror of
https://github.com/ckaczor/azuredatastudio.git
synced 2026-02-09 09:42:34 -05:00
Merge from vscode 31e03b8ffbb218a87e3941f2b63a249f061fe0e4 (#4986)
This commit is contained in:
Anthony Dresser
821
src/vs/workbench/services/extensions/common/rpcProtocol.ts
Normal file
821
src/vs/workbench/services/extensions/common/rpcProtocol.ts
Normal file
@@ -0,0 +1,821 @@
|
||||
/*---------------------------------------------------------------------------------------------
|
||||
* Copyright (c) Microsoft Corporation. All rights reserved.
|
||||
* Licensed under the Source EULA. See License.txt in the project root for license information.
|
||||
*--------------------------------------------------------------------------------------------*/
|
||||
|
||||
import { RunOnceScheduler } from 'vs/base/common/async';
|
||||
import { CancellationToken, CancellationTokenSource } from 'vs/base/common/cancellation';
|
||||
import { CharCode } from 'vs/base/common/charCode';
|
||||
import * as errors from 'vs/base/common/errors';
|
||||
import { Emitter, Event } from 'vs/base/common/event';
|
||||
import { Disposable } from 'vs/base/common/lifecycle';
|
||||
import { IURITransformer, transformIncomingURIs } from 'vs/base/common/uriIpc';
|
||||
import { IMessagePassingProtocol } from 'vs/base/parts/ipc/common/ipc';
|
||||
import { LazyPromise } from 'vs/workbench/services/extensions/common/lazyPromise';
|
||||
import { IRPCProtocol, ProxyIdentifier, getStringIdentifierForProxy } from 'vs/workbench/services/extensions/common/proxyIdentifier';
|
||||
import { VSBuffer } from 'vs/base/common/buffer';
|
||||
|
||||
export interface JSONStringifyReplacer {
|
||||
(key: string, value: any): any;
|
||||
}
|
||||
|
||||
function safeStringify(obj: any, replacer: JSONStringifyReplacer | null): string {
|
||||
try {
|
||||
return JSON.stringify(obj, <(key: string, value: any) => any>replacer);
|
||||
} catch (err) {
|
||||
return 'null';
|
||||
}
|
||||
}
|
||||
|
||||
function createURIReplacer(transformer: IURITransformer | null): JSONStringifyReplacer | null {
|
||||
if (!transformer) {
|
||||
return null;
|
||||
}
|
||||
return (key: string, value: any): any => {
|
||||
if (value && value.$mid === 1) {
|
||||
return transformer.transformOutgoing(value);
|
||||
}
|
||||
return value;
|
||||
};
|
||||
}
|
||||
|
||||
export const enum RequestInitiator {
|
||||
LocalSide = 0,
|
||||
OtherSide = 1
|
||||
}
|
||||
|
||||
export const enum ResponsiveState {
|
||||
Responsive = 0,
|
||||
Unresponsive = 1
|
||||
}
|
||||
|
||||
export interface IRPCProtocolLogger {
|
||||
logIncoming(msgLength: number, req: number, initiator: RequestInitiator, str: string, data?: any): void;
|
||||
logOutgoing(msgLength: number, req: number, initiator: RequestInitiator, str: string, data?: any): void;
|
||||
}
|
||||
|
||||
const noop = () => { };
|
||||
|
||||
export class RPCProtocol extends Disposable implements IRPCProtocol {
|
||||
|
||||
private static UNRESPONSIVE_TIME = 3 * 1000; // 3s
|
||||
|
||||
private readonly _onDidChangeResponsiveState: Emitter<ResponsiveState> = this._register(new Emitter<ResponsiveState>());
|
||||
public readonly onDidChangeResponsiveState: Event<ResponsiveState> = this._onDidChangeResponsiveState.event;
|
||||
|
||||
private readonly _protocol: IMessagePassingProtocol;
|
||||
private readonly _logger: IRPCProtocolLogger | null;
|
||||
private readonly _uriTransformer: IURITransformer | null;
|
||||
private readonly _uriReplacer: JSONStringifyReplacer | null;
|
||||
private _isDisposed: boolean;
|
||||
private readonly _locals: any[];
|
||||
private readonly _proxies: any[];
|
||||
private _lastMessageId: number;
|
||||
private readonly _cancelInvokedHandlers: { [req: string]: () => void; };
|
||||
private readonly _pendingRPCReplies: { [msgId: string]: LazyPromise; };
|
||||
private _responsiveState: ResponsiveState;
|
||||
private _unacknowledgedCount: number;
|
||||
private _unresponsiveTime: number;
|
||||
private _asyncCheckUresponsive: RunOnceScheduler;
|
||||
|
||||
constructor(protocol: IMessagePassingProtocol, logger: IRPCProtocolLogger | null = null, transformer: IURITransformer | null = null) {
|
||||
super();
|
||||
this._protocol = protocol;
|
||||
this._logger = logger;
|
||||
this._uriTransformer = transformer;
|
||||
this._uriReplacer = createURIReplacer(this._uriTransformer);
|
||||
this._isDisposed = false;
|
||||
this._locals = [];
|
||||
this._proxies = [];
|
||||
for (let i = 0, len = ProxyIdentifier.count; i < len; i++) {
|
||||
this._locals[i] = null;
|
||||
this._proxies[i] = null;
|
||||
}
|
||||
this._lastMessageId = 0;
|
||||
this._cancelInvokedHandlers = Object.create(null);
|
||||
this._pendingRPCReplies = {};
|
||||
this._responsiveState = ResponsiveState.Responsive;
|
||||
this._unacknowledgedCount = 0;
|
||||
this._unresponsiveTime = 0;
|
||||
this._asyncCheckUresponsive = this._register(new RunOnceScheduler(() => this._checkUnresponsive(), 1000));
|
||||
this._protocol.onMessage((msg) => this._receiveOneMessage(msg));
|
||||
}
|
||||
|
||||
public dispose(): void {
|
||||
this._isDisposed = true;
|
||||
|
||||
// Release all outstanding promises with a canceled error
|
||||
Object.keys(this._pendingRPCReplies).forEach((msgId) => {
|
||||
const pending = this._pendingRPCReplies[msgId];
|
||||
pending.resolveErr(errors.canceled());
|
||||
});
|
||||
}
|
||||
|
||||
private _onWillSendRequest(req: number): void {
|
||||
if (this._unacknowledgedCount === 0) {
|
||||
// Since this is the first request we are sending in a while,
|
||||
// mark this moment as the start for the countdown to unresponsive time
|
||||
this._unresponsiveTime = Date.now() + RPCProtocol.UNRESPONSIVE_TIME;
|
||||
}
|
||||
this._unacknowledgedCount++;
|
||||
if (!this._asyncCheckUresponsive.isScheduled()) {
|
||||
this._asyncCheckUresponsive.schedule();
|
||||
}
|
||||
}
|
||||
|
||||
private _onDidReceiveAcknowledge(req: number): void {
|
||||
// The next possible unresponsive time is now + delta.
|
||||
this._unresponsiveTime = Date.now() + RPCProtocol.UNRESPONSIVE_TIME;
|
||||
this._unacknowledgedCount--;
|
||||
if (this._unacknowledgedCount === 0) {
|
||||
// No more need to check for unresponsive
|
||||
this._asyncCheckUresponsive.cancel();
|
||||
}
|
||||
// The ext host is responsive!
|
||||
this._setResponsiveState(ResponsiveState.Responsive);
|
||||
}
|
||||
|
||||
private _checkUnresponsive(): void {
|
||||
if (this._unacknowledgedCount === 0) {
|
||||
// Not waiting for anything => cannot say if it is responsive or not
|
||||
return;
|
||||
}
|
||||
|
||||
if (Date.now() > this._unresponsiveTime) {
|
||||
// Unresponsive!!
|
||||
this._setResponsiveState(ResponsiveState.Unresponsive);
|
||||
} else {
|
||||
// Not (yet) unresponsive, be sure to check again soon
|
||||
this._asyncCheckUresponsive.schedule();
|
||||
}
|
||||
}
|
||||
|
||||
private _setResponsiveState(newResponsiveState: ResponsiveState): void {
|
||||
if (this._responsiveState === newResponsiveState) {
|
||||
// no change
|
||||
return;
|
||||
}
|
||||
this._responsiveState = newResponsiveState;
|
||||
this._onDidChangeResponsiveState.fire(this._responsiveState);
|
||||
}
|
||||
|
||||
public get responsiveState(): ResponsiveState {
|
||||
return this._responsiveState;
|
||||
}
|
||||
|
||||
public transformIncomingURIs<T>(obj: T): T {
|
||||
if (!this._uriTransformer) {
|
||||
return obj;
|
||||
}
|
||||
return transformIncomingURIs(obj, this._uriTransformer);
|
||||
}
|
||||
|
||||
public getProxy<T>(identifier: ProxyIdentifier<T>): T {
|
||||
const rpcId = identifier.nid;
|
||||
if (!this._proxies[rpcId]) {
|
||||
this._proxies[rpcId] = this._createProxy(rpcId);
|
||||
}
|
||||
return this._proxies[rpcId];
|
||||
}
|
||||
|
||||
private _createProxy<T>(rpcId: number): T {
|
||||
let handler = {
|
||||
get: (target: any, name: string) => {
|
||||
if (!target[name] && name.charCodeAt(0) === CharCode.DollarSign) {
|
||||
target[name] = (...myArgs: any[]) => {
|
||||
return this._remoteCall(rpcId, name, myArgs);
|
||||
};
|
||||
}
|
||||
return target[name];
|
||||
}
|
||||
};
|
||||
return new Proxy(Object.create(null), handler);
|
||||
}
|
||||
|
||||
public set<T, R extends T>(identifier: ProxyIdentifier<T>, value: R): R {
|
||||
this._locals[identifier.nid] = value;
|
||||
return value;
|
||||
}
|
||||
|
||||
public assertRegistered(identifiers: ProxyIdentifier<any>[]): void {
|
||||
for (let i = 0, len = identifiers.length; i < len; i++) {
|
||||
const identifier = identifiers[i];
|
||||
if (!this._locals[identifier.nid]) {
|
||||
throw new Error(`Missing actor ${identifier.sid} (isMain: ${identifier.isMain})`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private _receiveOneMessage(rawmsg: VSBuffer): void {
|
||||
if (this._isDisposed) {
|
||||
return;
|
||||
}
|
||||
|
||||
const msgLength = rawmsg.byteLength;
|
||||
const buff = MessageBuffer.read(rawmsg, 0);
|
||||
const messageType = <MessageType>buff.readUInt8();
|
||||
const req = buff.readUInt32();
|
||||
|
||||
switch (messageType) {
|
||||
case MessageType.RequestJSONArgs:
|
||||
case MessageType.RequestJSONArgsWithCancellation: {
|
||||
let { rpcId, method, args } = MessageIO.deserializeRequestJSONArgs(buff);
|
||||
if (this._uriTransformer) {
|
||||
args = transformIncomingURIs(args, this._uriTransformer);
|
||||
}
|
||||
this._receiveRequest(msgLength, req, rpcId, method, args, (messageType === MessageType.RequestJSONArgsWithCancellation));
|
||||
break;
|
||||
}
|
||||
case MessageType.RequestMixedArgs:
|
||||
case MessageType.RequestMixedArgsWithCancellation: {
|
||||
let { rpcId, method, args } = MessageIO.deserializeRequestMixedArgs(buff);
|
||||
if (this._uriTransformer) {
|
||||
args = transformIncomingURIs(args, this._uriTransformer);
|
||||
}
|
||||
this._receiveRequest(msgLength, req, rpcId, method, args, (messageType === MessageType.RequestMixedArgsWithCancellation));
|
||||
break;
|
||||
}
|
||||
case MessageType.Acknowledged: {
|
||||
if (this._logger) {
|
||||
this._logger.logIncoming(msgLength, req, RequestInitiator.LocalSide, `ack`);
|
||||
}
|
||||
this._onDidReceiveAcknowledge(req);
|
||||
break;
|
||||
}
|
||||
case MessageType.Cancel: {
|
||||
this._receiveCancel(msgLength, req);
|
||||
break;
|
||||
}
|
||||
case MessageType.ReplyOKEmpty: {
|
||||
this._receiveReply(msgLength, req, undefined);
|
||||
break;
|
||||
}
|
||||
case MessageType.ReplyOKJSON: {
|
||||
let value = MessageIO.deserializeReplyOKJSON(buff);
|
||||
if (this._uriTransformer) {
|
||||
value = transformIncomingURIs(value, this._uriTransformer);
|
||||
}
|
||||
this._receiveReply(msgLength, req, value);
|
||||
break;
|
||||
}
|
||||
case MessageType.ReplyOKBuffer: {
|
||||
let value = MessageIO.deserializeReplyOKBuffer(buff);
|
||||
this._receiveReply(msgLength, req, value);
|
||||
break;
|
||||
}
|
||||
case MessageType.ReplyOKVSBuffer: {
|
||||
let value = MessageIO.deserializeReplyOKVSBuffer(buff);
|
||||
this._receiveReply(msgLength, req, value);
|
||||
break;
|
||||
}
|
||||
case MessageType.ReplyErrError: {
|
||||
let err = MessageIO.deserializeReplyErrError(buff);
|
||||
if (this._uriTransformer) {
|
||||
err = transformIncomingURIs(err, this._uriTransformer);
|
||||
}
|
||||
this._receiveReplyErr(msgLength, req, err);
|
||||
break;
|
||||
}
|
||||
case MessageType.ReplyErrEmpty: {
|
||||
this._receiveReplyErr(msgLength, req, undefined);
|
||||
break;
|
||||
}
|
||||
default:
|
||||
console.error(`received unexpected message`);
|
||||
console.error(rawmsg);
|
||||
}
|
||||
}
|
||||
|
||||
private _receiveRequest(msgLength: number, req: number, rpcId: number, method: string, args: any[], usesCancellationToken: boolean): void {
|
||||
if (this._logger) {
|
||||
this._logger.logIncoming(msgLength, req, RequestInitiator.OtherSide, `receiveRequest ${getStringIdentifierForProxy(rpcId)}.${method}(`, args);
|
||||
}
|
||||
const callId = String(req);
|
||||
|
||||
let promise: Promise<any>;
|
||||
let cancel: () => void;
|
||||
if (usesCancellationToken) {
|
||||
const cancellationTokenSource = new CancellationTokenSource();
|
||||
args.push(cancellationTokenSource.token);
|
||||
promise = this._invokeHandler(rpcId, method, args);
|
||||
cancel = () => cancellationTokenSource.cancel();
|
||||
} else {
|
||||
// cannot be cancelled
|
||||
promise = this._invokeHandler(rpcId, method, args);
|
||||
cancel = noop;
|
||||
}
|
||||
|
||||
this._cancelInvokedHandlers[callId] = cancel;
|
||||
|
||||
// Acknowledge the request
|
||||
const msg = MessageIO.serializeAcknowledged(req);
|
||||
if (this._logger) {
|
||||
this._logger.logOutgoing(msg.byteLength, req, RequestInitiator.OtherSide, `ack`);
|
||||
}
|
||||
this._protocol.send(msg);
|
||||
|
||||
promise.then((r) => {
|
||||
delete this._cancelInvokedHandlers[callId];
|
||||
const msg = MessageIO.serializeReplyOK(req, r, this._uriReplacer);
|
||||
if (this._logger) {
|
||||
this._logger.logOutgoing(msg.byteLength, req, RequestInitiator.OtherSide, `reply:`, r);
|
||||
}
|
||||
this._protocol.send(msg);
|
||||
}, (err) => {
|
||||
delete this._cancelInvokedHandlers[callId];
|
||||
const msg = MessageIO.serializeReplyErr(req, err);
|
||||
if (this._logger) {
|
||||
this._logger.logOutgoing(msg.byteLength, req, RequestInitiator.OtherSide, `replyErr:`, err);
|
||||
}
|
||||
this._protocol.send(msg);
|
||||
});
|
||||
}
|
||||
|
||||
private _receiveCancel(msgLength: number, req: number): void {
|
||||
if (this._logger) {
|
||||
this._logger.logIncoming(msgLength, req, RequestInitiator.OtherSide, `receiveCancel`);
|
||||
}
|
||||
const callId = String(req);
|
||||
if (this._cancelInvokedHandlers[callId]) {
|
||||
this._cancelInvokedHandlers[callId]();
|
||||
}
|
||||
}
|
||||
|
||||
private _receiveReply(msgLength: number, req: number, value: any): void {
|
||||
if (this._logger) {
|
||||
this._logger.logIncoming(msgLength, req, RequestInitiator.LocalSide, `receiveReply:`, value);
|
||||
}
|
||||
const callId = String(req);
|
||||
if (!this._pendingRPCReplies.hasOwnProperty(callId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
const pendingReply = this._pendingRPCReplies[callId];
|
||||
delete this._pendingRPCReplies[callId];
|
||||
|
||||
pendingReply.resolveOk(value);
|
||||
}
|
||||
|
||||
private _receiveReplyErr(msgLength: number, req: number, value: any): void {
|
||||
if (this._logger) {
|
||||
this._logger.logIncoming(msgLength, req, RequestInitiator.LocalSide, `receiveReplyErr:`, value);
|
||||
}
|
||||
|
||||
const callId = String(req);
|
||||
if (!this._pendingRPCReplies.hasOwnProperty(callId)) {
|
||||
return;
|
||||
}
|
||||
|
||||
const pendingReply = this._pendingRPCReplies[callId];
|
||||
delete this._pendingRPCReplies[callId];
|
||||
|
||||
let err: Error | null = null;
|
||||
if (value && value.$isError) {
|
||||
err = new Error();
|
||||
err.name = value.name;
|
||||
err.message = value.message;
|
||||
err.stack = value.stack;
|
||||
}
|
||||
pendingReply.resolveErr(err);
|
||||
}
|
||||
|
||||
private _invokeHandler(rpcId: number, methodName: string, args: any[]): Promise<any> {
|
||||
try {
|
||||
return Promise.resolve(this._doInvokeHandler(rpcId, methodName, args));
|
||||
} catch (err) {
|
||||
return Promise.reject(err);
|
||||
}
|
||||
}
|
||||
|
||||
private _doInvokeHandler(rpcId: number, methodName: string, args: any[]): any {
|
||||
const actor = this._locals[rpcId];
|
||||
if (!actor) {
|
||||
throw new Error('Unknown actor ' + getStringIdentifierForProxy(rpcId));
|
||||
}
|
||||
let method = actor[methodName];
|
||||
if (typeof method !== 'function') {
|
||||
throw new Error('Unknown method ' + methodName + ' on actor ' + getStringIdentifierForProxy(rpcId));
|
||||
}
|
||||
return method.apply(actor, args);
|
||||
}
|
||||
|
||||
private _remoteCall(rpcId: number, methodName: string, args: any[]): Promise<any> {
|
||||
if (this._isDisposed) {
|
||||
return Promise.reject<any>(errors.canceled());
|
||||
}
|
||||
let cancellationToken: CancellationToken | null = null;
|
||||
if (args.length > 0 && CancellationToken.isCancellationToken(args[args.length - 1])) {
|
||||
cancellationToken = args.pop();
|
||||
}
|
||||
|
||||
if (cancellationToken && cancellationToken.isCancellationRequested) {
|
||||
// No need to do anything...
|
||||
return Promise.reject<any>(errors.canceled());
|
||||
}
|
||||
|
||||
const req = ++this._lastMessageId;
|
||||
const callId = String(req);
|
||||
const result = new LazyPromise();
|
||||
|
||||
if (cancellationToken) {
|
||||
cancellationToken.onCancellationRequested(() => {
|
||||
const msg = MessageIO.serializeCancel(req);
|
||||
if (this._logger) {
|
||||
this._logger.logOutgoing(msg.byteLength, req, RequestInitiator.LocalSide, `cancel`);
|
||||
}
|
||||
this._protocol.send(MessageIO.serializeCancel(req));
|
||||
});
|
||||
}
|
||||
|
||||
this._pendingRPCReplies[callId] = result;
|
||||
this._onWillSendRequest(req);
|
||||
const msg = MessageIO.serializeRequest(req, rpcId, methodName, args, !!cancellationToken, this._uriReplacer);
|
||||
if (this._logger) {
|
||||
this._logger.logOutgoing(msg.byteLength, req, RequestInitiator.LocalSide, `request: ${getStringIdentifierForProxy(rpcId)}.${methodName}(`, args);
|
||||
}
|
||||
this._protocol.send(msg);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
class MessageBuffer {
|
||||
|
||||
public static alloc(type: MessageType, req: number, messageSize: number): MessageBuffer {
|
||||
let result = new MessageBuffer(VSBuffer.alloc(messageSize + 1 /* type */ + 4 /* req */), 0);
|
||||
result.writeUInt8(type);
|
||||
result.writeUInt32(req);
|
||||
return result;
|
||||
}
|
||||
|
||||
public static read(buff: VSBuffer, offset: number): MessageBuffer {
|
||||
return new MessageBuffer(buff, offset);
|
||||
}
|
||||
|
||||
private _buff: VSBuffer;
|
||||
private _offset: number;
|
||||
|
||||
public get buffer(): VSBuffer {
|
||||
return this._buff;
|
||||
}
|
||||
|
||||
private constructor(buff: VSBuffer, offset: number) {
|
||||
this._buff = buff;
|
||||
this._offset = offset;
|
||||
}
|
||||
|
||||
public static sizeUInt8(): number {
|
||||
return 1;
|
||||
}
|
||||
|
||||
public writeUInt8(n: number): void {
|
||||
this._buff.writeUint8(n, this._offset); this._offset += 1;
|
||||
}
|
||||
|
||||
public readUInt8(): number {
|
||||
const n = this._buff.readUint8(this._offset); this._offset += 1;
|
||||
return n;
|
||||
}
|
||||
|
||||
public writeUInt32(n: number): void {
|
||||
this._buff.writeUint32BE(n, this._offset); this._offset += 4;
|
||||
}
|
||||
|
||||
public readUInt32(): number {
|
||||
const n = this._buff.readUint32BE(this._offset); this._offset += 4;
|
||||
return n;
|
||||
}
|
||||
|
||||
public static sizeShortString(str: VSBuffer): number {
|
||||
return 1 /* string length */ + str.byteLength /* actual string */;
|
||||
}
|
||||
|
||||
public writeShortString(str: VSBuffer): void {
|
||||
this._buff.writeUint8(str.byteLength, this._offset); this._offset += 1;
|
||||
this._buff.set(str, this._offset); this._offset += str.byteLength;
|
||||
}
|
||||
|
||||
public readShortString(): string {
|
||||
const strByteLength = this._buff.readUint8(this._offset); this._offset += 1;
|
||||
const strBuff = this._buff.slice(this._offset, this._offset + strByteLength);
|
||||
const str = strBuff.toString(); this._offset += strByteLength;
|
||||
return str;
|
||||
}
|
||||
|
||||
public static sizeLongString(str: VSBuffer): number {
|
||||
return 4 /* string length */ + str.byteLength /* actual string */;
|
||||
}
|
||||
|
||||
public writeLongString(str: VSBuffer): void {
|
||||
this._buff.writeUint32BE(str.byteLength, this._offset); this._offset += 4;
|
||||
this._buff.set(str, this._offset); this._offset += str.byteLength;
|
||||
}
|
||||
|
||||
public readLongString(): string {
|
||||
const strByteLength = this._buff.readUint32BE(this._offset); this._offset += 4;
|
||||
const strBuff = this._buff.slice(this._offset, this._offset + strByteLength);
|
||||
const str = strBuff.toString(); this._offset += strByteLength;
|
||||
return str;
|
||||
}
|
||||
|
||||
public static sizeBuffer(buff: VSBuffer): number {
|
||||
return 4 /* buffer length */ + buff.byteLength /* actual buffer */;
|
||||
}
|
||||
|
||||
public writeBuffer(buff: VSBuffer): void {
|
||||
this._buff.writeUint32BE(buff.byteLength, this._offset); this._offset += 4;
|
||||
this._buff.set(buff, this._offset); this._offset += buff.byteLength;
|
||||
}
|
||||
|
||||
public readBuffer(): Buffer {
|
||||
const buffLength = this._buff.readUint32BE(this._offset); this._offset += 4;
|
||||
const buff = this._buff.slice(this._offset, this._offset + buffLength); this._offset += buffLength;
|
||||
return <Buffer>buff.buffer;
|
||||
}
|
||||
|
||||
public static sizeVSBuffer(buff: VSBuffer): number {
|
||||
return 4 /* buffer length */ + buff.byteLength /* actual buffer */;
|
||||
}
|
||||
|
||||
public writeVSBuffer(buff: VSBuffer): void {
|
||||
this._buff.writeUint32BE(buff.byteLength, this._offset); this._offset += 4;
|
||||
this._buff.set(buff, this._offset); this._offset += buff.byteLength;
|
||||
}
|
||||
|
||||
public readVSBuffer(): VSBuffer {
|
||||
const buffLength = this._buff.readUint32BE(this._offset); this._offset += 4;
|
||||
const buff = this._buff.slice(this._offset, this._offset + buffLength); this._offset += buffLength;
|
||||
return buff;
|
||||
}
|
||||
|
||||
public static sizeMixedArray(arr: VSBuffer[], arrType: ArgType[]): number {
|
||||
let size = 0;
|
||||
size += 1; // arr length
|
||||
for (let i = 0, len = arr.length; i < len; i++) {
|
||||
const el = arr[i];
|
||||
const elType = arrType[i];
|
||||
size += 1; // arg type
|
||||
if (elType === ArgType.String) {
|
||||
size += this.sizeLongString(el);
|
||||
} else if (elType === ArgType.Buffer) {
|
||||
size += this.sizeBuffer(el);
|
||||
} else {
|
||||
size += this.sizeVSBuffer(el);
|
||||
}
|
||||
}
|
||||
return size;
|
||||
}
|
||||
|
||||
public writeMixedArray(arr: VSBuffer[], arrType: ArgType[]): void {
|
||||
this._buff.writeUint8(arr.length, this._offset); this._offset += 1;
|
||||
for (let i = 0, len = arr.length; i < len; i++) {
|
||||
const el = arr[i];
|
||||
const elType = arrType[i];
|
||||
if (elType === ArgType.String) {
|
||||
this.writeUInt8(ArgType.String);
|
||||
this.writeLongString(el);
|
||||
} else if (elType === ArgType.Buffer) {
|
||||
this.writeUInt8(ArgType.Buffer);
|
||||
this.writeVSBuffer(el);
|
||||
} else {
|
||||
this.writeUInt8(ArgType.VSBuffer);
|
||||
this.writeVSBuffer(el);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public readMixedArray(): Array<string | Buffer | VSBuffer> {
|
||||
const arrLen = this._buff.readUint8(this._offset); this._offset += 1;
|
||||
let arr: Array<string | Buffer | VSBuffer> = new Array(arrLen);
|
||||
for (let i = 0; i < arrLen; i++) {
|
||||
const argType = <ArgType>this.readUInt8();
|
||||
switch (argType) {
|
||||
case ArgType.String:
|
||||
arr[i] = this.readLongString();
|
||||
break;
|
||||
case ArgType.Buffer:
|
||||
arr[i] = this.readBuffer();
|
||||
break;
|
||||
case ArgType.VSBuffer:
|
||||
arr[i] = this.readVSBuffer();
|
||||
break;
|
||||
}
|
||||
}
|
||||
return arr;
|
||||
}
|
||||
}
|
||||
|
||||
class MessageIO {
|
||||
|
||||
private static _arrayContainsBuffer(arr: any[]): boolean {
|
||||
for (let i = 0, len = arr.length; i < len; i++) {
|
||||
if (Buffer.isBuffer(arr[i])) {
|
||||
return true;
|
||||
}
|
||||
if (arr[i] instanceof VSBuffer) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public static serializeRequest(req: number, rpcId: number, method: string, args: any[], usesCancellationToken: boolean, replacer: JSONStringifyReplacer | null): VSBuffer {
|
||||
if (this._arrayContainsBuffer(args)) {
|
||||
let massagedArgs: VSBuffer[] = [];
|
||||
let massagedArgsType: ArgType[] = [];
|
||||
for (let i = 0, len = args.length; i < len; i++) {
|
||||
const arg = args[i];
|
||||
if (Buffer.isBuffer(arg)) {
|
||||
massagedArgs[i] = VSBuffer.wrap(arg);
|
||||
massagedArgsType[i] = ArgType.Buffer;
|
||||
} else if (arg instanceof VSBuffer) {
|
||||
massagedArgs[i] = arg;
|
||||
massagedArgsType[i] = ArgType.VSBuffer;
|
||||
} else {
|
||||
massagedArgs[i] = VSBuffer.fromString(safeStringify(arg, replacer));
|
||||
massagedArgsType[i] = ArgType.String;
|
||||
}
|
||||
}
|
||||
return this._requestMixedArgs(req, rpcId, method, massagedArgs, massagedArgsType, usesCancellationToken);
|
||||
}
|
||||
return this._requestJSONArgs(req, rpcId, method, safeStringify(args, replacer), usesCancellationToken);
|
||||
}
|
||||
|
||||
private static _requestJSONArgs(req: number, rpcId: number, method: string, args: string, usesCancellationToken: boolean): VSBuffer {
|
||||
const methodBuff = VSBuffer.fromString(method);
|
||||
const argsBuff = VSBuffer.fromString(args);
|
||||
|
||||
let len = 0;
|
||||
len += MessageBuffer.sizeUInt8();
|
||||
len += MessageBuffer.sizeShortString(methodBuff);
|
||||
len += MessageBuffer.sizeLongString(argsBuff);
|
||||
|
||||
let result = MessageBuffer.alloc(usesCancellationToken ? MessageType.RequestJSONArgsWithCancellation : MessageType.RequestJSONArgs, req, len);
|
||||
result.writeUInt8(rpcId);
|
||||
result.writeShortString(methodBuff);
|
||||
result.writeLongString(argsBuff);
|
||||
return result.buffer;
|
||||
}
|
||||
|
||||
public static deserializeRequestJSONArgs(buff: MessageBuffer): { rpcId: number; method: string; args: any[]; } {
|
||||
const rpcId = buff.readUInt8();
|
||||
const method = buff.readShortString();
|
||||
const args = buff.readLongString();
|
||||
return {
|
||||
rpcId: rpcId,
|
||||
method: method,
|
||||
args: JSON.parse(args)
|
||||
};
|
||||
}
|
||||
|
||||
private static _requestMixedArgs(req: number, rpcId: number, method: string, args: VSBuffer[], argsType: ArgType[], usesCancellationToken: boolean): VSBuffer {
|
||||
const methodBuff = VSBuffer.fromString(method);
|
||||
|
||||
let len = 0;
|
||||
len += MessageBuffer.sizeUInt8();
|
||||
len += MessageBuffer.sizeShortString(methodBuff);
|
||||
len += MessageBuffer.sizeMixedArray(args, argsType);
|
||||
|
||||
let result = MessageBuffer.alloc(usesCancellationToken ? MessageType.RequestMixedArgsWithCancellation : MessageType.RequestMixedArgs, req, len);
|
||||
result.writeUInt8(rpcId);
|
||||
result.writeShortString(methodBuff);
|
||||
result.writeMixedArray(args, argsType);
|
||||
return result.buffer;
|
||||
}
|
||||
|
||||
public static deserializeRequestMixedArgs(buff: MessageBuffer): { rpcId: number; method: string; args: any[]; } {
|
||||
const rpcId = buff.readUInt8();
|
||||
const method = buff.readShortString();
|
||||
const rawargs = buff.readMixedArray();
|
||||
const args: any[] = new Array(rawargs.length);
|
||||
for (let i = 0, len = rawargs.length; i < len; i++) {
|
||||
const rawarg = rawargs[i];
|
||||
if (typeof rawarg === 'string') {
|
||||
args[i] = JSON.parse(rawarg);
|
||||
} else {
|
||||
args[i] = rawarg;
|
||||
}
|
||||
}
|
||||
return {
|
||||
rpcId: rpcId,
|
||||
method: method,
|
||||
args: args
|
||||
};
|
||||
}
|
||||
|
||||
public static serializeAcknowledged(req: number): VSBuffer {
|
||||
return MessageBuffer.alloc(MessageType.Acknowledged, req, 0).buffer;
|
||||
}
|
||||
|
||||
public static serializeCancel(req: number): VSBuffer {
|
||||
return MessageBuffer.alloc(MessageType.Cancel, req, 0).buffer;
|
||||
}
|
||||
|
||||
public static serializeReplyOK(req: number, res: any, replacer: JSONStringifyReplacer | null): VSBuffer {
|
||||
if (typeof res === 'undefined') {
|
||||
return this._serializeReplyOKEmpty(req);
|
||||
}
|
||||
if (Buffer.isBuffer(res)) {
|
||||
return this._serializeReplyOKBuffer(req, res);
|
||||
}
|
||||
if (res instanceof VSBuffer) {
|
||||
return this._serializeReplyOKVSBuffer(req, res);
|
||||
}
|
||||
return this._serializeReplyOKJSON(req, safeStringify(res, replacer));
|
||||
}
|
||||
|
||||
private static _serializeReplyOKEmpty(req: number): VSBuffer {
|
||||
return MessageBuffer.alloc(MessageType.ReplyOKEmpty, req, 0).buffer;
|
||||
}
|
||||
|
||||
private static _serializeReplyOKBuffer(req: number, res: Buffer): VSBuffer {
|
||||
const buff = VSBuffer.wrap(res);
|
||||
|
||||
let len = 0;
|
||||
len += MessageBuffer.sizeBuffer(buff);
|
||||
|
||||
let result = MessageBuffer.alloc(MessageType.ReplyOKBuffer, req, len);
|
||||
result.writeBuffer(buff);
|
||||
return result.buffer;
|
||||
}
|
||||
|
||||
private static _serializeReplyOKVSBuffer(req: number, res: VSBuffer): VSBuffer {
|
||||
let len = 0;
|
||||
len += MessageBuffer.sizeVSBuffer(res);
|
||||
|
||||
let result = MessageBuffer.alloc(MessageType.ReplyOKVSBuffer, req, len);
|
||||
result.writeVSBuffer(res);
|
||||
return result.buffer;
|
||||
}
|
||||
|
||||
public static deserializeReplyOKBuffer(buff: MessageBuffer): Buffer {
|
||||
return buff.readBuffer();
|
||||
}
|
||||
|
||||
public static deserializeReplyOKVSBuffer(buff: MessageBuffer): VSBuffer {
|
||||
return buff.readVSBuffer();
|
||||
}
|
||||
|
||||
private static _serializeReplyOKJSON(req: number, res: string): VSBuffer {
|
||||
const resBuff = VSBuffer.fromString(res);
|
||||
|
||||
let len = 0;
|
||||
len += MessageBuffer.sizeLongString(resBuff);
|
||||
|
||||
let result = MessageBuffer.alloc(MessageType.ReplyOKJSON, req, len);
|
||||
result.writeLongString(resBuff);
|
||||
return result.buffer;
|
||||
}
|
||||
|
||||
public static deserializeReplyOKJSON(buff: MessageBuffer): any {
|
||||
const res = buff.readLongString();
|
||||
return JSON.parse(res);
|
||||
}
|
||||
|
||||
public static serializeReplyErr(req: number, err: any): VSBuffer {
|
||||
if (err instanceof Error) {
|
||||
return this._serializeReplyErrEror(req, err);
|
||||
}
|
||||
return this._serializeReplyErrEmpty(req);
|
||||
}
|
||||
|
||||
private static _serializeReplyErrEror(req: number, _err: Error): VSBuffer {
|
||||
const errBuff = VSBuffer.fromString(safeStringify(errors.transformErrorForSerialization(_err), null));
|
||||
|
||||
let len = 0;
|
||||
len += MessageBuffer.sizeLongString(errBuff);
|
||||
|
||||
let result = MessageBuffer.alloc(MessageType.ReplyErrError, req, len);
|
||||
result.writeLongString(errBuff);
|
||||
return result.buffer;
|
||||
}
|
||||
|
||||
public static deserializeReplyErrError(buff: MessageBuffer): Error {
|
||||
const err = buff.readLongString();
|
||||
return JSON.parse(err);
|
||||
}
|
||||
|
||||
private static _serializeReplyErrEmpty(req: number): VSBuffer {
|
||||
return MessageBuffer.alloc(MessageType.ReplyErrEmpty, req, 0).buffer;
|
||||
}
|
||||
}
|
||||
|
||||
const enum MessageType {
|
||||
RequestJSONArgs = 1,
|
||||
RequestJSONArgsWithCancellation = 2,
|
||||
RequestMixedArgs = 3,
|
||||
RequestMixedArgsWithCancellation = 4,
|
||||
Acknowledged = 5,
|
||||
Cancel = 6,
|
||||
ReplyOKEmpty = 7,
|
||||
ReplyOKBuffer = 8,
|
||||
ReplyOKVSBuffer = 9,
|
||||
ReplyOKJSON = 10,
|
||||
ReplyErrError = 11,
|
||||
ReplyErrEmpty = 12,
|
||||
}
|
||||
|
||||
const enum ArgType {
|
||||
String = 1,
|
||||
Buffer = 2,
|
||||
VSBuffer = 3
|
||||
}
|
||||
Reference in New Issue
Block a user