Debounce terminal SSE subscription reconnects
This commit is contained in:
@@ -99,6 +99,18 @@ export function runTests() {
|
|||||||
assert(hook.includes("useLatestPatch"), "frontend patch hook must export useLatestPatch(city)");
|
assert(hook.includes("useLatestPatch"), "frontend patch hook must export useLatestPatch(city)");
|
||||||
assert(hook.includes("revision"), "frontend patch hook must track revisions and skip stale patches");
|
assert(hook.includes("revision"), "frontend patch hook must track revisions and skip stale patches");
|
||||||
assert(hook.includes("setTimeout"), "frontend patch hook must implement explicit reconnect backoff");
|
assert(hook.includes("setTimeout"), "frontend patch hook must implement explicit reconnect backoff");
|
||||||
|
assert(
|
||||||
|
hook.includes("SSE_SUBSCRIPTION_RECONNECT_DELAY_MS") &&
|
||||||
|
hook.includes("scheduleSubscriptionReconnect") &&
|
||||||
|
hook.includes("clearSubscriptionReconnectTimer"),
|
||||||
|
"frontend patch hook must debounce visible-city subscription changes before reopening SSE",
|
||||||
|
);
|
||||||
|
const subscriptionBlock = hook.match(/function registerCitySubscription[\s\S]*?\n}\r?\n\r?\nfunction normalizeLegacyPatch/)?.[0] || "";
|
||||||
|
assert(
|
||||||
|
subscriptionBlock.includes("scheduleSubscriptionReconnect()") &&
|
||||||
|
!subscriptionBlock.includes("ensureSsePatchConnection();"),
|
||||||
|
"city subscription mount/unmount should schedule one coalesced SSE reconnect instead of reconnecting per chart",
|
||||||
|
);
|
||||||
|
|
||||||
const bffEventsRoute = readFrontendFile("app", "api", "events", "route.ts");
|
const bffEventsRoute = readFrontendFile("app", "api", "events", "route.ts");
|
||||||
assert(bffEventsRoute.includes("searchParams"), "Next.js SSE proxy must forward query parameters to FastAPI");
|
assert(bffEventsRoute.includes("searchParams"), "Next.js SSE proxy must forward query parameters to FastAPI");
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import { useEffect, useSyncExternalStore } from "react";
|
|||||||
import { resolveBackendApiUrl } from "@/lib/backend-api";
|
import { resolveBackendApiUrl } from "@/lib/backend-api";
|
||||||
|
|
||||||
const V1_EVENT_TYPE = "city_observation_patch.v1";
|
const V1_EVENT_TYPE = "city_observation_patch.v1";
|
||||||
|
const SSE_SUBSCRIPTION_RECONNECT_DELAY_MS = 150;
|
||||||
|
|
||||||
export type CityPatch = {
|
export type CityPatch = {
|
||||||
type?: string;
|
type?: string;
|
||||||
@@ -38,6 +39,7 @@ const subscribedCities = new Map<string, number>();
|
|||||||
|
|
||||||
let eventSource: EventSource | null = null;
|
let eventSource: EventSource | null = null;
|
||||||
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
||||||
|
let subscriptionReconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
||||||
let reconnectAttempt = 0;
|
let reconnectAttempt = 0;
|
||||||
let patchVersion = 0;
|
let patchVersion = 0;
|
||||||
let resyncVersion = 0;
|
let resyncVersion = 0;
|
||||||
@@ -74,6 +76,12 @@ function clearReconnectTimer() {
|
|||||||
reconnectTimer = null;
|
reconnectTimer = null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function clearSubscriptionReconnectTimer() {
|
||||||
|
if (!subscriptionReconnectTimer) return;
|
||||||
|
clearTimeout(subscriptionReconnectTimer);
|
||||||
|
subscriptionReconnectTimer = null;
|
||||||
|
}
|
||||||
|
|
||||||
function buildSseUrl(baseUrl: string) {
|
function buildSseUrl(baseUrl: string) {
|
||||||
const params = new URLSearchParams();
|
const params = new URLSearchParams();
|
||||||
const cities = subscribedCityList();
|
const cities = subscribedCityList();
|
||||||
@@ -113,6 +121,7 @@ function closeEventSource() {
|
|||||||
function reconnectNow() {
|
function reconnectNow() {
|
||||||
if (typeof window === "undefined") return;
|
if (typeof window === "undefined") return;
|
||||||
clearReconnectTimer();
|
clearReconnectTimer();
|
||||||
|
clearSubscriptionReconnectTimer();
|
||||||
closeEventSource();
|
closeEventSource();
|
||||||
connectSsePatches();
|
connectSsePatches();
|
||||||
}
|
}
|
||||||
@@ -159,6 +168,7 @@ function connectSsePatches() {
|
|||||||
|
|
||||||
function ensureSsePatchConnection() {
|
function ensureSsePatchConnection() {
|
||||||
if (subscribedCities.size === 0) {
|
if (subscribedCities.size === 0) {
|
||||||
|
clearSubscriptionReconnectTimer();
|
||||||
closeEventSource();
|
closeEventSource();
|
||||||
clearReconnectTimer();
|
clearReconnectTimer();
|
||||||
return;
|
return;
|
||||||
@@ -167,6 +177,19 @@ function ensureSsePatchConnection() {
|
|||||||
reconnectNow();
|
reconnectNow();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function scheduleSubscriptionReconnect() {
|
||||||
|
if (typeof window === "undefined") return;
|
||||||
|
if (subscribedCities.size === 0) {
|
||||||
|
ensureSsePatchConnection();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
clearSubscriptionReconnectTimer();
|
||||||
|
subscriptionReconnectTimer = setTimeout(() => {
|
||||||
|
subscriptionReconnectTimer = null;
|
||||||
|
ensureSsePatchConnection();
|
||||||
|
}, SSE_SUBSCRIPTION_RECONNECT_DELAY_MS);
|
||||||
|
}
|
||||||
|
|
||||||
function registerCitySubscription(city: string) {
|
function registerCitySubscription(city: string) {
|
||||||
const cityKey = normalizeCityKey(city);
|
const cityKey = normalizeCityKey(city);
|
||||||
if (!cityKey) return () => {};
|
if (!cityKey) return () => {};
|
||||||
@@ -174,7 +197,7 @@ function registerCitySubscription(city: string) {
|
|||||||
const previousCount = subscribedCities.get(cityKey) ?? 0;
|
const previousCount = subscribedCities.get(cityKey) ?? 0;
|
||||||
subscribedCities.set(cityKey, previousCount + 1);
|
subscribedCities.set(cityKey, previousCount + 1);
|
||||||
if (previousCount === 0) {
|
if (previousCount === 0) {
|
||||||
ensureSsePatchConnection();
|
scheduleSubscriptionReconnect();
|
||||||
}
|
}
|
||||||
|
|
||||||
return () => {
|
return () => {
|
||||||
@@ -184,7 +207,7 @@ function registerCitySubscription(city: string) {
|
|||||||
} else {
|
} else {
|
||||||
subscribedCities.set(cityKey, nextCount);
|
subscribedCities.set(cityKey, nextCount);
|
||||||
}
|
}
|
||||||
ensureSsePatchConnection();
|
scheduleSubscriptionReconnect();
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user