| 181 | |
| 182 | super({ |
| 183 | transform(chunk, controller) { |
| 184 | buffer += decoder.decode(chunk, { stream: true }); |
| 185 | |
| 186 | // Preserve trailing \r - it might be part of \r\n split across chunks |
| 187 | let trailingCR = ""; |
| 188 | if (buffer[buffer.length - 1] === "\r") { |
| 189 | trailingCR = "\r"; |
| 190 | buffer = buffer.slice(0, -1); |
| 191 | } |
| 192 | |
| 193 | // Process complete lines |
| 194 | const lines = buffer.split(NEWLINE_REGEXP); |
| 195 | // Keep incomplete last line in buffer, restore any trailing \r |
| 196 | buffer = lines.pop()! + trailingCR; |
| 197 | |
| 198 | for (const line of lines) { |
| 199 | if (line === "") { |
| 200 | // Empty line signals end of message - dispatch if non-empty |
| 201 | if (hasFields) { |
| 202 | controller.enqueue(message); |
| 203 | } |
| 204 | message = {}; |
| 205 | hasFields = false; |
| 206 | } else if (parseLine(line, message, ignoreComments)) { |
| 207 | hasFields = true; |
| 208 | } |
| 209 | } |
| 210 | }, |
| 211 | |
| 212 | flush(controller) { |
| 213 | // Handle any remaining content in buffer |