Eric Allam
02d2334c8a
Pool Redis connections for non-blocking ops (ingestData, appendPart,
getLastChunkIndex)
using a shared singleton instead of new Redis() per request. Use
redis.disconnect()
for immediate teardown in streamResponse cleanup. Add 15s inactivity
timeout fallback.
Fix broken request.signal in Remix/Express by wiring Express
res.on('close') to an
AbortController via httpAsyncStorage. All SSE/streaming routes now use
getRequestAbortSignal() which fires reliably on client disconnect,
bypassing the
Node.js undici GC bug (nodejs/node#55428) that severs the signal chain.
88 lines
2.2 KiB
TypeScript
88 lines
2.2 KiB
TypeScript
import { eventStream } from "remix-utils/sse/server";
|
|
import { env } from "~/env.server";
|
|
import { getRequestAbortSignal } from "~/services/httpAsyncStorage.server";
|
|
import { logger } from "~/services/logger.server";
|
|
|
|
type SseProps = {
|
|
request: Request;
|
|
pingInterval?: number;
|
|
updateInterval?: number;
|
|
run: (send: (event: Event) => void, stop: () => void) => void;
|
|
};
|
|
|
|
type Event = {
|
|
/**
|
|
* @default "update"
|
|
*/
|
|
event?: string;
|
|
data: string;
|
|
};
|
|
|
|
export function sse({ request, pingInterval = 1000, updateInterval = 348, run }: SseProps) {
|
|
if (env.DISABLE_SSE === "1" || env.DISABLE_SSE === "true") {
|
|
return new Response("SSE disabled", { status: 200 });
|
|
}
|
|
|
|
const signal = getRequestAbortSignal();
|
|
|
|
let pinger: NodeJS.Timeout | undefined = undefined;
|
|
let updater: NodeJS.Timeout | undefined = undefined;
|
|
let timeout: NodeJS.Timeout | undefined = undefined;
|
|
|
|
const abort = () => {
|
|
clearInterval(pinger);
|
|
clearInterval(updater);
|
|
clearTimeout(timeout);
|
|
};
|
|
|
|
return eventStream(signal, (send, close) => {
|
|
const safeSend = (args: { event?: string; data: string }) => {
|
|
try {
|
|
send(args);
|
|
} catch (error) {
|
|
if (error instanceof Error) {
|
|
if (error.name !== "TypeError") {
|
|
logger.debug("Error sending SSE, aborting", {
|
|
error: {
|
|
name: error.name,
|
|
message: error.message,
|
|
stack: error.stack,
|
|
},
|
|
args,
|
|
});
|
|
}
|
|
} else {
|
|
logger.debug("Unknown error sending SSE, aborting", {
|
|
error,
|
|
args,
|
|
});
|
|
}
|
|
|
|
close();
|
|
}
|
|
};
|
|
|
|
pinger = setInterval(() => {
|
|
if (signal.aborted) {
|
|
return abort();
|
|
}
|
|
|
|
safeSend({ event: "ping", data: new Date().toISOString() });
|
|
}, pingInterval);
|
|
|
|
updater = setInterval(() => {
|
|
if (signal.aborted) {
|
|
return abort();
|
|
}
|
|
|
|
run(safeSend, abort);
|
|
}, updateInterval);
|
|
|
|
timeout = setTimeout(() => {
|
|
close(); // close the connection after 1 minute of inactivity, which will refresh the connection (that's why we aren't using abort)
|
|
}, 60 * 1000); // 1 minute
|
|
|
|
return abort;
|
|
});
|
|
}
|