Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
xxadev
vscode
提交
acfadb48
V
vscode
项目概览
xxadev
/
vscode
与 Fork 源项目一致
从无法访问的项目Fork
通知
2
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
V
vscode
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
acfadb48
编写于
8月 24, 2018
作者:
A
Alexandru Dima
提交者:
GitHub
8月 24, 2018
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #57144 from Microsoft/alex/binary-rpc
Convert renderer <-> ext host communication to a binary format
上级
6a733c7f
8e35ed1d
变更
7
显示空白变更内容
内联
并排
Showing
7 changed file
with
562 addition
and
134 deletion
+562
-134
src/vs/base/parts/ipc/node/ipc.net.ts
src/vs/base/parts/ipc/node/ipc.net.ts
+1
-1
src/vs/workbench/api/electron-browser/mainThreadFileSystem.ts
...vs/workbench/api/electron-browser/mainThreadFileSystem.ts
+3
-5
src/vs/workbench/api/node/extHost.protocol.ts
src/vs/workbench/api/node/extHost.protocol.ts
+2
-2
src/vs/workbench/api/node/extHostFileSystem.ts
src/vs/workbench/api/node/extHostFileSystem.ts
+4
-4
src/vs/workbench/services/extensions/electron-browser/extensionService.ts
.../services/extensions/electron-browser/extensionService.ts
+15
-14
src/vs/workbench/services/extensions/node/rpcProtocol.ts
src/vs/workbench/services/extensions/node/rpcProtocol.ts
+393
-108
src/vs/workbench/services/extensions/test/node/rpcProtocol.test.ts
...rkbench/services/extensions/test/node/rpcProtocol.test.ts
+144
-0
未找到文件。
src/vs/base/parts/ipc/node/ipc.net.ts
浏览文件 @
acfadb48
...
@@ -164,7 +164,7 @@ export class Protocol implements IDisposable, IMessagePassingProtocol {
...
@@ -164,7 +164,7 @@ export class Protocol implements IDisposable, IMessagePassingProtocol {
}
}
send
(
buffer
:
Buffer
):
void
{
send
(
buffer
:
Buffer
):
void
{
const
header
=
Buffer
.
alloc
(
Protocol
.
_headerLen
);
const
header
=
Buffer
.
alloc
Unsafe
(
Protocol
.
_headerLen
);
header
.
writeUInt32BE
(
buffer
.
length
,
0
,
true
);
header
.
writeUInt32BE
(
buffer
.
length
,
0
,
true
);
this
.
_writeSoon
(
header
,
buffer
);
this
.
_writeSoon
(
header
,
buffer
);
}
}
...
...
src/vs/workbench/api/electron-browser/mainThreadFileSystem.ts
浏览文件 @
acfadb48
...
@@ -99,15 +99,13 @@ class RemoteFileSystemProvider implements IFileSystemProvider {
...
@@ -99,15 +99,13 @@ class RemoteFileSystemProvider implements IFileSystemProvider {
}
}
readFile
(
resource
:
URI
):
TPromise
<
Uint8Array
>
{
readFile
(
resource
:
URI
):
TPromise
<
Uint8Array
>
{
return
this
.
_proxy
.
$readFile
(
this
.
_handle
,
resource
).
then
(
encoded
=>
{
return
this
.
_proxy
.
$readFile
(
this
.
_handle
,
resource
);
return
Buffer
.
from
(
encoded
,
'
base64
'
);
});
}
}
writeFile
(
resource
:
URI
,
content
:
Uint8Array
,
opts
:
FileWriteOptions
):
TPromise
<
void
>
{
writeFile
(
resource
:
URI
,
content
:
Uint8Array
,
opts
:
FileWriteOptions
):
TPromise
<
void
>
{
let
encoded
=
Buffer
.
isBuffer
(
content
)
let
encoded
=
Buffer
.
isBuffer
(
content
)
?
content
.
toString
(
'
base64
'
)
?
content
:
Buffer
.
from
(
content
.
buffer
,
content
.
byteOffset
,
content
.
byteLength
)
.
toString
(
'
base64
'
)
;
:
Buffer
.
from
(
content
.
buffer
,
content
.
byteOffset
,
content
.
byteLength
);
return
this
.
_proxy
.
$writeFile
(
this
.
_handle
,
resource
,
encoded
,
opts
);
return
this
.
_proxy
.
$writeFile
(
this
.
_handle
,
resource
,
encoded
,
opts
);
}
}
...
...
src/vs/workbench/api/node/extHost.protocol.ts
浏览文件 @
acfadb48
...
@@ -680,8 +680,8 @@ export interface ExtHostWorkspaceShape {
...
@@ -680,8 +680,8 @@ export interface ExtHostWorkspaceShape {
export
interface
ExtHostFileSystemShape
{
export
interface
ExtHostFileSystemShape
{
$stat
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
IStat
>
;
$stat
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
IStat
>
;
$readdir
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
[
string
,
FileType
][]
>
;
$readdir
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
[
string
,
FileType
][]
>
;
$readFile
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
string
>
;
$readFile
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
Buffer
>
;
$writeFile
(
handle
:
number
,
resource
:
UriComponents
,
base64Encoded
:
string
,
opts
:
FileWriteOptions
):
TPromise
<
void
>
;
$writeFile
(
handle
:
number
,
resource
:
UriComponents
,
content
:
Buffer
,
opts
:
FileWriteOptions
):
TPromise
<
void
>
;
$rename
(
handle
:
number
,
resource
:
UriComponents
,
target
:
UriComponents
,
opts
:
FileOverwriteOptions
):
TPromise
<
void
>
;
$rename
(
handle
:
number
,
resource
:
UriComponents
,
target
:
UriComponents
,
opts
:
FileOverwriteOptions
):
TPromise
<
void
>
;
$copy
(
handle
:
number
,
resource
:
UriComponents
,
target
:
UriComponents
,
opts
:
FileOverwriteOptions
):
TPromise
<
void
>
;
$copy
(
handle
:
number
,
resource
:
UriComponents
,
target
:
UriComponents
,
opts
:
FileOverwriteOptions
):
TPromise
<
void
>
;
$mkdir
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
void
>
;
$mkdir
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
void
>
;
...
...
src/vs/workbench/api/node/extHostFileSystem.ts
浏览文件 @
acfadb48
...
@@ -159,16 +159,16 @@ export class ExtHostFileSystem implements ExtHostFileSystemShape {
...
@@ -159,16 +159,16 @@ export class ExtHostFileSystem implements ExtHostFileSystemShape {
return
asWinJsPromise
(()
=>
this
.
_fsProvider
.
get
(
handle
).
readDirectory
(
URI
.
revive
(
resource
)));
return
asWinJsPromise
(()
=>
this
.
_fsProvider
.
get
(
handle
).
readDirectory
(
URI
.
revive
(
resource
)));
}
}
$readFile
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
string
>
{
$readFile
(
handle
:
number
,
resource
:
UriComponents
):
TPromise
<
Buffer
>
{
return
asWinJsPromise
(()
=>
{
return
asWinJsPromise
(()
=>
{
return
this
.
_fsProvider
.
get
(
handle
).
readFile
(
URI
.
revive
(
resource
));
return
this
.
_fsProvider
.
get
(
handle
).
readFile
(
URI
.
revive
(
resource
));
}).
then
(
data
=>
{
}).
then
(
data
=>
{
return
Buffer
.
isBuffer
(
data
)
?
data
.
toString
(
'
base64
'
)
:
Buffer
.
from
(
data
.
buffer
,
data
.
byteOffset
,
data
.
byteLength
).
toString
(
'
base64
'
);
return
Buffer
.
isBuffer
(
data
)
?
data
:
Buffer
.
from
(
data
.
buffer
,
data
.
byteOffset
,
data
.
byteLength
);
});
});
}
}
$writeFile
(
handle
:
number
,
resource
:
UriComponents
,
base64Content
:
string
,
opts
:
files
.
FileWriteOptions
):
TPromise
<
void
>
{
$writeFile
(
handle
:
number
,
resource
:
UriComponents
,
content
:
Buffer
,
opts
:
files
.
FileWriteOptions
):
TPromise
<
void
>
{
return
asWinJsPromise
(()
=>
this
.
_fsProvider
.
get
(
handle
).
writeFile
(
URI
.
revive
(
resource
),
Buffer
.
from
(
base64Content
,
'
base64
'
)
,
opts
));
return
asWinJsPromise
(()
=>
this
.
_fsProvider
.
get
(
handle
).
writeFile
(
URI
.
revive
(
resource
),
content
,
opts
));
}
}
$delete
(
handle
:
number
,
resource
:
UriComponents
,
opts
:
files
.
FileDeleteOptions
):
TPromise
<
void
>
{
$delete
(
handle
:
number
,
resource
:
UriComponents
,
opts
:
files
.
FileDeleteOptions
):
TPromise
<
void
>
{
...
...
src/vs/workbench/services/extensions/electron-browser/extensionService.ts
浏览文件 @
acfadb48
...
@@ -39,7 +39,7 @@ import { Event, Emitter } from 'vs/base/common/event';
...
@@ -39,7 +39,7 @@ import { Event, Emitter } from 'vs/base/common/event';
import
{
ExtensionHostProfiler
}
from
'
vs/workbench/services/extensions/electron-browser/extensionHostProfiler
'
;
import
{
ExtensionHostProfiler
}
from
'
vs/workbench/services/extensions/electron-browser/extensionHostProfiler
'
;
import
product
from
'
vs/platform/node/product
'
;
import
product
from
'
vs/platform/node/product
'
;
import
*
as
strings
from
'
vs/base/common/strings
'
;
import
*
as
strings
from
'
vs/base/common/strings
'
;
import
{
RPCProtocol
}
from
'
vs/workbench/services/extensions/node/rpcProtocol
'
;
import
{
RPCProtocol
,
IRPCProtocolLogger
}
from
'
vs/workbench/services/extensions/node/rpcProtocol
'
;
import
{
INotificationService
,
Severity
}
from
'
vs/platform/notification/common/notification
'
;
import
{
INotificationService
,
Severity
}
from
'
vs/platform/notification/common/notification
'
;
import
{
isFalsyOrEmpty
}
from
'
vs/base/common/arrays
'
;
import
{
isFalsyOrEmpty
}
from
'
vs/base/common/arrays
'
;
import
{
Schemas
}
from
'
vs/base/common/network
'
;
import
{
Schemas
}
from
'
vs/base/common/network
'
;
...
@@ -180,11 +180,12 @@ export class ExtensionHostProcessManager extends Disposable {
...
@@ -180,11 +180,12 @@ export class ExtensionHostProcessManager extends Disposable {
private
_createExtensionHostCustomers
(
protocol
:
IMessagePassingProtocol
):
ExtHostExtensionServiceShape
{
private
_createExtensionHostCustomers
(
protocol
:
IMessagePassingProtocol
):
ExtHostExtensionServiceShape
{
let
logger
:
IRPCProtocolLogger
=
null
;
if
(
logExtensionHostCommunication
||
this
.
_environmentService
.
logExtensionHostCommunication
)
{
if
(
logExtensionHostCommunication
||
this
.
_environmentService
.
logExtensionHostCommunication
)
{
protocol
=
asLoggingProtocol
(
protocol
);
logger
=
new
RPCLogger
(
);
}
}
this
.
_extensionHostProcessRPCProtocol
=
new
RPCProtocol
(
protocol
);
this
.
_extensionHostProcessRPCProtocol
=
new
RPCProtocol
(
protocol
,
logger
);
const
extHostContext
:
IExtHostContext
=
{
const
extHostContext
:
IExtHostContext
=
{
getProxy
:
<
T
>
(
identifier
:
ProxyIdentifier
<
T
>
):
T
=>
this
.
_extensionHostProcessRPCProtocol
.
getProxy
(
identifier
),
getProxy
:
<
T
>
(
identifier
:
ProxyIdentifier
<
T
>
):
T
=>
this
.
_extensionHostProcessRPCProtocol
.
getProxy
(
identifier
),
set
:
<
T
,
R
extends
T
>
(
identifier
:
ProxyIdentifier
<
T
>
,
instance
:
R
):
R
=>
this
.
_extensionHostProcessRPCProtocol
.
set
(
identifier
,
instance
),
set
:
<
T
,
R
extends
T
>
(
identifier
:
ProxyIdentifier
<
T
>
,
instance
:
R
):
R
=>
this
.
_extensionHostProcessRPCProtocol
.
set
(
identifier
,
instance
),
...
@@ -954,20 +955,20 @@ export class ExtensionService extends Disposable implements IExtensionService {
...
@@ -954,20 +955,20 @@ export class ExtensionService extends Disposable implements IExtensionService {
}
}
}
}
function
asLoggingProtocol
(
protocol
:
IMessagePassingProtocol
):
IMessagePassingProtocol
{
class
RPCLogger
implements
IRPCProtocolLogger
{
protocol
.
onMessage
(
msg
=>
{
private
_totalIncoming
=
0
;
console
.
log
(
'
%c[Extension
\
u2192 Window]%c[len:
'
+
strings
.
pad
(
msg
.
length
,
5
,
'
'
)
+
'
]
'
,
'
color: darkgreen
'
,
'
color: grey
'
,
msg
);
private
_totalOutgoing
=
0
;
});
return
{
logIncoming
(
msgLength
:
number
,
str
:
string
,
data
?:
any
):
void
{
onMessage
:
protocol
.
onMessage
,
this
.
_totalIncoming
+=
msgLength
;
console
.
log
(
`%c[Extension \u2192 Window]%c[
${
strings
.
pad
(
this
.
_totalIncoming
,
7
,
'
'
)}
]%c[len:
${
strings
.
pad
(
msgLength
,
5
,
'
'
)}
]`
,
'
color: darkgreen
'
,
'
color: grey
'
,
'
color: grey
'
,
str
,
data
);
}
send
(
msg
:
any
)
{
logOutgoing
(
msgLength
:
number
,
str
:
string
,
data
?:
any
):
void
{
protocol
.
send
(
msg
)
;
this
.
_totalOutgoing
+=
msgLength
;
console
.
log
(
'
%c[Window
\
u2192 Extension]%c[len:
'
+
strings
.
pad
(
msg
.
length
,
5
,
'
'
)
+
'
]
'
,
'
color: darkgreen
'
,
'
color: grey
'
,
msg
);
console
.
log
(
`%c[Window \u2192 Extension]%c[
${
strings
.
pad
(
this
.
_totalOutgoing
,
7
,
'
'
)}
]%c[len:
${
strings
.
pad
(
msgLength
,
5
,
'
'
)}
]`
,
'
color: darkgreen
'
,
'
color: grey
'
,
'
color: grey
'
,
str
,
data
);
}
}
};
}
}
interface
IExtensionCacheData
{
interface
IExtensionCacheData
{
...
...
src/vs/workbench/services/extensions/node/rpcProtocol.ts
浏览文件 @
acfadb48
...
@@ -85,8 +85,15 @@ function transformIncomingURIs(obj: any, transformer: IURITransformer): any {
...
@@ -85,8 +85,15 @@ function transformIncomingURIs(obj: any, transformer: IURITransformer): any {
return
result
;
return
result
;
}
}
export
interface
IRPCProtocolLogger
{
logIncoming
(
msgLength
:
number
,
str
:
string
,
data
?:
any
):
void
;
logOutgoing
(
msgLength
:
number
,
str
:
string
,
data
?:
any
):
void
;
}
export
class
RPCProtocol
implements
IRPCProtocol
{
export
class
RPCProtocol
implements
IRPCProtocol
{
private
readonly
_protocol
:
IMessagePassingProtocol
;
private
readonly
_logger
:
IRPCProtocolLogger
;
private
readonly
_uriTransformer
:
IURITransformer
;
private
readonly
_uriTransformer
:
IURITransformer
;
private
_isDisposed
:
boolean
;
private
_isDisposed
:
boolean
;
private
readonly
_locals
:
{
[
id
:
string
]:
any
;
};
private
readonly
_locals
:
{
[
id
:
string
]:
any
;
};
...
@@ -94,9 +101,10 @@ export class RPCProtocol implements IRPCProtocol {
...
@@ -94,9 +101,10 @@ export class RPCProtocol implements IRPCProtocol {
private
_lastMessageId
:
number
;
private
_lastMessageId
:
number
;
private
readonly
_invokedHandlers
:
{
[
req
:
string
]:
TPromise
<
any
>
;
};
private
readonly
_invokedHandlers
:
{
[
req
:
string
]:
TPromise
<
any
>
;
};
private
readonly
_pendingRPCReplies
:
{
[
msgId
:
string
]:
LazyPromise
;
};
private
readonly
_pendingRPCReplies
:
{
[
msgId
:
string
]:
LazyPromise
;
};
private
readonly
_multiplexor
:
RPCMultiplexer
;
constructor
(
protocol
:
IMessagePassingProtocol
,
transformer
:
IURITransformer
=
null
)
{
constructor
(
protocol
:
IMessagePassingProtocol
,
logger
:
IRPCProtocolLogger
=
null
,
transformer
:
IURITransformer
=
null
)
{
this
.
_protocol
=
protocol
;
this
.
_logger
=
logger
;
this
.
_uriTransformer
=
transformer
;
this
.
_uriTransformer
=
transformer
;
this
.
_isDisposed
=
false
;
this
.
_isDisposed
=
false
;
this
.
_locals
=
Object
.
create
(
null
);
this
.
_locals
=
Object
.
create
(
null
);
...
@@ -104,7 +112,7 @@ export class RPCProtocol implements IRPCProtocol {
...
@@ -104,7 +112,7 @@ export class RPCProtocol implements IRPCProtocol {
this
.
_lastMessageId
=
0
;
this
.
_lastMessageId
=
0
;
this
.
_invokedHandlers
=
Object
.
create
(
null
);
this
.
_invokedHandlers
=
Object
.
create
(
null
);
this
.
_pendingRPCReplies
=
{};
this
.
_pendingRPCReplies
=
{};
this
.
_
multiplexor
=
new
RPCMultiplexer
(
protocol
,
(
msg
)
=>
this
.
_receiveOneMessage
(
msg
));
this
.
_
protocol
.
onMessage
(
(
msg
)
=>
this
.
_receiveOneMessage
(
msg
));
}
}
public
dispose
():
void
{
public
dispose
():
void
{
...
@@ -164,54 +172,107 @@ export class RPCProtocol implements IRPCProtocol {
...
@@ -164,54 +172,107 @@ export class RPCProtocol implements IRPCProtocol {
return
;
return
;
}
}
let
msg
=
<
RPCMessage
>
JSON
.
parse
(
rawmsg
.
toString
());
const
msgLength
=
rawmsg
.
length
;
const
buff
=
MessageBuffer
.
read
(
rawmsg
,
0
);
const
messageType
=
<
MessageType
>
buff
.
readUInt8
();
const
req
=
buff
.
readUInt32
();
switch
(
messageType
)
{
case
MessageType
.
RequestJSONArgs
:
{
let
{
rpcId
,
method
,
args
}
=
MessageIO
.
deserializeRequestJSONArgs
(
buff
);
if
(
this
.
_uriTransformer
)
{
if
(
this
.
_uriTransformer
)
{
msg
=
transformIncomingURIs
(
msg
,
this
.
_uriTransformer
);
args
=
transformIncomingURIs
(
args
,
this
.
_uriTransformer
);
}
}
this
.
_receiveRequest
(
msgLength
,
req
,
rpcId
,
method
,
args
);
switch
(
msg
.
type
)
{
break
;
case
MessageType
.
Request
:
}
this
.
_receiveRequest
(
msg
);
case
MessageType
.
RequestMixedArgs
:
{
let
{
rpcId
,
method
,
args
}
=
MessageIO
.
deserializeRequestMixedArgs
(
buff
);
if
(
this
.
_uriTransformer
)
{
args
=
transformIncomingURIs
(
args
,
this
.
_uriTransformer
);
}
this
.
_receiveRequest
(
msgLength
,
req
,
rpcId
,
method
,
args
);
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
;
break
;
case
MessageType
.
Cancel
:
}
this
.
_receiveCancel
(
msg
);
case
MessageType
.
ReplyOKBuffer
:
{
let
value
=
MessageIO
.
deserializeReplyOKBuffer
(
buff
);
this
.
_receiveReply
(
msgLength
,
req
,
value
);
break
;
break
;
case
MessageType
.
Reply
:
}
this
.
_receiveReply
(
msg
);
case
MessageType
.
ReplyErrError
:
{
let
err
=
MessageIO
.
deserializeReplyErrError
(
buff
);
if
(
this
.
_uriTransformer
)
{
err
=
transformIncomingURIs
(
err
,
this
.
_uriTransformer
);
}
this
.
_receiveReplyErr
(
req
,
err
);
break
;
break
;
case
MessageType
.
ReplyErr
:
}
this
.
_receiveReplyErr
(
msg
);
case
MessageType
.
ReplyErrEmpty
:
{
this
.
_receiveReplyErr
(
req
,
undefined
);
break
;
break
;
}
}
}
}
}
private
_receiveRequest
(
msg
:
RequestMessage
):
void
{
private
_receiveRequest
(
msgLength
:
number
,
req
:
number
,
rpcId
:
string
,
method
:
string
,
args
:
any
[]):
void
{
const
callId
=
msg
.
id
;
if
(
this
.
_logger
)
{
const
proxyId
=
msg
.
proxyId
;
this
.
_logger
.
logIncoming
(
msgLength
,
`receiveRequest
${
req
}
,
${
rpcId
}
.
${
method
}
:`
,
args
);
}
const
callId
=
String
(
req
);
this
.
_invokedHandlers
[
callId
]
=
this
.
_invokeHandler
(
proxyId
,
msg
.
method
,
msg
.
args
);
this
.
_invokedHandlers
[
callId
]
=
this
.
_invokeHandler
(
rpcId
,
method
,
args
);
this
.
_invokedHandlers
[
callId
].
then
((
r
)
=>
{
this
.
_invokedHandlers
[
callId
].
then
((
r
)
=>
{
delete
this
.
_invokedHandlers
[
callId
];
delete
this
.
_invokedHandlers
[
callId
];
if
(
this
.
_uriTransformer
)
{
if
(
this
.
_uriTransformer
)
{
r
=
transformOutgoingURIs
(
r
,
this
.
_uriTransformer
);
r
=
transformOutgoingURIs
(
r
,
this
.
_uriTransformer
);
}
}
this
.
_multiplexor
.
send
(
Buffer
.
from
(
MessageFactory
.
replyOK
(
callId
,
r
)));
const
msg
=
MessageIO
.
serializeReplyOK
(
req
,
r
);
if
(
this
.
_logger
)
{
this
.
_logger
.
logOutgoing
(
msg
.
byteLength
,
`replyOK
${
req
}
:`
,
r
);
}
this
.
_protocol
.
send
(
msg
);
},
(
err
)
=>
{
},
(
err
)
=>
{
delete
this
.
_invokedHandlers
[
callId
];
delete
this
.
_invokedHandlers
[
callId
];
this
.
_multiplexor
.
send
(
Buffer
.
from
(
MessageFactory
.
replyErr
(
callId
,
err
)));
const
msg
=
MessageIO
.
serializeReplyErr
(
req
,
err
);
if
(
this
.
_logger
)
{
this
.
_logger
.
logOutgoing
(
msg
.
byteLength
,
`replyErr
${
req
}
:`
,
err
);
}
this
.
_protocol
.
send
(
msg
);
});
});
}
}
private
_receiveCancel
(
msg
:
CancelMessage
):
void
{
private
_receiveCancel
(
msgLength
:
number
,
req
:
number
):
void
{
const
callId
=
msg
.
id
;
if
(
this
.
_logger
)
{
this
.
_logger
.
logIncoming
(
msgLength
,
`receiveCancel
${
req
}
`
);
}
const
callId
=
String
(
req
);
if
(
this
.
_invokedHandlers
[
callId
])
{
if
(
this
.
_invokedHandlers
[
callId
])
{
this
.
_invokedHandlers
[
callId
].
cancel
();
this
.
_invokedHandlers
[
callId
].
cancel
();
}
}
}
}
private
_receiveReply
(
msg
:
ReplyMessage
):
void
{
private
_receiveReply
(
msgLength
:
number
,
req
:
number
,
value
:
any
):
void
{
const
callId
=
msg
.
id
;
if
(
this
.
_logger
)
{
this
.
_logger
.
logIncoming
(
msgLength
,
`receiveReply
${
req
}
:`
,
value
);
}
const
callId
=
String
(
req
);
if
(
!
this
.
_pendingRPCReplies
.
hasOwnProperty
(
callId
))
{
if
(
!
this
.
_pendingRPCReplies
.
hasOwnProperty
(
callId
))
{
return
;
return
;
}
}
...
@@ -219,11 +280,11 @@ export class RPCProtocol implements IRPCProtocol {
...
@@ -219,11 +280,11 @@ export class RPCProtocol implements IRPCProtocol {
const
pendingReply
=
this
.
_pendingRPCReplies
[
callId
];
const
pendingReply
=
this
.
_pendingRPCReplies
[
callId
];
delete
this
.
_pendingRPCReplies
[
callId
];
delete
this
.
_pendingRPCReplies
[
callId
];
pendingReply
.
resolveOk
(
msg
.
res
);
pendingReply
.
resolveOk
(
value
);
}
}
private
_receiveReplyErr
(
msg
:
ReplyErrMessage
):
void
{
private
_receiveReplyErr
(
req
:
number
,
value
:
any
):
void
{
const
callId
=
msg
.
id
;
const
callId
=
String
(
req
)
;
if
(
!
this
.
_pendingRPCReplies
.
hasOwnProperty
(
callId
))
{
if
(
!
this
.
_pendingRPCReplies
.
hasOwnProperty
(
callId
))
{
return
;
return
;
}
}
...
@@ -232,11 +293,11 @@ export class RPCProtocol implements IRPCProtocol {
...
@@ -232,11 +293,11 @@ export class RPCProtocol implements IRPCProtocol {
delete
this
.
_pendingRPCReplies
[
callId
];
delete
this
.
_pendingRPCReplies
[
callId
];
let
err
:
Error
=
null
;
let
err
:
Error
=
null
;
if
(
msg
.
err
&&
msg
.
err
.
$isError
)
{
if
(
value
&&
value
.
$isError
)
{
err
=
new
Error
();
err
=
new
Error
();
err
.
name
=
msg
.
err
.
name
;
err
.
name
=
value
.
name
;
err
.
message
=
msg
.
err
.
message
;
err
.
message
=
value
.
message
;
err
.
stack
=
msg
.
err
.
stack
;
err
.
stack
=
value
.
stack
;
}
}
pendingReply
.
resolveErr
(
err
);
pendingReply
.
resolveErr
(
err
);
}
}
...
@@ -266,123 +327,347 @@ export class RPCProtocol implements IRPCProtocol {
...
@@ -266,123 +327,347 @@ export class RPCProtocol implements IRPCProtocol {
return
TPromise
.
wrapError
<
any
>
(
errors
.
canceled
());
return
TPromise
.
wrapError
<
any
>
(
errors
.
canceled
());
}
}
const
callId
=
String
(
++
this
.
_lastMessageId
);
const
req
=
++
this
.
_lastMessageId
;
const
callId
=
String
(
req
);
const
result
=
new
LazyPromise
(()
=>
{
const
result
=
new
LazyPromise
(()
=>
{
this
.
_multiplexor
.
send
(
Buffer
.
from
(
MessageFactory
.
cancel
(
callId
)));
const
msg
=
MessageIO
.
serializeCancel
(
req
);
if
(
this
.
_logger
)
{
this
.
_logger
.
logOutgoing
(
msg
.
byteLength
,
`cancel
${
req
}
`
);
}
this
.
_protocol
.
send
(
MessageIO
.
serializeCancel
(
req
));
});
});
this
.
_pendingRPCReplies
[
callId
]
=
result
;
this
.
_pendingRPCReplies
[
callId
]
=
result
;
if
(
this
.
_uriTransformer
)
{
if
(
this
.
_uriTransformer
)
{
args
=
transformOutgoingURIs
(
args
,
this
.
_uriTransformer
);
args
=
transformOutgoingURIs
(
args
,
this
.
_uriTransformer
);
}
}
this
.
_multiplexor
.
send
(
Buffer
.
from
(
MessageFactory
.
request
(
callId
,
proxyId
,
methodName
,
args
)));
const
msg
=
MessageIO
.
serializeRequest
(
req
,
proxyId
,
methodName
,
args
);
if
(
this
.
_logger
)
{
this
.
_logger
.
logOutgoing
(
msg
.
byteLength
,
`request
${
req
}
:
${
proxyId
}
.
${
methodName
}
:`
,
args
);
}
this
.
_protocol
.
send
(
msg
);
return
result
;
return
result
;
}
}
}
}
/**
class
MessageBuffer
{
* Sends/Receives multiple messages in one go:
* - multiple messages to be sent from one stack get sent in bulk at `process.nextTick`.
* - each incoming message is handled in a separate `process.nextTick`.
*/
class
RPCMultiplexer
{
private
readonly
_protocol
:
IMessagePassingProtocol
;
public
static
alloc
(
type
:
MessageType
,
req
:
number
,
messageSize
:
number
):
MessageBuffer
{
private
readonly
_sendAccumulatedBound
:
()
=>
void
;
let
result
=
new
MessageBuffer
(
Buffer
.
allocUnsafe
(
messageSize
+
1
/* type */
+
4
/* req */
),
0
);
result
.
writeUInt8
(
type
);
result
.
writeUInt32
(
req
);
return
result
;
}
private
_messagesToSend
:
Buffer
[];
public
static
read
(
buff
:
Buffer
,
offset
:
number
):
MessageBuffer
{
return
new
MessageBuffer
(
buff
,
offset
);
}
constructor
(
protocol
:
IMessagePassingProtocol
,
onMessage
:
(
msg
:
Buffer
)
=>
void
)
{
private
_buff
:
Buffer
;
this
.
_protocol
=
protocol
;
private
_offset
:
number
;
this
.
_sendAccumulatedBound
=
this
.
_sendAccumulated
.
bind
(
this
);
this
.
_messagesToSend
=
[];
public
get
buffer
():
Buffer
{
return
this
.
_buff
;
}
this
.
_protocol
.
onMessage
(
data
=>
{
private
constructor
(
buff
:
Buffer
,
offset
:
number
)
{
let
i
=
0
;
this
.
_buff
=
buff
;
this
.
_offset
=
offset
;
}
while
(
i
<
data
.
length
)
{
public
writeUInt8
(
n
:
number
):
void
{
const
size
=
data
.
readUInt32BE
(
i
);
this
.
_buff
.
writeUInt8
(
n
,
this
.
_offset
,
true
);
this
.
_offset
+=
1
;
onMessage
(
data
.
slice
(
i
+
4
,
i
+
4
+
size
));
i
+=
4
+
size
;
}
}
});
public
readUInt8
():
number
{
const
n
=
this
.
_buff
.
readUInt8
(
this
.
_offset
,
true
);
this
.
_offset
+=
1
;
return
n
;
}
}
private
_sendAccumulated
():
void
{
public
writeUInt32
(
n
:
number
):
void
{
const
size
=
this
.
_messagesToSend
.
reduce
((
r
,
b
)
=>
r
+
4
+
b
.
byteLength
,
0
);
this
.
_buff
.
writeUInt32BE
(
n
,
this
.
_offset
,
true
);
this
.
_offset
+=
4
;
const
buffer
=
Buffer
.
allocUnsafe
(
size
);
}
let
i
=
0
;
for
(
const
msg
of
this
.
_messagesToSend
)
{
public
readUInt32
():
number
{
buffer
.
writeUInt32BE
(
msg
.
byteLength
,
i
);
const
n
=
this
.
_buff
.
readUInt32BE
(
this
.
_offset
,
true
);
this
.
_offset
+=
4
;
msg
.
copy
(
buffer
,
i
+
4
);
return
n
;
i
+=
4
+
msg
.
byteLength
;
}
}
this
.
_messagesToSend
=
[];
public
static
sizeShortString
(
str
:
string
,
strByteLength
:
number
):
number
{
this
.
_protocol
.
send
(
buffer
)
;
return
1
/* string length */
+
strByteLength
/* actual string */
;
}
}
public
send
(
msg
:
Buff
er
):
void
{
public
writeShortString
(
str
:
string
,
strByteLength
:
numb
er
):
void
{
if
(
this
.
_messagesToSend
.
length
===
0
)
{
this
.
_buff
.
writeUInt8
(
strByteLength
,
this
.
_offset
,
true
);
this
.
_offset
+=
1
;
process
.
nextTick
(
this
.
_sendAccumulatedBound
)
;
this
.
_buff
.
write
(
str
,
this
.
_offset
,
strByteLength
,
'
utf8
'
);
this
.
_offset
+=
strByteLength
;
}
}
this
.
_messagesToSend
.
push
(
msg
);
public
readShortString
():
string
{
const
strLength
=
this
.
_buff
.
readUInt8
(
this
.
_offset
,
true
);
this
.
_offset
+=
1
;
const
str
=
this
.
_buff
.
toString
(
'
utf8
'
,
this
.
_offset
,
this
.
_offset
+
strLength
);
this
.
_offset
+=
strLength
;
return
str
;
}
public
static
sizeLongString
(
str
:
string
,
strByteLength
:
number
):
number
{
return
4
/* string length */
+
strByteLength
/* actual string */
;
}
public
writeLongString
(
str
:
string
,
strByteLength
:
number
):
void
{
this
.
_buff
.
writeUInt32LE
(
strByteLength
,
this
.
_offset
,
true
);
this
.
_offset
+=
4
;
this
.
_buff
.
write
(
str
,
this
.
_offset
,
strByteLength
,
'
utf8
'
);
this
.
_offset
+=
strByteLength
;
}
public
readLongString
():
string
{
const
strLength
=
this
.
_buff
.
readUInt32LE
(
this
.
_offset
,
true
);
this
.
_offset
+=
4
;
const
str
=
this
.
_buff
.
toString
(
'
utf8
'
,
this
.
_offset
,
this
.
_offset
+
strLength
);
this
.
_offset
+=
strLength
;
return
str
;
}
public
static
sizeBuffer
(
buff
:
Buffer
,
buffByteLength
:
number
):
number
{
return
4
/* buffer length */
+
buffByteLength
/* actual buffer */
;
}
public
writeBuffer
(
buff
:
Buffer
,
buffByteLength
:
number
):
void
{
this
.
_buff
.
writeUInt32LE
(
buffByteLength
,
this
.
_offset
,
true
);
this
.
_offset
+=
4
;
buff
.
copy
(
this
.
_buff
,
this
.
_offset
);
this
.
_offset
+=
buffByteLength
;
}
public
readBuffer
():
Buffer
{
const
buffLength
=
this
.
_buff
.
readUInt32LE
(
this
.
_offset
,
true
);
this
.
_offset
+=
4
;
const
buff
=
this
.
_buff
.
slice
(
this
.
_offset
,
this
.
_offset
+
buffLength
);
this
.
_offset
+=
buffLength
;
return
buff
;
}
public
static
sizeMixedArray
(
arr
:
(
string
|
Buffer
)[],
arrLengths
:
number
[]):
number
{
let
size
=
0
;
size
+=
1
;
// arr length
for
(
let
i
=
0
,
len
=
arr
.
length
;
i
<
len
;
i
++
)
{
const
el
=
arr
[
i
];
const
elLength
=
arrLengths
[
i
];
size
+=
1
;
// arg type
if
(
typeof
el
===
'
string
'
)
{
size
+=
this
.
sizeLongString
(
el
,
elLength
);
}
else
{
size
+=
this
.
sizeBuffer
(
el
,
elLength
);
}
}
return
size
;
}
public
writeMixedArray
(
arr
:
(
string
|
Buffer
)[],
arrLengths
:
number
[]):
void
{
this
.
_buff
.
writeUInt8
(
arr
.
length
,
this
.
_offset
,
true
);
this
.
_offset
+=
1
;
for
(
let
i
=
0
,
len
=
arr
.
length
;
i
<
len
;
i
++
)
{
const
el
=
arr
[
i
];
const
elLength
=
arrLengths
[
i
];
if
(
typeof
el
===
'
string
'
)
{
this
.
writeUInt8
(
ArgType
.
String
);
this
.
writeLongString
(
el
,
elLength
);
}
else
{
this
.
writeUInt8
(
ArgType
.
Buffer
);
this
.
writeBuffer
(
el
,
elLength
);
}
}
}
public
readMixedArray
():
(
string
|
Buffer
)[]
{
const
arrLen
=
this
.
_buff
.
readUInt8
(
this
.
_offset
,
true
);
this
.
_offset
+=
1
;
let
arr
:
(
string
|
Buffer
)[]
=
new
Array
(
arrLen
);
for
(
let
i
=
0
;
i
<
arrLen
;
i
++
)
{
const
argType
=
<
ArgType
>
this
.
readUInt8
();
if
(
argType
===
ArgType
.
String
)
{
arr
[
i
]
=
this
.
readLongString
();
}
else
{
arr
[
i
]
=
this
.
readBuffer
();
}
}
return
arr
;
}
}
}
}
class
MessageFactory
{
class
MessageIO
{
public
static
cancel
(
req
:
string
):
string
{
return
`{"type":
${
MessageType
.
Cancel
}
,"id":"
${
req
}
"}`
;
private
static
_arrayContainsBuffer
(
arr
:
any
[]):
boolean
{
for
(
let
i
=
0
,
len
=
arr
.
length
;
i
<
len
;
i
++
)
{
if
(
Buffer
.
isBuffer
(
arr
[
i
]))
{
return
true
;
}
}
return
false
;
}
public
static
serializeRequest
(
req
:
number
,
rpcId
:
string
,
method
:
string
,
args
:
any
[]):
Buffer
{
if
(
this
.
_arrayContainsBuffer
(
args
))
{
let
massagedArgs
:
(
string
|
Buffer
)[]
=
new
Array
(
args
.
length
);
let
argsLengths
:
number
[]
=
new
Array
(
args
.
length
);
for
(
let
i
=
0
,
len
=
args
.
length
;
i
<
len
;
i
++
)
{
const
arg
=
args
[
i
];
if
(
Buffer
.
isBuffer
(
arg
))
{
massagedArgs
[
i
]
=
arg
;
argsLengths
[
i
]
=
arg
.
byteLength
;
}
else
{
massagedArgs
[
i
]
=
JSON
.
stringify
(
arg
);
argsLengths
[
i
]
=
Buffer
.
byteLength
(
massagedArgs
[
i
],
'
utf8
'
);
}
}
return
this
.
_requestMixedArgs
(
req
,
rpcId
,
method
,
massagedArgs
,
argsLengths
);
}
return
this
.
_requestJSONArgs
(
req
,
rpcId
,
method
,
JSON
.
stringify
(
args
));
}
private
static
_requestJSONArgs
(
req
:
number
,
rpcId
:
string
,
method
:
string
,
args
:
string
):
Buffer
{
const
rpcIdByteLength
=
Buffer
.
byteLength
(
rpcId
,
'
utf8
'
);
const
methodByteLength
=
Buffer
.
byteLength
(
method
,
'
utf8
'
);
const
argsByteLength
=
Buffer
.
byteLength
(
args
,
'
utf8
'
);
let
len
=
0
;
len
+=
MessageBuffer
.
sizeShortString
(
rpcId
,
rpcIdByteLength
);
len
+=
MessageBuffer
.
sizeShortString
(
method
,
methodByteLength
);
len
+=
MessageBuffer
.
sizeLongString
(
args
,
argsByteLength
);
let
result
=
MessageBuffer
.
alloc
(
MessageType
.
RequestJSONArgs
,
req
,
len
);
result
.
writeShortString
(
rpcId
,
rpcIdByteLength
);
result
.
writeShortString
(
method
,
methodByteLength
);
result
.
writeLongString
(
args
,
argsByteLength
);
return
result
.
buffer
;
}
public
static
deserializeRequestJSONArgs
(
buff
:
MessageBuffer
):
{
rpcId
:
string
;
method
:
string
;
args
:
any
[];
}
{
const
rpcId
=
buff
.
readShortString
();
const
method
=
buff
.
readShortString
();
const
args
=
buff
.
readLongString
();
return
{
rpcId
:
rpcId
,
method
:
method
,
args
:
JSON
.
parse
(
args
)
};
}
private
static
_requestMixedArgs
(
req
:
number
,
rpcId
:
string
,
method
:
string
,
args
:
(
string
|
Buffer
)[],
argsLengths
:
number
[]):
Buffer
{
const
rpcIdByteLength
=
Buffer
.
byteLength
(
rpcId
,
'
utf8
'
);
const
methodByteLength
=
Buffer
.
byteLength
(
method
,
'
utf8
'
);
let
len
=
0
;
len
+=
MessageBuffer
.
sizeShortString
(
rpcId
,
rpcIdByteLength
);
len
+=
MessageBuffer
.
sizeShortString
(
method
,
methodByteLength
);
len
+=
MessageBuffer
.
sizeMixedArray
(
args
,
argsLengths
);
let
result
=
MessageBuffer
.
alloc
(
MessageType
.
RequestMixedArgs
,
req
,
len
);
result
.
writeShortString
(
rpcId
,
rpcIdByteLength
);
result
.
writeShortString
(
method
,
methodByteLength
);
result
.
writeMixedArray
(
args
,
argsLengths
);
return
result
.
buffer
;
}
public
static
deserializeRequestMixedArgs
(
buff
:
MessageBuffer
):
{
rpcId
:
string
;
method
:
string
;
args
:
any
[];
}
{
const
rpcId
=
buff
.
readShortString
();
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
request
(
req
:
string
,
rpcId
:
string
,
method
:
string
,
args
:
any
[]):
string
{
public
static
serializeCancel
(
req
:
number
):
Buffer
{
return
`{"type":
${
MessageType
.
Request
}
,"id":"
${
req
}
","proxyId":"
${
rpcId
}
","method":"
${
method
}
","args":
${
JSON
.
stringify
(
args
)}
}`
;
return
MessageBuffer
.
alloc
(
MessageType
.
Cancel
,
req
,
0
).
buffer
;
}
}
public
static
replyOK
(
req
:
string
,
res
:
any
):
string
{
public
static
serializeReplyOK
(
req
:
number
,
res
:
any
):
Buffer
{
if
(
typeof
res
===
'
undefined
'
)
{
if
(
typeof
res
===
'
undefined
'
)
{
return
`{"type":
${
MessageType
.
Reply
}
,"id":"
${
req
}
"}`
;
return
this
.
_serializeReplyOKEmpty
(
req
);
}
if
(
Buffer
.
isBuffer
(
res
))
{
return
this
.
_serializeReplyOKBuffer
(
req
,
res
);
}
return
this
.
_serializeReplyOKJSON
(
req
,
JSON
.
stringify
(
res
));
}
}
return
`{"type":
${
MessageType
.
Reply
}
,"id":"
${
req
}
","res":
${
JSON
.
stringify
(
res
)}
}`
;
private
static
_serializeReplyOKEmpty
(
req
:
number
):
Buffer
{
return
MessageBuffer
.
alloc
(
MessageType
.
ReplyOKEmpty
,
req
,
0
).
buffer
;
}
private
static
_serializeReplyOKBuffer
(
req
:
number
,
res
:
Buffer
):
Buffer
{
const
resByteLength
=
res
.
byteLength
;
let
len
=
0
;
len
+=
MessageBuffer
.
sizeBuffer
(
res
,
resByteLength
);
let
result
=
MessageBuffer
.
alloc
(
MessageType
.
ReplyOKBuffer
,
req
,
len
);
result
.
writeBuffer
(
res
,
resByteLength
);
return
result
.
buffer
;
}
public
static
deserializeReplyOKBuffer
(
buff
:
MessageBuffer
):
Buffer
{
return
buff
.
readBuffer
();
}
private
static
_serializeReplyOKJSON
(
req
:
number
,
res
:
string
):
Buffer
{
const
resByteLength
=
Buffer
.
byteLength
(
res
,
'
utf8
'
);
let
len
=
0
;
len
+=
MessageBuffer
.
sizeLongString
(
res
,
resByteLength
);
let
result
=
MessageBuffer
.
alloc
(
MessageType
.
ReplyOKJSON
,
req
,
len
);
result
.
writeLongString
(
res
,
resByteLength
);
return
result
.
buffer
;
}
}
public
static
replyErr
(
req
:
string
,
err
:
any
):
string
{
public
static
deserializeReplyOKJSON
(
buff
:
MessageBuffer
):
any
{
const
res
=
buff
.
readLongString
();
return
JSON
.
parse
(
res
);
}
public
static
serializeReplyErr
(
req
:
number
,
err
:
any
):
Buffer
{
if
(
err
instanceof
Error
)
{
if
(
err
instanceof
Error
)
{
return
`{"type":
${
MessageType
.
ReplyErr
}
,"id":"
${
req
}
","err":
${
JSON
.
stringify
(
errors
.
transformErrorForSerialization
(
err
))}
}`
;
return
this
.
_serializeReplyErrEror
(
req
,
err
);
}
return
this
.
_serializeReplyErrEmpty
(
req
);
}
private
static
_serializeReplyErrEror
(
req
:
number
,
_err
:
Error
):
Buffer
{
const
err
=
JSON
.
stringify
(
errors
.
transformErrorForSerialization
(
_err
));
const
errByteLength
=
Buffer
.
byteLength
(
err
,
'
utf8
'
);
let
len
=
0
;
len
+=
MessageBuffer
.
sizeLongString
(
err
,
errByteLength
);
let
result
=
MessageBuffer
.
alloc
(
MessageType
.
ReplyErrError
,
req
,
len
);
result
.
writeLongString
(
err
,
errByteLength
);
return
result
.
buffer
;
}
}
return
`{"type":
${
MessageType
.
ReplyErr
}
,"id":"
${
req
}
","err":null}`
;
public
static
deserializeReplyErrError
(
buff
:
MessageBuffer
):
Error
{
const
err
=
buff
.
readLongString
();
return
JSON
.
parse
(
err
);
}
private
static
_serializeReplyErrEmpty
(
req
:
number
):
Buffer
{
return
MessageBuffer
.
alloc
(
MessageType
.
ReplyErrEmpty
,
req
,
0
).
buffer
;
}
}
}
}
const
enum
MessageType
{
const
enum
MessageType
{
Request
=
1
,
RequestJSONArgs
=
1
,
Cancel
=
2
,
RequestMixedArgs
=
2
,
Reply
=
3
,
Cancel
=
3
,
ReplyErr
=
4
ReplyOKEmpty
=
4
,
ReplyOKBuffer
=
5
,
ReplyOKJSON
=
6
,
ReplyErrError
=
7
,
ReplyErrEmpty
=
8
,
}
}
class
RequestMessage
{
const
enum
ArgType
{
type
:
MessageType
.
Request
;
String
=
1
,
id
:
string
;
Buffer
=
2
proxyId
:
string
;
method
:
string
;
args
:
any
[];
}
class
CancelMessage
{
type
:
MessageType
.
Cancel
;
id
:
string
;
}
}
class
ReplyMessage
{
type
:
MessageType
.
Reply
;
id
:
string
;
res
:
any
;
}
class
ReplyErrMessage
{
type
:
MessageType
.
ReplyErr
;
id
:
string
;
err
:
errors
.
SerializedError
;
}
type
RPCMessage
=
RequestMessage
|
CancelMessage
|
ReplyMessage
|
ReplyErrMessage
;
src/vs/workbench/services/extensions/test/node/rpcProtocol.test.ts
0 → 100644
浏览文件 @
acfadb48
/*---------------------------------------------------------------------------------------------
* 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
{
RPCProtocol
}
from
'
vs/workbench/services/extensions/node/rpcProtocol
'
;
import
{
IMessagePassingProtocol
}
from
'
vs/base/parts/ipc/node/ipc
'
;
import
{
Event
,
Emitter
}
from
'
vs/base/common/event
'
;
import
{
ProxyIdentifier
}
from
'
vs/workbench/services/extensions/node/proxyIdentifier
'
;
import
{
TPromise
}
from
'
vs/base/common/winjs.base
'
;
suite
(
'
RPCProtocol
'
,
()
=>
{
class
MessagePassingProtocol
implements
IMessagePassingProtocol
{
private
_pair
:
MessagePassingProtocol
;
private
readonly
_onMessage
:
Emitter
<
Buffer
>
=
new
Emitter
<
Buffer
>
();
public
readonly
onMessage
:
Event
<
Buffer
>
=
this
.
_onMessage
.
event
;
public
setPair
(
other
:
MessagePassingProtocol
)
{
this
.
_pair
=
other
;
}
public
send
(
buffer
:
Buffer
):
void
{
process
.
nextTick
(()
=>
{
this
.
_pair
.
_onMessage
.
fire
(
buffer
);
});
}
}
let
delegate
:
(
a1
:
any
,
a2
:
any
)
=>
any
;
let
bProxy
:
BClass
;
class
BClass
{
$m
(
a1
:
any
,
a2
:
any
):
TPromise
<
any
>
{
return
TPromise
.
as
(
delegate
.
call
(
null
,
a1
,
a2
));
}
}
setup
(()
=>
{
let
a_protocol
=
new
MessagePassingProtocol
();
let
b_protocol
=
new
MessagePassingProtocol
();
a_protocol
.
setPair
(
b_protocol
);
b_protocol
.
setPair
(
a_protocol
);
let
A
=
new
RPCProtocol
(
a_protocol
);
let
B
=
new
RPCProtocol
(
b_protocol
);
delegate
=
null
;
const
bIdentifier
=
new
ProxyIdentifier
<
BClass
>
(
false
,
'
bb
'
);
const
bInstance
=
new
BClass
();
B
.
set
(
bIdentifier
,
bInstance
);
bProxy
=
A
.
getProxy
(
bIdentifier
);
});
test
(
'
simple call
'
,
function
(
done
)
{
delegate
=
(
a1
:
number
,
a2
:
number
)
=>
a1
+
a2
;
bProxy
.
$m
(
4
,
1
).
then
((
res
:
number
)
=>
{
assert
.
equal
(
res
,
5
);
done
(
null
);
},
done
);
});
test
(
'
simple call without result
'
,
function
(
done
)
{
delegate
=
(
a1
:
number
,
a2
:
number
)
=>
{
};
bProxy
.
$m
(
4
,
1
).
then
((
res
:
number
)
=>
{
assert
.
equal
(
res
,
undefined
);
done
(
null
);
},
done
);
});
test
(
'
passing buffer as argument
'
,
function
(
done
)
{
delegate
=
(
a1
:
Buffer
,
a2
:
number
)
=>
{
assert
.
ok
(
Buffer
.
isBuffer
(
a1
));
return
a1
[
a2
];
};
let
b
=
Buffer
.
allocUnsafe
(
4
);
b
[
0
]
=
1
;
b
[
1
]
=
2
;
b
[
2
]
=
3
;
b
[
3
]
=
4
;
bProxy
.
$m
(
b
,
2
).
then
((
res
:
number
)
=>
{
assert
.
equal
(
res
,
3
);
done
(
null
);
},
done
);
});
test
(
'
returning a buffer
'
,
function
(
done
)
{
delegate
=
(
a1
:
number
,
a2
:
number
)
=>
{
let
b
=
Buffer
.
allocUnsafe
(
4
);
b
[
0
]
=
1
;
b
[
1
]
=
2
;
b
[
2
]
=
3
;
b
[
3
]
=
4
;
return
b
;
};
bProxy
.
$m
(
4
,
1
).
then
((
res
:
Buffer
)
=>
{
assert
.
ok
(
Buffer
.
isBuffer
(
res
));
assert
.
equal
(
res
[
0
],
1
);
assert
.
equal
(
res
[
1
],
2
);
assert
.
equal
(
res
[
2
],
3
);
assert
.
equal
(
res
[
3
],
4
);
done
(
null
);
},
done
);
});
test
(
'
cancelling a call
'
,
function
()
{
delegate
=
(
a1
:
number
,
a2
:
number
)
=>
a1
+
a2
;
let
p
=
bProxy
.
$m
(
4
,
1
);
p
.
then
((
res
:
number
)
=>
{
assert
.
fail
(
'
should not receive result
'
);
});
p
.
cancel
();
});
test
(
'
throwing an error
'
,
function
(
done
)
{
delegate
=
(
a1
:
number
,
a2
:
number
)
=>
{
throw
new
Error
(
`nope`
);
};
bProxy
.
$m
(
4
,
1
).
then
((
res
)
=>
{
assert
.
fail
(
'
unexpected
'
);
done
(
null
);
},
(
err
)
=>
{
assert
.
equal
(
err
.
message
,
'
nope
'
);
done
(
null
);
});
});
test
(
'
error promise
'
,
function
(
done
)
{
delegate
=
(
a1
:
number
,
a2
:
number
)
=>
{
return
TPromise
.
wrapError
(
undefined
);
};
bProxy
.
$m
(
4
,
1
).
then
((
res
)
=>
{
assert
.
fail
(
'
unexpected
'
);
done
(
null
);
},
(
err
)
=>
{
assert
.
equal
(
err
,
undefined
);
done
(
null
);
});
});
});
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录