"use strict";
/*
 * Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT.
 */
Object.defineProperty(exports, "__esModule", { value: true });
exports.EventStream = void 0;
exports.discardSentinel = discardSentinel;
const LF = 0x0a;
const CR = 0x0d;
const NEWLINE_CHARS = new Set([LF, CR]);
const MESSAGE_BOUNDARIES = [
    new Uint8Array([CR, LF, CR, LF]),
    new Uint8Array([CR, CR]),
    new Uint8Array([LF, LF]),
];
class EventStream {
    constructor(init) {
        this.stream = init.stream;
        this.decoder = init.decoder;
    }
    async *[Symbol.asyncIterator]() {
        const reader = this.stream.getReader();
        let buffer = new Uint8Array([]);
        let position = 0;
        try {
            while (true) {
                const { done, value } = await reader.read();
                if (done) {
                    break;
                }
                const newBuffer = new Uint8Array(buffer.length + value.length);
                newBuffer.set(buffer);
                newBuffer.set(value, buffer.length);
                buffer = newBuffer;
                for (let i = position; i < buffer.length; i++) {
                    const boundary = findBoundary(buffer, i);
                    if (boundary == null) {
                        continue;
                    }
                    const chunk = buffer.slice(position, i);
                    position = i + boundary.length;
                    const event = parseEvent(chunk, this.decoder);
                    if (event != null) {
                        yield event;
                    }
                }
                if (position > 0) {
                    buffer = buffer.slice(position);
                    position = 0;
                }
            }
            if (buffer.length > 0) {
                const event = parseEvent(buffer, this.decoder);
                if (event != null) {
                    yield event;
                }
            }
        }
        catch (e) {
            if (e instanceof Error && e.name === "AbortError") {
                return;
            }
            throw e;
        }
        finally {
            reader.releaseLock();
        }
    }
}
exports.EventStream = EventStream;
function findBoundary(buffer, start) {
    const char1 = buffer[start];
    const char2 = buffer[start + 1];
    // Don't bother checking if the first two characters are not new line
    // characters.
    if (char1 == null
        || char2 == null
        || !NEWLINE_CHARS.has(char1)
        || !NEWLINE_CHARS.has(char2)) {
        return null;
    }
    for (const s of MESSAGE_BOUNDARIES) {
        const seq = peekSequence(start, buffer, s);
        if (seq != null) {
            return seq;
        }
    }
    return null;
}
function peekSequence(position, buffer, sequence) {
    if (sequence.length > buffer.length - position) {
        return null;
    }
    for (let i = 0; i < sequence.length; i++) {
        if (buffer[position + i] !== sequence[i]) {
            return null;
        }
    }
    return sequence;
}
function parseEvent(chunk, decoder) {
    if (!chunk.length) {
        return null;
    }
    const td = new TextDecoder();
    const raw = td.decode(chunk);
    const lines = raw.split(/\r?\n|\r/g);
    let publish = false;
    const rawEvent = {};
    for (const line of lines) {
        if (!line) {
            continue;
        }
        const delim = line.indexOf(":");
        // Lines starting with a colon are ignored.
        if (delim === 0) {
            continue;
        }
        const field = delim > 0 ? line.substring(0, delim) : "";
        let value = delim > 0 ? line.substring(delim + 1) : "";
        if (value.charAt(0) === " ") {
            value = value.substring(1);
        }
        switch (field) {
            case "event": {
                publish = true;
                rawEvent.event = value;
                break;
            }
            case "data": {
                publish = true;
                rawEvent.data ?? (rawEvent.data = "");
                rawEvent.data += value + "\n";
                break;
            }
            case "id": {
                publish = true;
                rawEvent.id = value;
                break;
            }
            case "retry": {
                const r = parseInt(value, 10);
                if (!Number.isNaN(r)) {
                    publish = true;
                    rawEvent.retry = r;
                }
                break;
            }
        }
    }
    if (!publish) {
        return null;
    }
    if (rawEvent.data != null) {
        rawEvent.data = rawEvent.data.slice(0, -1);
    }
    return decoder(rawEvent);
}
function discardSentinel(stream, sentinel) {
    return new ReadableStream({
        async start(controller) {
            let buffer = new Uint8Array([]);
            let position = 0;
            let done = false;
            let discard = false;
            const rdr = stream.getReader();
            try {
                while (!done) {
                    const result = await rdr.read();
                    const value = result.value;
                    done = done || result.done;
                    // We keep consuming from the source to its completion so it can
                    // flush all its contents and release resources.
                    if (discard) {
                        continue;
                    }
                    if (typeof value === "undefined") {
                        continue;
                    }
                    const newBuffer = new Uint8Array(buffer.length + value.length);
                    newBuffer.set(buffer);
                    newBuffer.set(value, buffer.length);
                    buffer = newBuffer;
                    for (let i = position; i < buffer.length; i++) {
                        const boundary = findBoundary(buffer, i);
                        if (boundary == null) {
                            continue;
                        }
                        const start = position;
                        const chunk = buffer.slice(start, i);
                        position = i + boundary.length;
                        const event = parseEvent(chunk, id);
                        if (event?.data === sentinel) {
                            controller.enqueue(buffer.slice(0, start));
                            discard = true;
                        }
                        else {
                            controller.enqueue(buffer.slice(0, position));
                            buffer = buffer.slice(position);
                            position = 0;
                        }
                    }
                }
            }
            catch (e) {
                controller.error(e);
            }
            finally {
                // If the source stream terminates, flush its contents and terminate.
                // If the sentinel event was found, flush everything up to its start.
                controller.close();
                rdr.releaseLock();
            }
        },
    });
}
function id(v) {
    return v;
}
//# sourceMappingURL=event-streams.js.map