codekingpro/portable-devtools
114k
1'use strict'
2const { Transform } = require('node:stream')
3const { isASCIINumber, isValidLastEventId } = require('./util')
4
5/**
6 * @type {number[]} BOM
7 */
8const BOM = [0xEF, 0xBB, 0xBF]
9/**
10 * @type {10} LF
11 */
12const LF = 0x0A
13/**
14 * @type {13} CR
15 */
16const CR = 0x0D
17/**
18 * @type {58} COLON
19 */
20const COLON = 0x3A
21/**
22 * @type {32} SPACE
23 */
24const SPACE = 0x20
25
26/**
27 * @typedef {object} EventSourceStreamEvent
28 * @type {object}
29 * @property {string} [event] The event type.
30 * @property {string} [data] The data of the message.
31 * @property {string} [id] A unique ID for the event.
32 * @property {string} [retry] The reconnection time, in milliseconds.
33 */
34
35/**
36 * @typedef eventSourceSettings
37 * @type {object}
38 * @property {string} lastEventId The last event ID received from the server.
39 * @property {string} origin The origin of the event source.
40 * @property {number} reconnectionTime The reconnection time, in milliseconds.
41 */
42
43class EventSourceStream extends Transform {
44 /**
45 * @type {eventSourceSettings}
46 */
47 state = null
48
49 /**
50 * Leading byte-order-mark check.
51 * @type {boolean}
52 */
53 checkBOM = true
54
55 /**
56 * @type {boolean}
57 */
58 crlfCheck = false
59
60 /**
61 * @type {boolean}
62 */
63 eventEndCheck = false
64
65 /**
66 * @type {Buffer}
67 */
68 buffer = null
69
70 pos = 0
71
72 event = {
73 data: undefined,
74 event: undefined,
75 id: undefined,
76 retry: undefined
77 }
78
79 /**
80 * @param {object} options
81 * @param {eventSourceSettings} options.eventSourceSettings
82 * @param {Function} [options.push]
83 */
84 constructor (options = {}) {
85 // Enable object mode as EventSourceStream emits objects of shape
86 // EventSourceStreamEvent
87 options.readableObjectMode = true
88
89 super(options)
90
91 this.state = options.eventSourceSettings || {}
92 if (options.push) {
93 this.push = options.push
94 }
95 }
96
97 /**
98 * @param {Buffer} chunk
99 * @param {string} _encoding
100 * @param {Function} callback
101 * @returns {void}
102 */
103 _transform (chunk, _encoding, callback) {
104 if (chunk.length === 0) {
105 callback()
106 return
107 }
108
109 // Cache the chunk in the buffer, as the data might not be complete while
110 // processing it
111 // TODO: Investigate if there is a more performant way to handle
112 // incoming chunks
113 // see: https://github.com/nodejs/undici/issues/2630
114 if (this.buffer) {
115 this.buffer = Buffer.concat([this.buffer, chunk])
116 } else {
117 this.buffer = chunk
118 }
119
120 // Strip leading byte-order-mark if we opened the stream and started
121 // the processing of the incoming data
122 if (this.checkBOM) {
123 switch (this.buffer.length) {
124 case 1:
125 // Check if the first byte is the same as the first byte of the BOM
126 if (this.buffer[0] === BOM[0]) {
127 // If it is, we need to wait for more data
128 callback()
129 return
130 }
131 // Set the checkBOM flag to false as we don't need to check for the
132 // BOM anymore
133 this.checkBOM = false
134
135 // The buffer only contains one byte so we need to wait for more data
136 callback()
137 return
138 case 2:
139 // Check if the first two bytes are the same as the first two bytes
140 // of the BOM
141 if (
142 this.buffer[0] === BOM[0] &&
143 this.buffer[1] === BOM[1]
144 ) {
145 // If it is, we need to wait for more data, because the third byte
146 // is needed to determine if it is the BOM or not
147 callback()
148 return
149 }
150
151 // Set the checkBOM flag to false as we don't need to check for the
152 // BOM anymore
153 this.checkBOM = false
154 break
155 case 3:
156 // Check if the first three bytes are the same as the first three
157 // bytes of the BOM
158 if (
159 this.buffer[0] === BOM[0] &&
160 this.buffer[1] === BOM[1] &&
161 this.buffer[2] === BOM[2]
162 ) {
163 // If it is, we can drop the buffered data, as it is only the BOM
164 this.buffer = Buffer.alloc(0)
165 // Set the checkBOM flag to false as we don't need to check for the
166 // BOM anymore
167 this.checkBOM = false
168
169 // Await more data
170 callback()
171 return
172 }
173 // If it is not the BOM, we can start processing the data
174 this.checkBOM = false
175 break
176 default:
177 // The buffer is longer than 3 bytes, so we can drop the BOM if it is
178 // present
179 if (
180 this.buffer[0] === BOM[0] &&
181 this.buffer[1] === BOM[1] &&
182 this.buffer[2] === BOM[2]
183 ) {
184 // Remove the BOM from the buffer
185 this.buffer = this.buffer.subarray(3)
186 }
187
188 // Set the checkBOM flag to false as we don't need to check for the
189 this.checkBOM = false
190 break
191 }
192 }
193
194 while (this.pos < this.buffer.length) {
195 // If the previous line ended with an end-of-line, we need to check
196 // if the next character is also an end-of-line.
197 if (this.eventEndCheck) {
198 // If the the current character is an end-of-line, then the event
199 // is finished and we can process it
200
201 // If the previous line ended with a carriage return, we need to
202 // check if the current character is a line feed and remove it
203 // from the buffer.
204 if (this.crlfCheck) {
205 // If the current character is a line feed, we can remove it
206 // from the buffer and reset the crlfCheck flag
207 if (this.buffer[this.pos] === LF) {
208 this.buffer = this.buffer.subarray(this.pos + 1)
209 this.pos = 0
210 this.crlfCheck = false
211
212 // It is possible that the line feed is not the end of the
213 // event. We need to check if the next character is an
214 // end-of-line character to determine if the event is
215 // finished. We simply continue the loop to check the next
216 // character.
217
218 // As we removed the line feed from the buffer and set the
219 // crlfCheck flag to false, we basically don't make any
220 // distinction between a line feed and a carriage return.
221 continue
222 }
223 this.crlfCheck = false
224 }
225
226 if (this.buffer[this.pos] === LF || this.buffer[this.pos] === CR) {
227 // If the current character is a carriage return, we need to
228 // set the crlfCheck flag to true, as we need to check if the
229 // next character is a line feed so we can remove it from the
230 // buffer
231 if (this.buffer[this.pos] === CR) {
232 this.crlfCheck = true
233 }
234
235 this.buffer = this.buffer.subarray(this.pos + 1)
236 this.pos = 0
237 if (
238 this.event.data !== undefined || this.event.event || this.event.id || this.event.retry) {
239 this.processEvent(this.event)
240 }
241 this.clearEvent()
242 continue
243 }
244 // If the current character is not an end-of-line, then the event
245 // is not finished and we have to reset the eventEndCheck flag
246 this.eventEndCheck = false
247 continue
248 }
249
250 // If the current character is an end-of-line, we can process the
251 // line
252 if (this.buffer[this.pos] === LF || this.buffer[this.pos] === CR) {
253 // If the current character is a carriage return, we need to
254 // set the crlfCheck flag to true, as we need to check if the
255 // next character is a line feed
256 if (this.buffer[this.pos] === CR) {
257 this.crlfCheck = true
258 }
259
260 // In any case, we can process the line as we reached an
261 // end-of-line character
262 this.parseLine(this.buffer.subarray(0, this.pos), this.event)
263
264 // Remove the processed line from the buffer
265 this.buffer = this.buffer.subarray(this.pos + 1)
266 // Reset the position as we removed the processed line from the buffer
267 this.pos = 0
268 // A line was processed and this could be the end of the event. We need
269 // to check if the next line is empty to determine if the event is
270 // finished.
271 this.eventEndCheck = true
272 continue
273 }
274
275 this.pos++
276 }
277
278 callback()
279 }
280
281 /**
282 * @param {Buffer} line
283 * @param {EventStreamEvent} event
284 */
285 parseLine (line, event) {
286 // If the line is empty (a blank line)
287 // Dispatch the event, as defined below.
288 // This will be handled in the _transform method
289 if (line.length === 0) {
290 return
291 }
292
293 // If the line starts with a U+003A COLON character (:)
294 // Ignore the line.
295 const colonPosition = line.indexOf(COLON)
296 if (colonPosition === 0) {
297 return
298 }
299
300 let field = ''
301 let value = ''
302
303 // If the line contains a U+003A COLON character (:)
304 if (colonPosition !== -1) {
305 // Collect the characters on the line before the first U+003A COLON
306 // character (:), and let field be that string.
307 // TODO: Investigate if there is a more performant way to extract the
308 // field
309 // see: https://github.com/nodejs/undici/issues/2630
310 field = line.subarray(0, colonPosition).toString('utf8')
311
312 // Collect the characters on the line after the first U+003A COLON
313 // character (:), and let value be that string.
314 // If value starts with a U+0020 SPACE character, remove it from value.
315 let valueStart = colonPosition + 1
316 if (line[valueStart] === SPACE) {
317 ++valueStart
318 }
319 // TODO: Investigate if there is a more performant way to extract the
320 // value
321 // see: https://github.com/nodejs/undici/issues/2630
322 value = line.subarray(valueStart).toString('utf8')
323
324 // Otherwise, the string is not empty but does not contain a U+003A COLON
325 // character (:)
326 } else {
327 // Process the field using the steps described below, using the whole
328 // line as the field name, and the empty string as the field value.
329 field = line.toString('utf8')
330 value = ''
331 }
332
333 // Modify the event with the field name and value. The value is also
334 // decoded as UTF-8
335 switch (field) {
336 case 'data':
337 if (event[field] === undefined) {
338 event[field] = value
339 } else {
340 event[field] += `\n${value}`
341 }
342 break
343 case 'retry':
344 if (isASCIINumber(value)) {
345 event[field] = value
346 }
347 break
348 case 'id':
349 if (isValidLastEventId(value)) {
350 event[field] = value
351 }
352 break
353 case 'event':
354 if (value.length > 0) {
355 event[field] = value
356 }
357 break
358 }
359 }
360
361 /**
362 * @param {EventSourceStreamEvent} event
363 */
364 processEvent (event) {
365 if (event.retry && isASCIINumber(event.retry)) {
366 this.state.reconnectionTime = parseInt(event.retry, 10)
367 }
368
369 if (event.id && isValidLastEventId(event.id)) {
370 this.state.lastEventId = event.id
371 }
372
373 // only dispatch event, when data is provided
374 if (event.data !== undefined) {
375 this.push({
376 type: event.event || 'message',
377 options: {
378 data: event.data,
379 lastEventId: this.state.lastEventId,
380 origin: this.state.origin
381 }
382 })
383 }
384 }
385
386 clearEvent () {
387 this.event = {
388 data: undefined,
389 event: undefined,
390 id: undefined,
391 retry: undefined
392 }
393 }
394}
395
396module.exports = {
397 EventSourceStream
398}
399