Fix: use ReadableStream cancel() to cleanup interval (avoid controller.signal TS error)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
2026-05-16 17:44:03 +02:00
parent 93056b7ecd
commit 5358ee6f97
+60
View File
@@ -0,0 +1,60 @@
import { fetchLatestBar } from "../../lib/alpacaClient";
export async function loader({ request }: { request: Request }) {
const url = new URL(request.url);
const ticker = (url.searchParams.get("ticker") || "").toUpperCase();
if (!ticker) return new Response("ticker required", { status: 400 });
const timeframe = url.searchParams.get("timeframe") || "1Min"; // default to 1Min bars for live price
const mode = url.searchParams.get("mode") === "live" ? 'live' : 'paper';
const headers = new Headers({
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
Connection: "keep-alive",
});
// Create a ReadableStream that polls latest bar every second and pushes SSE
let closed = false;
let interval: any;
const stream = new ReadableStream({
start(controller) {
// helper to push SSE event
function pushEvent(obj: any) {
try {
const payload = `data: ${JSON.stringify(obj)}\n\n`;
controller.enqueue(new TextEncoder().encode(payload));
} catch (e) {
console.warn("price-stream: failed to enqueue", e);
}
}
// initial ping
pushEvent({ event: "connected", ticker, timeframe });
interval = setInterval(async () => {
if (closed) return;
try {
const last = await fetchLatestBar(ticker, timeframe, mode);
const price = last ? (last.ClosePrice ?? last.c ?? null) : null;
if (price != null) {
pushEvent({ price, ts: Date.now(), timeframe });
} else {
pushEvent({ error: "no_bar", ts: Date.now() });
}
} catch (err) {
console.error("price-stream: error fetching latest bar", err);
pushEvent({ error: String(err), ts: Date.now() });
// keep trying
}
}, 1000);
// no-op here; cleanup handled in cancel
},
cancel() {
closed = true;
clearInterval(interval);
},
});
return new Response(stream, { headers });
}