Fix missing headers in SSE queries

This commit is contained in:
Ian Renton
2026-08-14 23:47:22 +01:00
parent efec8e220e
commit ef2c23c8b6
16 changed files with 358 additions and 129 deletions
+89
View File
@@ -0,0 +1,89 @@
var __rest = (this && this.__rest) || function (s, e) {
var t = {};
for (var p in s) if (Object.prototype.hasOwnProperty.call(s, p) && e.indexOf(p) < 0)
t[p] = s[p];
if (s != null && typeof Object.getOwnPropertySymbols === "function")
for (var i = 0, p = Object.getOwnPropertySymbols(s); i < p.length; i++) {
if (e.indexOf(p[i]) < 0 && Object.prototype.propertyIsEnumerable.call(s, p[i]))
t[p[i]] = s[p[i]];
}
return t;
};
import { getBytes, getLines, getMessages } from './parse.js';
export const EventStreamContentType = 'text/event-stream';
const DefaultRetryInterval = 1000;
const LastEventId = 'last-event-id';
export function fetchEventSource(input, _a) {
var { signal: inputSignal, headers: inputHeaders, onopen: inputOnOpen, onmessage, onclose, onerror, openWhenHidden, fetch: inputFetch } = _a, rest = __rest(_a, ["signal", "headers", "onopen", "onmessage", "onclose", "onerror", "openWhenHidden", "fetch"]);
return new Promise((resolve, reject) => {
const headers = Object.assign({}, inputHeaders);
if (!headers.accept) {
headers.accept = EventStreamContentType;
}
let curRequestController;
function onVisibilityChange() {
curRequestController.abort();
if (!document.hidden) {
create();
}
}
if (!openWhenHidden) {
document.addEventListener('visibilitychange', onVisibilityChange);
}
let retryInterval = DefaultRetryInterval;
let retryTimer = 0;
function dispose() {
document.removeEventListener('visibilitychange', onVisibilityChange);
window.clearTimeout(retryTimer);
curRequestController.abort();
}
inputSignal === null || inputSignal === void 0 ? void 0 : inputSignal.addEventListener('abort', () => {
dispose();
resolve();
});
const fetch = inputFetch !== null && inputFetch !== void 0 ? inputFetch : window.fetch;
const onopen = inputOnOpen !== null && inputOnOpen !== void 0 ? inputOnOpen : defaultOnOpen;
async function create() {
var _a;
curRequestController = new AbortController();
try {
const response = await fetch(input, Object.assign(Object.assign({}, rest), { headers, signal: curRequestController.signal }));
await onopen(response);
await getBytes(response.body, getLines(getMessages(id => {
if (id) {
headers[LastEventId] = id;
}
else {
delete headers[LastEventId];
}
}, retry => {
retryInterval = retry;
}, onmessage)));
onclose === null || onclose === void 0 ? void 0 : onclose();
dispose();
resolve();
}
catch (err) {
if (!curRequestController.signal.aborted) {
try {
const interval = (_a = onerror === null || onerror === void 0 ? void 0 : onerror(err)) !== null && _a !== void 0 ? _a : retryInterval;
window.clearTimeout(retryTimer);
retryTimer = window.setTimeout(create, interval);
}
catch (innerErr) {
dispose();
reject(innerErr);
}
}
}
}
create();
});
}
function defaultOnOpen(response) {
const contentType = response.headers.get('content-type');
if (!(contentType === null || contentType === void 0 ? void 0 : contentType.startsWith(EventStreamContentType))) {
throw new Error(`Expected content-type to be ${EventStreamContentType}, Actual: ${contentType}`);
}
}
//# sourceMappingURL=fetch.js.map
+2
View File
@@ -0,0 +1,2 @@
export { fetchEventSource, EventStreamContentType } from './fetch.js';
//# sourceMappingURL=index.js.map
+110
View File
@@ -0,0 +1,110 @@
export async function getBytes(stream, onChunk) {
const reader = stream.getReader();
let result;
while (!(result = await reader.read()).done) {
onChunk(result.value);
}
}
export function getLines(onLine) {
let buffer;
let position;
let fieldLength;
let discardTrailingNewline = false;
return function onChunk(arr) {
if (buffer === undefined) {
buffer = arr;
position = 0;
fieldLength = -1;
}
else {
buffer = concat(buffer, arr);
}
const bufLength = buffer.length;
let lineStart = 0;
while (position < bufLength) {
if (discardTrailingNewline) {
if (buffer[position] === 10) {
lineStart = ++position;
}
discardTrailingNewline = false;
}
let lineEnd = -1;
for (; position < bufLength && lineEnd === -1; ++position) {
switch (buffer[position]) {
case 58:
if (fieldLength === -1) {
fieldLength = position - lineStart;
}
break;
case 13:
discardTrailingNewline = true;
case 10:
lineEnd = position;
break;
}
}
if (lineEnd === -1) {
break;
}
onLine(buffer.subarray(lineStart, lineEnd), fieldLength);
lineStart = position;
fieldLength = -1;
}
if (lineStart === bufLength) {
buffer = undefined;
}
else if (lineStart !== 0) {
buffer = buffer.subarray(lineStart);
position -= lineStart;
}
};
}
export function getMessages(onId, onRetry, onMessage) {
let message = newMessage();
const decoder = new TextDecoder();
return function onLine(line, fieldLength) {
if (line.length === 0) {
onMessage === null || onMessage === void 0 ? void 0 : onMessage(message);
message = newMessage();
}
else if (fieldLength > 0) {
const field = decoder.decode(line.subarray(0, fieldLength));
const valueOffset = fieldLength + (line[fieldLength + 1] === 32 ? 2 : 1);
const value = decoder.decode(line.subarray(valueOffset));
switch (field) {
case 'data':
message.data = message.data
? message.data + '\n' + value
: value;
break;
case 'event':
message.event = value;
break;
case 'id':
onId(message.id = value);
break;
case 'retry':
const retry = parseInt(value, 10);
if (!isNaN(retry)) {
onRetry(message.retry = retry);
}
break;
}
}
};
}
function concat(a, b) {
const res = new Uint8Array(a.length + b.length);
res.set(a);
res.set(b, a.length);
return res;
}
function newMessage() {
return {
data: '',
event: '',
id: '',
retry: undefined,
};
}
//# sourceMappingURL=parse.js.map