parse-chunked.js 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569
  1. import { isIterable } from './utils.js';
  2. const NO_VALUE = Symbol('empty');
  3. const STACK_OBJECT = 1;
  4. const STACK_ARRAY = 2;
  5. const MODE_JSON = 0;
  6. const MODE_JSONL = 1;
  7. const MODE_JSONL_AUTO = 2;
  8. const decoder = new TextDecoder();
  9. function adjustPosition(error, jsonParseOffset) {
  10. if (error.name === 'SyntaxError' && jsonParseOffset) {
  11. error.message = error.message.replace(/at position (\d+)/, (_, pos) =>
  12. 'at position ' + (Number(pos) + jsonParseOffset)
  13. );
  14. }
  15. return error;
  16. }
  17. function append(array, elements) {
  18. // Note: Avoid using array.push(...elements) since it may lead to
  19. // "RangeError: Maximum call stack size exceeded" for long arrays
  20. const initialLength = array.length;
  21. array.length += elements.length;
  22. for (let i = 0; i < elements.length; i++) {
  23. array[initialLength + i] = elements[i];
  24. }
  25. }
  26. function resolveParseMode(mode) {
  27. switch (mode) {
  28. case 'json':
  29. return MODE_JSON;
  30. case 'jsonl':
  31. return MODE_JSONL;
  32. case 'auto':
  33. return MODE_JSONL_AUTO;
  34. default:
  35. throw new TypeError('Invalid options: `mode` should be "json", "jsonl", or "auto"');
  36. }
  37. }
  38. function parseChunkedOptions(value) {
  39. const options = typeof value === 'function'
  40. ? { reviver: value }
  41. : value || {};
  42. return {
  43. mode: resolveParseMode(options.mode ?? 'json'),
  44. reviver: options.reviver ?? null,
  45. onRootValue: options.onRootValue ?? null,
  46. onChunk: options.onChunk ?? null
  47. };
  48. }
  49. function applyReviver(value, reviver) {
  50. return walk({ '': value }, '', value);
  51. function walk(holder, key, value) {
  52. if (value && typeof value === 'object') {
  53. for (const childKey of Object.keys(value)) {
  54. const childValue = value[childKey];
  55. const newValue = walk(value, childKey, childValue);
  56. if (newValue === undefined) {
  57. delete value[childKey];
  58. } else if (newValue !== childValue) {
  59. value[childKey] = newValue;
  60. }
  61. }
  62. }
  63. return reviver.call(holder, key, value);
  64. }
  65. }
  66. export async function parseChunked(chunkEmitter, optionsOrReviver) {
  67. const { mode, reviver, onRootValue, onChunk } = parseChunkedOptions(optionsOrReviver);
  68. const iterable = typeof chunkEmitter === 'function'
  69. ? chunkEmitter()
  70. : chunkEmitter;
  71. if (isIterable(iterable)) {
  72. const parser = createChunkParser(mode, reviver, onRootValue, onChunk);
  73. try {
  74. for await (const chunk of iterable) {
  75. if (typeof chunk !== 'string' && !ArrayBuffer.isView(chunk)) {
  76. throw new TypeError('Invalid chunk: Expected string, TypedArray or Buffer');
  77. }
  78. parser.push(chunk);
  79. }
  80. return parser.finish();
  81. } catch (e) {
  82. throw adjustPosition(e, parser.jsonParseOffset);
  83. }
  84. }
  85. throw new TypeError(
  86. 'Invalid chunk emitter: Expected an Iterable, AsyncIterable, generator, ' +
  87. 'async generator, or a function returning an Iterable or AsyncIterable'
  88. );
  89. };
  90. function createChunkParser(parseMode, reviver, onRootValue, onChunk) {
  91. let rootValues = parseMode === MODE_JSONL ? [] : null;
  92. let rootValuesCount = 0;
  93. let currentRootValue = NO_VALUE;
  94. let currentRootValueCursor = null;
  95. let consumedChunkLength = 0;
  96. let parsedChunkLength = 0;
  97. let prevArray = null;
  98. let prevArraySlices = [];
  99. let stack = new Array(100);
  100. let lastFlushDepth = 0;
  101. let flushDepth = 0;
  102. let stateString = false;
  103. let stateStringEscape = false;
  104. let seenNonWhiteSpace = false;
  105. let allowNewRootValue = true;
  106. let pendingByteSeq = null;
  107. let pendingChunk = null;
  108. let jsonParseOffset = 0;
  109. const state = Object.freeze({
  110. get mode() {
  111. return parseMode === MODE_JSONL ? 'jsonl' : 'json';
  112. },
  113. get returnValue() {
  114. return typeof onRootValue === 'function'
  115. ? rootValuesCount
  116. : rootValues !== null
  117. ? rootValues
  118. : currentRootValue !== NO_VALUE
  119. ? currentRootValue
  120. : undefined;
  121. },
  122. get currentRootValue() {
  123. return currentRootValue !== NO_VALUE ? currentRootValue : undefined;
  124. },
  125. get rootValuesCount() {
  126. return rootValuesCount;
  127. },
  128. get consumed() {
  129. return consumedChunkLength;
  130. },
  131. get parsed() {
  132. return parsedChunkLength;
  133. }
  134. });
  135. return {
  136. push,
  137. finish,
  138. state,
  139. get jsonParseOffset() {
  140. return jsonParseOffset;
  141. }
  142. };
  143. function startRootValue(fragment) {
  144. // Extra non-whitespace after complete root value should fail to parse
  145. if (!allowNewRootValue) {
  146. jsonParseOffset -= 2;
  147. JSON.parse('[]' + fragment);
  148. }
  149. // In "auto" mode, switch to JSONL when a second root value is starting after a newline
  150. if (currentRootValue !== NO_VALUE && parseMode === MODE_JSONL_AUTO) {
  151. parseMode = MODE_JSONL;
  152. rootValues = [currentRootValue];
  153. }
  154. // Block parsing of an additional root value until a newline is encountered
  155. allowNewRootValue = false;
  156. // Parse fragment as a new root value
  157. currentRootValue = JSON.parse(fragment);
  158. }
  159. function finishRootValue() {
  160. rootValuesCount++;
  161. if (typeof reviver === 'function') {
  162. currentRootValue = applyReviver(currentRootValue, reviver);
  163. }
  164. if (typeof onRootValue === 'function') {
  165. onRootValue(currentRootValue, state);
  166. } else if (parseMode === MODE_JSONL) {
  167. rootValues.push(currentRootValue);
  168. }
  169. }
  170. function mergeArraySlices() {
  171. if (prevArray === null) {
  172. return;
  173. }
  174. if (prevArraySlices.length !== 0) {
  175. const newArray = prevArraySlices.length === 1
  176. ? prevArray.concat(prevArraySlices[0])
  177. : prevArray.concat(...prevArraySlices);
  178. if (currentRootValueCursor.prev !== null) {
  179. currentRootValueCursor.prev.value[currentRootValueCursor.key] = newArray;
  180. } else {
  181. currentRootValue = newArray;
  182. }
  183. currentRootValueCursor.value = newArray;
  184. prevArraySlices = [];
  185. }
  186. prevArray = null;
  187. }
  188. function parseAndAppend(fragment, wrap) {
  189. // Append new entries or elements
  190. if (stack[lastFlushDepth - 1] === STACK_OBJECT) {
  191. if (wrap) {
  192. jsonParseOffset--;
  193. fragment = '{' + fragment + '}';
  194. }
  195. Object.assign(currentRootValueCursor.value, JSON.parse(fragment));
  196. } else {
  197. if (wrap) {
  198. jsonParseOffset--;
  199. fragment = '[' + fragment + ']';
  200. }
  201. if (prevArray === currentRootValueCursor.value) {
  202. prevArraySlices.push(JSON.parse(fragment));
  203. } else {
  204. append(currentRootValueCursor.value, JSON.parse(fragment));
  205. prevArray = currentRootValueCursor.value;
  206. }
  207. }
  208. }
  209. function prepareAddition(fragment) {
  210. const { value } = currentRootValueCursor;
  211. const expectComma = Array.isArray(value)
  212. ? value.length !== 0
  213. : Object.keys(value).length !== 0;
  214. if (expectComma) {
  215. // Skip a comma at the beginning of fragment, otherwise it would
  216. // fail to parse
  217. if (fragment[0] === ',') {
  218. jsonParseOffset++;
  219. return fragment.slice(1);
  220. }
  221. // When value (an object or array) is not empty and a fragment
  222. // doesn't start with a comma, a single valid fragment starting
  223. // is a closing bracket. If it's not, a prefix is adding to fail
  224. // parsing. Otherwise, the sequence of chunks can be successfully
  225. // parsed, although it should not, e.g. ["[{}", "{}]"]
  226. if (fragment[0] !== '}' && fragment[0] !== ']') {
  227. jsonParseOffset -= 3;
  228. return '[[]' + fragment;
  229. }
  230. }
  231. return fragment;
  232. }
  233. function flush(chunk, start, end) {
  234. let fragment = chunk.slice(start, end);
  235. // Save position correction for an error in JSON.parse() if any
  236. jsonParseOffset = consumedChunkLength + start;
  237. parsedChunkLength += end - start;
  238. // Prepend pending chunk if any
  239. if (pendingChunk !== null) {
  240. fragment = pendingChunk + fragment;
  241. jsonParseOffset -= pendingChunk.length;
  242. parsedChunkLength += pendingChunk.length;
  243. pendingChunk = null;
  244. }
  245. if (flushDepth === lastFlushDepth) {
  246. // Depth didn't change, so it's a continuation of the current value or entire value if it's a root one
  247. if (lastFlushDepth === 0) {
  248. startRootValue(fragment);
  249. } else {
  250. parseAndAppend(prepareAddition(fragment), true);
  251. }
  252. } else if (flushDepth > lastFlushDepth) {
  253. // Add missed closing brackets/parentheses
  254. for (let i = flushDepth - 1; i >= lastFlushDepth; i--) {
  255. fragment += stack[i] === STACK_OBJECT ? '}' : ']';
  256. }
  257. if (lastFlushDepth === 0) {
  258. startRootValue(fragment);
  259. currentRootValueCursor = {
  260. value: currentRootValue,
  261. key: null,
  262. prev: null
  263. };
  264. } else {
  265. parseAndAppend(prepareAddition(fragment), true);
  266. mergeArraySlices();
  267. }
  268. // Move down to the depths to the last object/array, which is current now
  269. for (let i = lastFlushDepth || 1; i < flushDepth; i++) {
  270. let { value } = currentRootValueCursor;
  271. let key = null;
  272. if (stack[i - 1] === STACK_OBJECT) {
  273. // Find last entry
  274. // eslint-disable-next-line curly
  275. for (key in value);
  276. value = value[key];
  277. } else {
  278. // Last element
  279. key = value.length - 1;
  280. value = value[key];
  281. }
  282. currentRootValueCursor = {
  283. value,
  284. key,
  285. prev: currentRootValueCursor
  286. };
  287. }
  288. } else /* flushDepth < lastFlushDepth */ {
  289. fragment = prepareAddition(fragment);
  290. // Add missed opening brackets/parentheses
  291. for (let i = lastFlushDepth - 1; i >= flushDepth; i--) {
  292. jsonParseOffset--;
  293. fragment = (stack[i] === STACK_OBJECT ? '{' : '[') + fragment;
  294. }
  295. parseAndAppend(fragment, false);
  296. mergeArraySlices();
  297. for (let i = lastFlushDepth - 1; i >= flushDepth; i--) {
  298. currentRootValueCursor = currentRootValueCursor.prev;
  299. }
  300. }
  301. if (flushDepth === 0) {
  302. finishRootValue();
  303. }
  304. lastFlushDepth = flushDepth;
  305. seenNonWhiteSpace = false;
  306. }
  307. function ensureChunkString(chunk) {
  308. if (typeof chunk !== 'string') {
  309. // Suppose chunk is Buffer or Uint8Array
  310. // Prepend uncompleted byte sequence if any
  311. if (pendingByteSeq !== null) {
  312. const origRawChunk = chunk;
  313. chunk = new Uint8Array(pendingByteSeq.length + origRawChunk.length);
  314. chunk.set(pendingByteSeq);
  315. chunk.set(origRawChunk, pendingByteSeq.length);
  316. pendingByteSeq = null;
  317. }
  318. // In case Buffer/Uint8Array, an input is encoded in UTF8
  319. // Seek for parts of uncompleted UTF8 symbol on the ending
  320. // This makes sense only if we expect more chunks and last char is not multi-bytes
  321. if (chunk[chunk.length - 1] > 127) {
  322. for (let seqLength = 0; seqLength < chunk.length; seqLength++) {
  323. const byte = chunk[chunk.length - 1 - seqLength];
  324. // 10xxxxxx - 2nd, 3rd or 4th byte
  325. // 110xxxxx – first byte of 2-byte sequence
  326. // 1110xxxx - first byte of 3-byte sequence
  327. // 11110xxx - first byte of 4-byte sequence
  328. if (byte >> 6 === 3) {
  329. seqLength++;
  330. // If the sequence is really incomplete, then preserve it
  331. // for the future chunk and cut off it from the current chunk
  332. if ((seqLength !== 4 && byte >> 3 === 0b11110) ||
  333. (seqLength !== 3 && byte >> 4 === 0b1110) ||
  334. (seqLength !== 2 && byte >> 5 === 0b110)) {
  335. pendingByteSeq = chunk.slice(chunk.length - seqLength); // use slice to avoid tying chunk
  336. chunk = chunk.subarray(0, -seqLength); // use subarray to avoid buffer copy
  337. }
  338. break;
  339. }
  340. }
  341. }
  342. // Convert chunk to a string, since single decode per chunk
  343. // is much effective than decode multiple small substrings
  344. chunk = decoder.decode(chunk);
  345. }
  346. return chunk;
  347. }
  348. function push(chunk) {
  349. chunk = ensureChunkString(chunk);
  350. const chunkLength = chunk.length;
  351. const prevParsedChunkLength = parsedChunkLength;
  352. let lastFlushPoint = 0;
  353. let flushPoint = 0;
  354. // Main scan loop
  355. scan: for (let i = 0; i < chunkLength; i++) {
  356. if (stateString) {
  357. for (; i < chunkLength; i++) {
  358. if (stateStringEscape) {
  359. stateStringEscape = false;
  360. } else {
  361. switch (chunk.charCodeAt(i)) {
  362. case 0x22: /* " */
  363. stateString = false;
  364. continue scan;
  365. case 0x5C: /* \ */
  366. stateStringEscape = true;
  367. }
  368. }
  369. }
  370. break;
  371. }
  372. switch (chunk.charCodeAt(i)) {
  373. case 0x22: /* " */
  374. stateString = true;
  375. stateStringEscape = false;
  376. seenNonWhiteSpace = true;
  377. break;
  378. case 0x2C: /* , */
  379. flushPoint = i;
  380. break;
  381. case 0x7B: /* { */
  382. // Open an object
  383. flushPoint = i + 1;
  384. stack[flushDepth++] = STACK_OBJECT;
  385. seenNonWhiteSpace = true;
  386. break;
  387. case 0x5B: /* [ */
  388. // Open an array
  389. flushPoint = i + 1;
  390. stack[flushDepth++] = STACK_ARRAY;
  391. seenNonWhiteSpace = true;
  392. break;
  393. case 0x5D: /* ] */
  394. case 0x7D: /* } */
  395. // Close an object or array
  396. flushPoint = i + 1;
  397. if (flushDepth === 0) {
  398. // Unmatched closing bracket/brace at top level, should fail to parse
  399. break scan;
  400. }
  401. flushDepth--;
  402. // Flush on depth decrease related to last flush, otherwise wait for more chunks to flush together
  403. if (flushDepth < lastFlushDepth) {
  404. flush(chunk, lastFlushPoint, flushPoint);
  405. lastFlushPoint = flushPoint;
  406. }
  407. break;
  408. case 0x09: /* \t */
  409. case 0x0A: /* \n */
  410. case 0x0D: /* \r */
  411. case 0x20: /* space */
  412. if (flushDepth === 0) {
  413. if (seenNonWhiteSpace) {
  414. flushPoint = i;
  415. flush(chunk, lastFlushPoint, flushPoint);
  416. lastFlushPoint = flushPoint;
  417. }
  418. if (parseMode !== MODE_JSON &&
  419. allowNewRootValue === false &&
  420. (chunk.charCodeAt(i) === 0x0A || chunk.charCodeAt(i) === 0x0D)
  421. ) {
  422. allowNewRootValue = true;
  423. }
  424. if (flushPoint === i) {
  425. parsedChunkLength++;
  426. }
  427. }
  428. // Move points forward when they point to current position and it's a whitespace
  429. if (lastFlushPoint === i) {
  430. lastFlushPoint++;
  431. }
  432. if (flushPoint === i) {
  433. flushPoint++;
  434. }
  435. break;
  436. default:
  437. seenNonWhiteSpace = true;
  438. }
  439. }
  440. if (flushPoint > lastFlushPoint) {
  441. flush(chunk, lastFlushPoint, flushPoint);
  442. }
  443. // Produce pendingChunk if something left
  444. if (flushPoint < chunkLength) {
  445. if (pendingChunk !== null) {
  446. // When there is already a pending chunk then no flush happened,
  447. // appending entire chunk to pending one
  448. pendingChunk += chunk;
  449. } else {
  450. // Create a pending chunk, it will start with non-whitespace since
  451. // flushPoint was moved forward away from whitespaces on scan
  452. pendingChunk = chunk.slice(flushPoint, chunkLength);
  453. }
  454. }
  455. consumedChunkLength += chunkLength;
  456. if (typeof onChunk === 'function') {
  457. onChunk(parsedChunkLength - prevParsedChunkLength, chunk, pendingChunk, state);
  458. }
  459. }
  460. function finish() {
  461. if (pendingChunk !== null || (currentRootValue === NO_VALUE && parseMode !== MODE_JSONL)) {
  462. // Force the `flushDepth < lastFlushDepth` branch in flush() to prepend missed
  463. // opening brackets/parentheses and produce a natural JSON.parse() EOF error
  464. flushDepth = 0;
  465. flush('', 0, 0);
  466. }
  467. if (typeof onChunk === 'function') {
  468. parsedChunkLength = consumedChunkLength;
  469. onChunk(0, null, null, state);
  470. }
  471. const result = state.returnValue;
  472. rootValues = null;
  473. currentRootValue = NO_VALUE;
  474. return result;
  475. }
  476. }