From 4a41b0062dda8c371aab377fe9410de3754dd631 Mon Sep 17 00:00:00 2001 From: Daniel Pupius Date: Fri, 25 Apr 2025 09:00:00 -0700 Subject: [PATCH] Cleanup error handling --- server/server.go | 4 +- web/src/lib/api.ts | 218 +++++++++++++++++++++++++++-------- web/src/lib/types.ts | 233 ++++++++++++++++++++------------------ web/src/routes/__root.tsx | 125 ++++++++------------ 4 files changed, 340 insertions(+), 240 deletions(-) diff --git a/server/server.go b/server/server.go index efcaf87..4809ea9 100644 --- a/server/server.go +++ b/server/server.go @@ -143,7 +143,7 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) { // flush so the client knows the connection is open. w.WriteHeader(http.StatusOK) - initialMessage := "Connected to Meshtastic stream" + initialMessage := "Connected to stream" paddingSize := 1500 padding := make([]byte, paddingSize) for i := 0; i < paddingSize; i++ { @@ -152,7 +152,7 @@ func (s *Server) handleStream(w http.ResponseWriter, r *http.Request) { // Send the event with the padded data fmt.Fprintf(w, "event: info\ndata: %s\n\n", initialMessage) - fmt.Fprintf(w, "event: info\ndata: %s\n\n", padding) + fmt.Fprintf(w, "event: padding\ndata: %s\n\n", padding) flusher.Flush() // Log that we sent the large initial message diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index aee523f..ee866c1 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -8,6 +8,7 @@ import { StreamEventHandler, InfoEvent, MessageEvent, + PaddingEvent, BadDataEvent, } from "./types"; @@ -22,63 +23,188 @@ export type { MessageEvent, BadDataEvent, StreamEvent, + PaddingEvent, StreamEventHandler, }; /** * Establish a Server-Sent Events connection to receive real-time packets + * with automatic reconnection + * + * @param onEvent Handler function for stream events (messages, info, errors) + * @param onError Optional handler for connection errors + * @returns A function that closes the connection when called */ export function streamPackets( onEvent: StreamEventHandler, onError?: (error: Event) => void ): () => void { - const evtSource = new EventSource(API_ENDPOINTS.STREAM); - - console.log("[SSE] Connecting to stream...", evtSource); - - // Handle info events specifically - evtSource.addEventListener("info", (event) => { + // Connection state + let source: EventSource | null = null; + let reconnectTimer: number | null = null; + let shouldReconnect = true; + let reconnectAttempt = 0; + + // Reconnection settings + const INITIAL_RECONNECT_DELAY = 1000; // 1 second + const MAX_RECONNECT_DELAY = 30000; // 30 seconds + const MAX_RECONNECT_ATTEMPTS = 30; // Give up after this many attempts + + /** + * Calculate delay for exponential backoff + */ + function getReconnectDelay(): number { + // Calculate exponential backoff: 1s, 2s, 4s, 8s, etc. up to max + const delay = Math.min( + INITIAL_RECONNECT_DELAY * Math.pow(2, reconnectAttempt), + MAX_RECONNECT_DELAY + ); + + return delay; + } + + /** + * Clean up the current connection and timers + */ + function cleanup(): void { + // Clear any pending reconnect timer + if (reconnectTimer !== null) { + window.clearTimeout(reconnectTimer); + reconnectTimer = null; + } + + // Close and cleanup the event source if it exists + if (source !== null) { + // Remove all event listeners + source.removeEventListener("message", handleMessage as EventListener); + source.removeEventListener("info", handleInfo as EventListener); + source.onerror = null; + + // Close the connection + source.close(); + source = null; + } + } + + /** + * Handle info events + */ + function handleInfo(event: Event): void { + const evtData = (event as any).data; + // If we receive an info event, connection is working + if (reconnectAttempt > 0) { + console.log("[SSE] Connection restored"); + reconnectAttempt = 0; + } + + // Forward the event to the caller onEvent({ type: "info", - data: event.data, - }); - }); - - // Handle message events specifically - evtSource.addEventListener("message", (event) => { - handleEventData(event.data, onEvent); - }); - - // Handle errors - if (onError) { - evtSource.onerror = onError; - } else { - evtSource.onerror = () => { - console.error("EventSource failed"); - evtSource.close(); - }; - } - - // Return cleanup function - return () => evtSource.close(); -} - -/** - * Helper to handle event data based on type - */ -function handleEventData(data: string, callback: StreamEventHandler): void { - try { - const parsedData = JSON.parse(data) as Packet; - - callback({ - type: "message", - data: parsedData, - }); - } catch (error) { - console.warn("Failed to parse message as JSON:", error, "Raw data:", data); - callback({ - type: "bad_data", - data, + data: String(evtData), }); } -} + + /** + * Handle message events + */ + function handleMessage(event: Event): void { + const evtData = (event as any).data; + try { + // Parse the event data as JSON + const parsedData = JSON.parse(String(evtData)) as Packet; + + // Forward the message to the caller + onEvent({ + type: "message", + data: parsedData, + }); + } catch (error) { + console.warn("[SSE] Failed to parse message:", error); + + // Forward bad data to the caller + onEvent({ + type: "bad_data", + data: String(evtData), + }); + } + } + + /** + * Handle connection errors and reconnection + */ + function handleError(event: Event): void { + // Log the error + console.error("[SSE] Connection error"); + + // Notify the caller + if (onError) { + onError(event); + } + + // Clean up the current connection + cleanup(); + + // Reconnect if we should + if (shouldReconnect && reconnectAttempt < MAX_RECONNECT_ATTEMPTS) { + reconnectAttempt++; + + // Calculate backoff delay + const delay = getReconnectDelay(); + console.log(`[SSE] Reconnecting in ${delay}ms (attempt ${reconnectAttempt})`); + + // Schedule reconnection + reconnectTimer = window.setTimeout(connect, delay); + } else if (reconnectAttempt >= MAX_RECONNECT_ATTEMPTS) { + console.error(`[SSE] Giving up after ${MAX_RECONNECT_ATTEMPTS} failed reconnection attempts`); + } + } + + /** + * Connect to the event stream + */ + function connect(): void { + // Clean up any existing connection first + cleanup(); + + // Don't continue if we're no longer supposed to reconnect + if (!shouldReconnect) { + return; + } + + try { + // Create a new EventSource connection + source = new EventSource(API_ENDPOINTS.STREAM); + + // Log connection attempt + if (reconnectAttempt === 0) { + console.log("[SSE] Connecting to stream"); + } else { + console.log(`[SSE] Reconnecting to stream (attempt ${reconnectAttempt})`); + } + + // Set up event handlers + source.addEventListener("info", handleInfo as EventListener); + source.addEventListener("message", handleMessage as EventListener); + source.onerror = handleError; + } catch (error) { + console.error("[SSE] Failed to create EventSource:", error); + handleError(new Event("error")); + } + } + + // Start the initial connection + connect(); + + /** + * Return a function that will close the connection when called + */ + return function close(): void { + console.log("[SSE] Closing connection permanently"); + + // Set flag to prevent reconnection + shouldReconnect = false; + + // Clean up resources + cleanup(); + }; +} \ No newline at end of file diff --git a/web/src/lib/types.ts b/web/src/lib/types.ts index ba592af..18ae4d5 100644 --- a/web/src/lib/types.ts +++ b/web/src/lib/types.ts @@ -37,153 +37,153 @@ export enum PortNum { // Map of PortNum string names to numeric enum values export const PortNumByName: Record = { - "UNKNOWN_APP": PortNum.UNKNOWN_APP, - "TEXT_MESSAGE_APP": PortNum.TEXT_MESSAGE_APP, - "REMOTE_HARDWARE_APP": PortNum.REMOTE_HARDWARE_APP, - "POSITION_APP": PortNum.POSITION_APP, - "NODEINFO_APP": PortNum.NODEINFO_APP, - "ROUTING_APP": PortNum.ROUTING_APP, - "ADMIN_APP": PortNum.ADMIN_APP, - "TEXT_MESSAGE_COMPRESSED_APP": PortNum.TEXT_MESSAGE_COMPRESSED_APP, - "WAYPOINT_APP": PortNum.WAYPOINT_APP, - "AUDIO_APP": PortNum.AUDIO_APP, - "REPLY_APP": PortNum.REPLY_APP, - "IP_TUNNEL_APP": PortNum.IP_TUNNEL_APP, - "SERIAL_APP": PortNum.SERIAL_APP, - "STORE_FORWARD_APP": PortNum.STORE_FORWARD_APP, - "RANGE_TEST_APP": PortNum.RANGE_TEST_APP, - "TELEMETRY_APP": PortNum.TELEMETRY_APP, - "PRIVATE_APP": PortNum.PRIVATE_APP, - "DETECTION_SENSOR_APP": PortNum.DETECTION_SENSOR_APP, - "ZPS_APP": PortNum.ZPS_APP, - "SIMULATOR_APP": PortNum.SIMULATOR_APP, - "TRACEROUTE_APP": PortNum.TRACEROUTE_APP, - "NEIGHBORINFO_APP": PortNum.NEIGHBORINFO_APP, - "PAXCOUNTER_APP": PortNum.PAXCOUNTER_APP, - "MAP_REPORT_APP": PortNum.MAP_REPORT_APP, - "ALERT_APP": PortNum.ALERT_APP, - "ATAK_PLUGIN": PortNum.ATAK_PLUGIN, - "POWERSTRESS_APP": PortNum.POWERSTRESS_APP, - "RETICULUM_TUNNEL_APP": PortNum.RETICULUM_TUNNEL_APP, -} + UNKNOWN_APP: PortNum.UNKNOWN_APP, + TEXT_MESSAGE_APP: PortNum.TEXT_MESSAGE_APP, + REMOTE_HARDWARE_APP: PortNum.REMOTE_HARDWARE_APP, + POSITION_APP: PortNum.POSITION_APP, + NODEINFO_APP: PortNum.NODEINFO_APP, + ROUTING_APP: PortNum.ROUTING_APP, + ADMIN_APP: PortNum.ADMIN_APP, + TEXT_MESSAGE_COMPRESSED_APP: PortNum.TEXT_MESSAGE_COMPRESSED_APP, + WAYPOINT_APP: PortNum.WAYPOINT_APP, + AUDIO_APP: PortNum.AUDIO_APP, + REPLY_APP: PortNum.REPLY_APP, + IP_TUNNEL_APP: PortNum.IP_TUNNEL_APP, + SERIAL_APP: PortNum.SERIAL_APP, + STORE_FORWARD_APP: PortNum.STORE_FORWARD_APP, + RANGE_TEST_APP: PortNum.RANGE_TEST_APP, + TELEMETRY_APP: PortNum.TELEMETRY_APP, + PRIVATE_APP: PortNum.PRIVATE_APP, + DETECTION_SENSOR_APP: PortNum.DETECTION_SENSOR_APP, + ZPS_APP: PortNum.ZPS_APP, + SIMULATOR_APP: PortNum.SIMULATOR_APP, + TRACEROUTE_APP: PortNum.TRACEROUTE_APP, + NEIGHBORINFO_APP: PortNum.NEIGHBORINFO_APP, + PAXCOUNTER_APP: PortNum.PAXCOUNTER_APP, + MAP_REPORT_APP: PortNum.MAP_REPORT_APP, + ALERT_APP: PortNum.ALERT_APP, + ATAK_PLUGIN: PortNum.ATAK_PLUGIN, + POWERSTRESS_APP: PortNum.POWERSTRESS_APP, + RETICULUM_TUNNEL_APP: PortNum.RETICULUM_TUNNEL_APP, +}; // Different payload types based on the Meshtastic protocol export interface Position { - latitudeI?: number; // multiply by 1e-7 to get degrees - longitudeI?: number; // multiply by 1e-7 to get degrees - altitude?: number; // in meters above MSL - time: number; // seconds since 1970 - locationSource?: string; // "LOC_UNSET", "LOC_MANUAL", "LOC_INTERNAL", "LOC_EXTERNAL" - altitudeSource?: string; // "ALT_UNSET", "ALT_MANUAL", "ALT_INTERNAL", "ALT_EXTERNAL", "ALT_BAROMETRIC" - timestamp?: number; // positional timestamp in integer epoch seconds + latitudeI?: number; // multiply by 1e-7 to get degrees + longitudeI?: number; // multiply by 1e-7 to get degrees + altitude?: number; // in meters above MSL + time: number; // seconds since 1970 + locationSource?: string; // "LOC_UNSET", "LOC_MANUAL", "LOC_INTERNAL", "LOC_EXTERNAL" + altitudeSource?: string; // "ALT_UNSET", "ALT_MANUAL", "ALT_INTERNAL", "ALT_EXTERNAL", "ALT_BAROMETRIC" + timestamp?: number; // positional timestamp in integer epoch seconds timestampMillisAdjust?: number; - altitudeHae?: number; // HAE altitude in meters + altitudeHae?: number; // HAE altitude in meters altitudeGeoSeparation?: number; - pdop?: number; // Position Dilution of Precision, in 1/100 units - hdop?: number; // Horizontal Dilution of Precision - vdop?: number; // Vertical Dilution of Precision - gpsAccuracy?: number; // GPS accuracy in mm - groundSpeed?: number; // in m/s - groundTrack?: number; // True North track in 1/100 degrees - fixQuality?: number; // GPS fix quality - fixType?: number; // GPS fix type 2D/3D - satsInView?: number; // Satellites in view - sensorId?: number; // For multiple positioning sensors - nextUpdate?: number; // Estimated time until next update in seconds - seqNumber?: number; // Sequence number for this packet - precisionBits?: number; // Bits of precision set by the sending node + pdop?: number; // Position Dilution of Precision, in 1/100 units + hdop?: number; // Horizontal Dilution of Precision + vdop?: number; // Vertical Dilution of Precision + gpsAccuracy?: number; // GPS accuracy in mm + groundSpeed?: number; // in m/s + groundTrack?: number; // True North track in 1/100 degrees + fixQuality?: number; // GPS fix quality + fixType?: number; // GPS fix type 2D/3D + satsInView?: number; // Satellites in view + sensorId?: number; // For multiple positioning sensors + nextUpdate?: number; // Estimated time until next update in seconds + seqNumber?: number; // Sequence number for this packet + precisionBits?: number; // Bits of precision set by the sending node } export interface User { - id: string; // Globally unique ID for this user - longName?: string; // Full name for this user - shortName?: string; // Very short name, ideally two characters - macaddr?: string; // MAC address of the device - hwModel?: string; // Hardware model name - hasGps?: boolean; // Whether the node has GPS capability - role?: string; // User's role in the mesh (e.g., "ROUTER") - snr?: number; // Signal-to-noise ratio - batteryLevel?: number; // Battery level 0-100 - voltage?: number; // Battery voltage + id: string; // Globally unique ID for this user + longName?: string; // Full name for this user + shortName?: string; // Very short name, ideally two characters + macaddr?: string; // MAC address of the device + hwModel?: string; // Hardware model name + hasGps?: boolean; // Whether the node has GPS capability + role?: string; // User's role in the mesh (e.g., "ROUTER") + snr?: number; // Signal-to-noise ratio + batteryLevel?: number; // Battery level 0-100 + voltage?: number; // Battery voltage channelUtilization?: number; // Channel utilization percentage - airUtilTx?: number; // Air utilization for transmission - lastHeard?: number; // Last time the node was heard from + airUtilTx?: number; // Air utilization for transmission + lastHeard?: number; // Last time the node was heard from } // Device metrics for telemetry export interface DeviceMetrics { - batteryLevel?: number; // 0-100 (>100 means powered) - voltage?: number; // Voltage measured + batteryLevel?: number; // 0-100 (>100 means powered) + voltage?: number; // Voltage measured channelUtilization?: number; - airUtilTx?: number; // Percent of airtime for transmission used within the last hour - uptimeSeconds?: number; // How long the device has been running since last reboot (in seconds) + airUtilTx?: number; // Percent of airtime for transmission used within the last hour + uptimeSeconds?: number; // How long the device has been running since last reboot (in seconds) } // Environment metrics for telemetry export interface EnvironmentMetrics { - temperature?: number; // Temperature measured + temperature?: number; // Temperature measured relativeHumidity?: number; // Relative humidity percent measured barometricPressure?: number; // Barometric pressure in hPA - gasResistance?: number; // Gas resistance in MOhm - voltage?: number; // Voltage measured (To be deprecated) - current?: number; // Current measured (To be deprecated) - iaq?: number; // IAQ value (0-500) - distance?: number; // Distance in mm (water level detection) - lux?: number; // Ambient light (Lux) - whiteLux?: number; // White light (irradiance) - irLux?: number; // Infrared lux - uvLux?: number; // Ultraviolet lux - windDirection?: number; // Wind direction in degrees (0 = North, 90 = East) - windSpeed?: number; // Wind speed in m/s - weight?: number; // Weight in KG - windGust?: number; // Wind gust in m/s - windLull?: number; // Wind lull in m/s - radiation?: number; // Radiation in µR/h - rainfall1h?: number; // Rainfall in the last hour in mm - rainfall24h?: number; // Rainfall in the last 24 hours in mm - soilMoisture?: number; // Soil moisture measured (% 1-100) + gasResistance?: number; // Gas resistance in MOhm + voltage?: number; // Voltage measured (To be deprecated) + current?: number; // Current measured (To be deprecated) + iaq?: number; // IAQ value (0-500) + distance?: number; // Distance in mm (water level detection) + lux?: number; // Ambient light (Lux) + whiteLux?: number; // White light (irradiance) + irLux?: number; // Infrared lux + uvLux?: number; // Ultraviolet lux + windDirection?: number; // Wind direction in degrees (0 = North, 90 = East) + windSpeed?: number; // Wind speed in m/s + weight?: number; // Weight in KG + windGust?: number; // Wind gust in m/s + windLull?: number; // Wind lull in m/s + radiation?: number; // Radiation in µR/h + rainfall1h?: number; // Rainfall in the last hour in mm + rainfall24h?: number; // Rainfall in the last 24 hours in mm + soilMoisture?: number; // Soil moisture measured (% 1-100) soilTemperature?: number; // Soil temperature measured (°C) } // Main telemetry interface export interface Telemetry { - time: number; // seconds since 1970 + time: number; // seconds since 1970 deviceMetrics?: DeviceMetrics; environmentMetrics?: EnvironmentMetrics; // Other telemetry types could be added as needed (PowerMetrics, AirQualityMetrics, etc.) } export interface Waypoint { - id: number; // ID of the waypoint - latitudeI?: number; // multiply by 1e-7 to get degrees - longitudeI?: number; // multiply by 1e-7 to get degrees - expire?: number; // Time the waypoint is to expire (epoch) - lockedTo?: number; // If greater than zero, nodenum allowed to update the waypoint - name?: string; // Name of the waypoint - max 30 chars - description?: string; // Description of the waypoint - max 100 chars - icon?: number; // Designator icon for the waypoint (unicode emoji) + id: number; // ID of the waypoint + latitudeI?: number; // multiply by 1e-7 to get degrees + longitudeI?: number; // multiply by 1e-7 to get degrees + expire?: number; // Time the waypoint is to expire (epoch) + lockedTo?: number; // If greater than zero, nodenum allowed to update the waypoint + name?: string; // Name of the waypoint - max 30 chars + description?: string; // Description of the waypoint - max 100 chars + icon?: number; // Designator icon for the waypoint (unicode emoji) } export interface RouteDiscovery { - route?: number[]; // The list of nodenums this packet has visited - snrTowards?: number[]; // The list of SNRs (in dB, scaled by 4) in the route towards destination - routeBack?: number[]; // The list of nodenums the packet has visited on the way back - snrBack?: number[]; // The list of SNRs (in dB, scaled by 4) in the route back + route?: number[]; // The list of nodenums this packet has visited + snrTowards?: number[]; // The list of SNRs (in dB, scaled by 4) in the route towards destination + routeBack?: number[]; // The list of nodenums the packet has visited on the way back + snrBack?: number[]; // The list of SNRs (in dB, scaled by 4) in the route back } // A neighbor in the mesh export interface Neighbor { - nodeId: number; // Node ID of neighbor - snr: number; // SNR of last heard message - lastRxTime?: number; // Reception time of last message + nodeId: number; // Node ID of neighbor + snr: number; // SNR of last heard message + lastRxTime?: number; // Reception time of last message nodeBroadcastIntervalSecs?: number; // Broadcast interval of this neighbor } export interface NeighborInfo { - nodeId: number; // The node ID of the node sending info on its neighbors - lastSentById?: number; // Field to pass neighbor info for the next sending cycle + nodeId: number; // The node ID of the node sending info on its neighbors + lastSentById?: number; // Field to pass neighbor info for the next sending cycle nodeBroadcastIntervalSecs?: number; // Broadcast interval of the represented node - neighbors?: Neighbor[]; // The list of out edges from this node + neighbors?: Neighbor[]; // The list of out edges from this node } export interface MapReport { @@ -217,13 +217,13 @@ export enum RoutingError { PKI_FAILED = 34, PKI_UNKNOWN_PUBKEY = 35, ADMIN_BAD_SESSION_KEY = 36, - ADMIN_PUBLIC_KEY_UNAUTHORIZED = 37 + ADMIN_PUBLIC_KEY_UNAUTHORIZED = 37, } export interface Routing { - routeRequest?: RouteDiscovery; // A route request going from the requester - routeReply?: RouteDiscovery; // A route reply - errorReason?: RoutingError; // A failure in delivering a message + routeRequest?: RouteDiscovery; // A route request going from the requester + routeReply?: RouteDiscovery; // A route reply + errorReason?: RoutingError; // A failure in delivering a message } export interface AdminMessage { @@ -268,7 +268,7 @@ export interface Data { // From Data portNum: PortNum | string; // Can be either enum value or string name - + // Payload is one of these types, depending on portNum textMessage?: string; binaryData?: string; // Base64 encoded binary data @@ -309,7 +309,7 @@ export interface Data { // Error tracking decodeError?: string; - + // Reception timestamp (added by decoder) rxTime?: number; } @@ -338,6 +338,15 @@ export interface BadDataEvent { data: string; // Raw data that failed to parse } -export type StreamEvent = InfoEvent | MessageEvent | BadDataEvent; +export interface PaddingEvent { + type: "padding"; + data: string; // Random data to trigger a flush. +} -export type StreamEventHandler = (event: StreamEvent) => void; \ No newline at end of file +export type StreamEvent = + | InfoEvent + | MessageEvent + | PaddingEvent + | BadDataEvent; + +export type StreamEventHandler = (event: StreamEvent) => void; diff --git a/web/src/routes/__root.tsx b/web/src/routes/__root.tsx index aba741e..a479bdb 100644 --- a/web/src/routes/__root.tsx +++ b/web/src/routes/__root.tsx @@ -1,5 +1,5 @@ import { Outlet } from "@tanstack/react-router"; -import { useState, useEffect } from "react"; +import { useState, useEffect, useRef } from "react"; import { useAppDispatch } from "../hooks"; import { Nav } from "../components"; import { streamPackets, StreamEvent } from "../lib/api"; @@ -13,92 +13,57 @@ export const Route = createRootRoute({ function RootLayout() { const dispatch = useAppDispatch(); - const [connectionStatus, setConnectionStatus] = useState("Connecting..."); - const [connectionAttempts, setConnectionAttempts] = useState(0); - const [isReconnecting, setIsReconnecting] = useState(false); + const [connectionStatus, setConnectionStatus] = + useState("Connecting..."); + + // Use refs instead of state for tracking reconnection status + // This way changes to these values won't trigger useEffect reruns + const connectionAttemptsRef = useRef(0); + const isReconnectingRef = useRef(false); useEffect(() => { - // Connection management - let reconnectTimer: number | null = null; - const MAX_RECONNECT_DELAY = 10000; // Maximum reconnect delay in ms (10 seconds) - const INITIAL_RECONNECT_DELAY = 1000; // Start with 1 second + console.log("[SSE] Setting up event source (should happen only once)"); - // Calculate exponential backoff delay with a cap - const getReconnectDelay = (attempts: number) => { - const delay = Math.min( - INITIAL_RECONNECT_DELAY * Math.pow(1.5, attempts), - MAX_RECONNECT_DELAY - ); - return delay; - }; + // Set up Server-Sent Events connection + const cleanup = streamPackets( + // Event handler for all event types + (event: StreamEvent) => { + if (event.type === "info") { + // Handle info events (connection status, etc.) + console.log(`[SSE] Info: ${event.data}`); + setConnectionStatus(event.data); - const connectToEventStream = () => { - // Log connection attempt - const attemptNum = connectionAttempts + 1; - console.log(`[SSE] Connection attempt ${attemptNum}${isReconnecting ? ' (reconnecting)' : ''}`); - - // Set up Server-Sent Events connection - const cleanup = streamPackets( - // Event handler for all event types - (event: StreamEvent) => { - if (event.type === "info") { - // Handle info events (connection status, etc.) - console.log(`[SSE] Info: ${event.data}`); - setConnectionStatus(event.data); - - // Reset connection attempts on successful connection - if (event.data.includes("Connected") && isReconnecting) { - console.log("[SSE] Connection restored successfully"); - setConnectionAttempts(0); - setIsReconnecting(false); - } - } else if (event.type === "message") { - // Process message for both the aggregator and packet display - dispatch(processNewPacket(event.data)); - dispatch(addPacket(event.data)); - } else if (event.type === "bad_data") { - console.warn("[SSE] Received bad data:", event.data); + // Reset connection attempts on successful connection + if (event.data.includes("Connected") && isReconnectingRef.current) { + console.log("[SSE] Connection restored successfully"); + connectionAttemptsRef.current = 0; + isReconnectingRef.current = false; } - }, - // On error handler - (error) => { - console.error("[SSE] Connection error:", error); - setConnectionStatus("Connection error. Reconnecting..."); - setIsReconnecting(true); - - // Increment connection attempts - const newAttempts = connectionAttempts + 1; - setConnectionAttempts(newAttempts); - - // Calculate backoff delay - const reconnectDelay = getReconnectDelay(newAttempts); - console.log(`[SSE] Will attempt to reconnect in ${reconnectDelay}ms (attempt ${newAttempts})`); - - // Close the current connection - cleanup(); - - // Schedule reconnection - reconnectTimer = window.setTimeout(() => { - connectToEventStream(); - }, reconnectDelay); + } else if (event.type === "message") { + // Process message for both the aggregator and packet display + dispatch(processNewPacket(event.data)); + dispatch(addPacket(event.data)); + } else if (event.type === "bad_data") { + console.warn("[SSE] Received bad data:", event.data); } - ); - - // Return cleanup function - return () => { - if (reconnectTimer) { - window.clearTimeout(reconnectTimer); - } - cleanup(); - }; - }; + }, + // On error handler - only for reporting UI state + (error) => { + console.error("[SSE] Connection error:", error); + setConnectionStatus("Connection error. Reconnecting..."); + isReconnectingRef.current = true; - // Initial connection - const cleanupFn = connectToEventStream(); - - // Cleanup when component unmounts - return cleanupFn; - }, [dispatch, connectionAttempts, isReconnecting]); + // Increment connection attempts for UI tracking + connectionAttemptsRef.current += 1; + console.log( + `[SSE] Connection attempt ${connectionAttemptsRef.current}` + ); + } + ); + + // This cleanup will only be called when the component unmounts + return cleanup; + }, [dispatch]); return (