1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
const Duplex = require('readable-stream').Duplex
const inherits = require('util').inherits
module.exports = StreamProvider
inherits(StreamProvider, Duplex)
function StreamProvider(){
Duplex.call(this, {
objectMode: true,
})
this._payloads = {}
}
// public
StreamProvider.prototype.send = function(payload){
throw new Error('StreamProvider - does not support synchronous RPC calls. called: "'+payload.method+'"')
}
StreamProvider.prototype.sendAsync = function(payload, callback){
// console.log('StreamProvider - sending payload', payload)
var id = payload.id
if (Array.isArray(payload)) {
id = 'batch'+payload[0].id
}
this._payloads[id] = [payload, callback]
// console.log('payload for plugin:', payload)
this.push(payload)
}
StreamProvider.prototype.isConnected = function(){
return true
}
// private
StreamProvider.prototype._onResponse = function(response){
// console.log('StreamProvider - got response', payload)
var id = response.id
if (Array.isArray(response)) {
id = 'batch'+response[0].id
}
var data = this._payloads[id]
if (!data) throw new Error('StreamProvider - Unknown response id')
delete this._payloads[id]
var payload = data[0]
var callback = data[1]
// logging
var res = Array.isArray(response) ? response : [response]
// ;(Array.isArray(payload) ? payload : [payload]).forEach(function(payload, index){
// console.log('plugin response:', payload.id, payload.method, payload.params, '->', res[index].result)
// })
callback(null, response)
}
// stream plumbing
StreamProvider.prototype._read = noop
StreamProvider.prototype._write = function(msg, encoding, cb){
this._onResponse(msg)
cb()
}
// util
function noop(){}
|
