-
-
Notifications
You must be signed in to change notification settings - Fork 7
/
protocol.js
118 lines (93 loc) · 2.55 KB
/
protocol.js
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
var messages = require('./messages')
var lpstream = require('length-prefixed-stream')
var duplexify = require('duplexify')
var util = require('util')
var DESTROYED = new Error('Stream was destroyed')
DESTROYED.retry = true
var ENCODERS = [
messages.Handshake,
messages.Have,
messages.Request,
messages.Response
]
module.exports = ProtocolStream
function ProtocolStream (opts) {
if (!(this instanceof ProtocolStream)) return new ProtocolStream(opts)
if (!opts) opts = {}
this._encode = lpstream.encode()
this._decode = lpstream.decode({limit: 5 * 1024 * 1024})
this._requests = []
this._handshook = false
this.destroyed = false
duplexify.call(this, this._decode, this._encode)
var self = this
this._decode.on('data', function (data) {
self._parse(data)
})
this._send(0, opts)
}
util.inherits(ProtocolStream, duplexify)
ProtocolStream.prototype.destroy = function (err) {
if (this.destroyed) return
this.destroyed = true
while (this._requests.length) {
var cb = this._requests.shift()
if (cb) cb(err || DESTROYED)
}
if (err) this.emit('error', err)
this.emit('close')
}
ProtocolStream.prototype._parse = function (buf) {
if (this.destroyed) return
var dec = ENCODERS[buf[0]]
try {
var data = dec.decode(buf, 1)
} catch (err) {
return this.destroy(err)
}
if (!this._handshook && buf[0] !== 0) {
this.emit('error', new Error('First message should be a handshake'))
return
}
switch (buf[0]) {
case 0:
this._handshook = true
this.emit('handshake', data)
return
case 1:
this.emit('have', data)
return
case 2:
this.emit('request', data)
return
case 3:
if (this._requests.length > data.id) {
var cb = this._requests[data.id]
this._requests[data.id] = null
while (this._requests.length && !this._requests[this._requests.length - 1]) this._requests.pop()
if (cb) cb(null, data)
}
this.emit('response', data)
return
}
this.emit('unknown', buf)
}
ProtocolStream.prototype.have = function (have) {
this._send(1, have)
}
ProtocolStream.prototype.request = function (req, cb) {
req.id = this._requests.indexOf(null)
if (req.id === -1) req.id = this._requests.push(null) - 1
this._requests[req.id] = cb
this._send(2, req)
}
ProtocolStream.prototype.response = function (res) {
this._send(3, res)
}
ProtocolStream.prototype._send = function (type, data, cb) {
var enc = ENCODERS[type]
var buf = new Buffer(1 + enc.encodingLength(data))
buf[0] = type
enc.encode(data, buf, 1)
this._encode.write(buf, cb)
}