message-parser.js
4.72 KB
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
'use strict';
const Transform = require('stream').Transform;
/**
* MessageParser instance is a transform stream that separates message headers
* from the rest of the body. Headers are emitted with the 'headers' event. Message
* body is passed on as the resulting stream.
*/
class MessageParser extends Transform {
constructor(options) {
super(options);
this.lastBytes = Buffer.alloc(4);
this.headersParsed = false;
this.headerBytes = 0;
this.headerChunks = [];
this.rawHeaders = false;
this.bodySize = 0;
}
/**
* Keeps count of the last 4 bytes in order to detect line breaks on chunk boundaries
*
* @param {Buffer} data Next data chunk from the stream
*/
updateLastBytes(data) {
let lblen = this.lastBytes.length;
let nblen = Math.min(data.length, lblen);
// shift existing bytes
for (let i = 0, len = lblen - nblen; i < len; i++) {
this.lastBytes[i] = this.lastBytes[i + nblen];
}
// add new bytes
for (let i = 1; i <= nblen; i++) {
this.lastBytes[lblen - i] = data[data.length - i];
}
}
/**
* Finds and removes message headers from the remaining body. We want to keep
* headers separated until final delivery to be able to modify these
*
* @param {Buffer} data Next chunk of data
* @return {Boolean} Returns true if headers are already found or false otherwise
*/
checkHeaders(data) {
if (this.headersParsed) {
return true;
}
let lblen = this.lastBytes.length;
let headerPos = 0;
this.curLinePos = 0;
for (let i = 0, len = this.lastBytes.length + data.length; i < len; i++) {
let chr;
if (i < lblen) {
chr = this.lastBytes[i];
} else {
chr = data[i - lblen];
}
if (chr === 0x0a && i) {
let pr1 = i - 1 < lblen ? this.lastBytes[i - 1] : data[i - 1 - lblen];
let pr2 = i > 1 ? (i - 2 < lblen ? this.lastBytes[i - 2] : data[i - 2 - lblen]) : false;
if (pr1 === 0x0a) {
this.headersParsed = true;
headerPos = i - lblen + 1;
this.headerBytes += headerPos;
break;
} else if (pr1 === 0x0d && pr2 === 0x0a) {
this.headersParsed = true;
headerPos = i - lblen + 1;
this.headerBytes += headerPos;
break;
}
}
}
if (this.headersParsed) {
this.headerChunks.push(data.slice(0, headerPos));
this.rawHeaders = Buffer.concat(this.headerChunks, this.headerBytes);
this.headerChunks = null;
this.emit('headers', this.parseHeaders());
if (data.length - 1 > headerPos) {
let chunk = data.slice(headerPos);
this.bodySize += chunk.length;
// this would be the first chunk of data sent downstream
setImmediate(() => this.push(chunk));
}
return false;
} else {
this.headerBytes += data.length;
this.headerChunks.push(data);
}
// store last 4 bytes to catch header break
this.updateLastBytes(data);
return false;
}
_transform(chunk, encoding, callback) {
if (!chunk || !chunk.length) {
return callback();
}
if (typeof chunk === 'string') {
chunk = new Buffer(chunk, encoding);
}
let headersFound;
try {
headersFound = this.checkHeaders(chunk);
} catch (E) {
return callback(E);
}
if (headersFound) {
this.bodySize += chunk.length;
this.push(chunk);
}
setImmediate(callback);
}
_flush(callback) {
if (this.headerChunks) {
let chunk = Buffer.concat(this.headerChunks, this.headerBytes);
this.bodySize += chunk.length;
this.push(chunk);
this.headerChunks = null;
}
callback();
}
parseHeaders() {
let lines = (this.rawHeaders || '').toString().split(/\r?\n/);
for (let i = lines.length - 1; i > 0; i--) {
if (/^\s/.test(lines[i])) {
lines[i - 1] += '\n' + lines[i];
lines.splice(i, 1);
}
}
return lines.filter(line => line.trim()).map(line => ({
key: line
.substr(0, line.indexOf(':'))
.trim()
.toLowerCase(),
line
}));
}
}
module.exports = MessageParser;