{"version":3,"file":"event_source_parse.cjs","names":["IterableReadableStream"],"sources":["../../src/utils/event_source_parse.ts"],"sourcesContent":["/* oxlint-disable prefer-template */\n/* oxlint-disable default-case */\n// Adapted from https://github.com/gfortaine/fetch-event-source/blob/main/src/parse.ts\n// due to a packaging issue in the original.\n// MIT License\nimport { type Readable } from \"stream\";\nimport { IterableReadableStream } from \"@langchain/core/utils/stream\";\n\nexport const EventStreamContentType = \"text/event-stream\";\n\n/**\n * Represents a message sent in an event stream\n * https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#Event_stream_format\n */\nexport interface EventSourceMessage {\n /** The event ID to set the EventSource object's last event ID value. */\n id: string;\n /** A string identifying the type of event described. */\n event: string;\n /** The event data */\n data: string;\n /** The reconnection interval (in milliseconds) to wait before retrying the connection */\n retry?: number;\n}\n\nfunction isNodeJSReadable(x: unknown): x is Readable {\n return x != null && typeof x === \"object\" && \"on\" in x;\n}\n\n/**\n * Converts a ReadableStream into a callback pattern.\n * @param stream The input ReadableStream.\n * @param onChunk A function that will be called on each new byte chunk in the stream.\n * @returns {Promise} A promise that will be resolved when the stream closes.\n */\nexport async function getBytes(\n stream: ReadableStream,\n onChunk: (arr: Uint8Array, flush?: boolean) => void\n) {\n // stream is a Node.js Readable / PassThrough stream\n // this can happen if node-fetch is polyfilled\n if (isNodeJSReadable(stream)) {\n return new Promise((resolve) => {\n stream.on(\"readable\", () => {\n let chunk;\n while (true) {\n chunk = stream.read();\n if (chunk == null) {\n onChunk(new Uint8Array(), true);\n break;\n }\n onChunk(chunk);\n }\n\n resolve();\n });\n });\n }\n\n const reader = stream.getReader();\n while (true) {\n const result = await reader.read();\n if (result.done) {\n onChunk(new Uint8Array(), true);\n break;\n }\n onChunk(result.value);\n }\n}\n\nconst enum ControlChars {\n NewLine = 10,\n CarriageReturn = 13,\n Space = 32,\n Colon = 58,\n}\n\n/**\n * Parses arbitary byte chunks into EventSource line buffers.\n * Each line should be of the format \"field: value\" and ends with \\r, \\n, or \\r\\n.\n * @param onLine A function that will be called on each new EventSource line.\n * @returns A function that should be called for each incoming byte chunk.\n */\nexport function getLines(\n onLine: (line: Uint8Array, fieldLength: number, flush?: boolean) => void\n) {\n let buffer: Uint8Array | undefined;\n let position: number;\n let fieldLength: number;\n let discardTrailingNewline = false;\n\n return function onChunk(arr: Uint8Array, flush?: boolean) {\n if (flush) {\n onLine(arr, 0, true);\n return;\n }\n\n if (buffer === undefined) {\n buffer = arr;\n position = 0;\n fieldLength = -1;\n } else {\n buffer = concat(buffer, arr);\n }\n\n const bufLength = buffer.length;\n let lineStart = 0;\n while (position < bufLength) {\n if (discardTrailingNewline) {\n if (buffer[position] === ControlChars.NewLine) {\n lineStart = ++position;\n }\n\n discardTrailingNewline = false;\n }\n\n let lineEnd = -1;\n for (; position < bufLength && lineEnd === -1; ++position) {\n switch (buffer[position]) {\n case ControlChars.Colon:\n if (fieldLength === -1) {\n fieldLength = position - lineStart;\n }\n break;\n // oxlint-disable-next-line @typescript-eslint/ban-ts-comment\n // @ts-ignore:7029 \\r case below should fallthrough to \\n:\n case ControlChars.CarriageReturn:\n discardTrailingNewline = true;\n // oxlint-disable-next-line no-fallthrough\n case ControlChars.NewLine:\n lineEnd = position;\n break;\n }\n }\n\n if (lineEnd === -1) {\n break;\n }\n\n onLine(buffer.subarray(lineStart, lineEnd), fieldLength);\n lineStart = position;\n fieldLength = -1;\n }\n\n if (lineStart === bufLength) {\n buffer = undefined;\n } else if (lineStart !== 0) {\n buffer = buffer.subarray(lineStart);\n position -= lineStart;\n }\n };\n}\n\n/**\n * Parses line buffers into EventSourceMessages.\n * @param onId A function that will be called on each `id` field.\n * @param onRetry A function that will be called on each `retry` field.\n * @param onMessage A function that will be called on each message.\n * @returns A function that should be called for each incoming line buffer.\n */\nexport function getMessages(\n onMessage?: (msg: EventSourceMessage) => void,\n onId?: (id: string) => void,\n onRetry?: (retry: number) => void\n) {\n let message = newMessage();\n const decoder = new TextDecoder();\n\n return function onLine(\n line: Uint8Array,\n fieldLength: number,\n flush?: boolean\n ) {\n if (flush) {\n if (!isEmpty(message)) {\n onMessage?.(message);\n message = newMessage();\n }\n return;\n }\n\n if (line.length === 0) {\n onMessage?.(message);\n message = newMessage();\n } else if (fieldLength > 0) {\n const field = decoder.decode(line.subarray(0, fieldLength));\n const valueOffset =\n fieldLength + (line[fieldLength + 1] === ControlChars.Space ? 2 : 1);\n const value = decoder.decode(line.subarray(valueOffset));\n\n switch (field) {\n case \"data\":\n message.data = message.data ? `${message.data}\\n${value}` : value;\n break;\n case \"event\":\n message.event = value;\n break;\n case \"id\":\n onId?.((message.id = value));\n break;\n case \"retry\": {\n const retry = parseInt(value, 10);\n if (!Number.isNaN(retry)) {\n onRetry?.((message.retry = retry));\n }\n break;\n }\n }\n }\n };\n}\n\nfunction concat(a: Uint8Array, b: Uint8Array) {\n const res = new Uint8Array(a.length + b.length);\n res.set(a);\n res.set(b, a.length);\n return res;\n}\n\nfunction newMessage(): EventSourceMessage {\n return {\n data: \"\",\n event: \"\",\n id: \"\",\n retry: undefined,\n };\n}\n\nexport function convertEventStreamToIterableReadableDataStream(\n stream: ReadableStream\n) {\n const dataStream = new ReadableStream({\n async start(controller) {\n const enqueueLine = getMessages((msg) => {\n if (msg.data) controller.enqueue(msg.data);\n });\n const onLine = (\n line: Uint8Array,\n fieldLength: number,\n flush?: boolean\n ) => {\n enqueueLine(line, fieldLength, flush);\n if (flush) controller.close();\n };\n await getBytes(stream, getLines(onLine));\n },\n });\n return IterableReadableStream.fromReadableStream(dataStream);\n}\n\nfunction isEmpty(message: EventSourceMessage): boolean {\n return (\n message.data === \"\" &&\n message.event === \"\" &&\n message.id === \"\" &&\n message.retry === undefined\n );\n}\n"],"mappings":";;AAyBA,SAAS,iBAAiB,GAA2B;CACnD,OAAO,KAAK,QAAQ,OAAO,MAAM,YAAY,QAAQ;AACvD;;;;;;;AAQA,eAAsB,SACpB,QACA,SACA;CAGA,IAAI,iBAAiB,MAAM,GACzB,OAAO,IAAI,SAAe,YAAY;EACpC,OAAO,GAAG,kBAAkB;GAC1B,IAAI;GACJ,OAAO,MAAM;IACX,QAAQ,OAAO,KAAK;IACpB,IAAI,SAAS,MAAM;KACjB,wBAAQ,IAAI,WAAW,GAAG,IAAI;KAC9B;IACF;IACA,QAAQ,KAAK;GACf;GAEA,QAAQ;EACV,CAAC;CACH,CAAC;CAGH,MAAM,SAAS,OAAO,UAAU;CAChC,OAAO,MAAM;EACX,MAAM,SAAS,MAAM,OAAO,KAAK;EACjC,IAAI,OAAO,MAAM;GACf,wBAAQ,IAAI,WAAW,GAAG,IAAI;GAC9B;EACF;EACA,QAAQ,OAAO,KAAK;CACtB;AACF;;;;;;;AAeA,SAAgB,SACd,QACA;CACA,IAAI;CACJ,IAAI;CACJ,IAAI;CACJ,IAAI,yBAAyB;CAE7B,OAAO,SAAS,QAAQ,KAAiB,OAAiB;EACxD,IAAI,OAAO;GACT,OAAO,KAAK,GAAG,IAAI;GACnB;EACF;EAEA,IAAI,WAAW,KAAA,GAAW;GACxB,SAAS;GACT,WAAW;GACX,cAAc;EAChB,OACE,SAAS,OAAO,QAAQ,GAAG;EAG7B,MAAM,YAAY,OAAO;EACzB,IAAI,YAAY;EAChB,OAAO,WAAW,WAAW;GAC3B,IAAI,wBAAwB;IAC1B,IAAI,OAAO,cAAA,IACT,YAAY,EAAE;IAGhB,yBAAyB;GAC3B;GAEA,IAAI,UAAU;GACd,OAAO,WAAW,aAAa,YAAY,IAAI,EAAE,UAC/C,QAAQ,OAAO,WAAf;IACE,KAAA;KACE,IAAI,gBAAgB,IAClB,cAAc,WAAW;KAE3B;IAGF,KAAA,IACE,yBAAyB;IAE3B,KAAA,IACE,UAAU;GAEd;GAGF,IAAI,YAAY,IACd;GAGF,OAAO,OAAO,SAAS,WAAW,OAAO,GAAG,WAAW;GACvD,YAAY;GACZ,cAAc;EAChB;EAEA,IAAI,cAAc,WAChB,SAAS,KAAA;OACJ,IAAI,cAAc,GAAG;GAC1B,SAAS,OAAO,SAAS,SAAS;GAClC,YAAY;EACd;CACF;AACF;;;;;;;;AASA,SAAgB,YACd,WACA,MACA,SACA;CACA,IAAI,UAAU,WAAW;CACzB,MAAM,UAAU,IAAI,YAAY;CAEhC,OAAO,SAAS,OACd,MACA,aACA,OACA;EACA,IAAI,OAAO;GACT,IAAI,CAAC,QAAQ,OAAO,GAAG;IACrB,YAAY,OAAO;IACnB,UAAU,WAAW;GACvB;GACA;EACF;EAEA,IAAI,KAAK,WAAW,GAAG;GACrB,YAAY,OAAO;GACnB,UAAU,WAAW;EACvB,OAAO,IAAI,cAAc,GAAG;GAC1B,MAAM,QAAQ,QAAQ,OAAO,KAAK,SAAS,GAAG,WAAW,CAAC;GAC1D,MAAM,cACJ,eAAe,KAAK,cAAc,OAAA,KAA4B,IAAI;GACpE,MAAM,QAAQ,QAAQ,OAAO,KAAK,SAAS,WAAW,CAAC;GAEvD,QAAQ,OAAR;IACE,KAAK;KACH,QAAQ,OAAO,QAAQ,OAAO,GAAG,QAAQ,KAAK,IAAI,UAAU;KAC5D;IACF,KAAK;KACH,QAAQ,QAAQ;KAChB;IACF,KAAK;KACH,OAAQ,QAAQ,KAAK,KAAM;KAC3B;IACF,KAAK,SAAS;KACZ,MAAM,QAAQ,SAAS,OAAO,EAAE;KAChC,IAAI,CAAC,OAAO,MAAM,KAAK,GACrB,UAAW,QAAQ,QAAQ,KAAM;KAEnC;IACF;GACF;EACF;CACF;AACF;AAEA,SAAS,OAAO,GAAe,GAAe;CAC5C,MAAM,MAAM,IAAI,WAAW,EAAE,SAAS,EAAE,MAAM;CAC9C,IAAI,IAAI,CAAC;CACT,IAAI,IAAI,GAAG,EAAE,MAAM;CACnB,OAAO;AACT;AAEA,SAAS,aAAiC;CACxC,OAAO;EACL,MAAM;EACN,OAAO;EACP,IAAI;EACJ,OAAO,KAAA;CACT;AACF;AAEA,SAAgB,+CACd,QACA;CACA,MAAM,aAAa,IAAI,eAAe,EACpC,MAAM,MAAM,YAAY;EACtB,MAAM,cAAc,aAAa,QAAQ;GACvC,IAAI,IAAI,MAAM,WAAW,QAAQ,IAAI,IAAI;EAC3C,CAAC;EACD,MAAM,UACJ,MACA,aACA,UACG;GACH,YAAY,MAAM,aAAa,KAAK;GACpC,IAAI,OAAO,WAAW,MAAM;EAC9B;EACA,MAAM,SAAS,QAAQ,SAAS,MAAM,CAAC;CACzC,EACF,CAAC;CACD,OAAOA,6BAAAA,uBAAuB,mBAAmB,UAAU;AAC7D;AAEA,SAAS,QAAQ,SAAsC;CACrD,OACE,QAAQ,SAAS,MACjB,QAAQ,UAAU,MAClB,QAAQ,OAAO,MACf,QAAQ,UAAU,KAAA;AAEtB"}