fix: SSE stream endpoint with proper HTTP/2 stream handling and heartbeat
- Fixed /api/stream/requests endpoint HTTP/2 INTERNAL_ERROR - Use reply.raw.writeHead() instead of Fastify headers API for SSE - Added 30s heartbeat to keep connection alive - Proper event format with 'event:' and 'data:' fields - Comprehensive error handling and cleanup on disconnect - Mirrors working pattern from /api/stream/costs endpoint - Resolves dashboard perpetual 'Loading...' state
This commit is contained in:
parent
200cc7f2dc
commit
91384dbb2a
@ -463,31 +463,82 @@ export async function dashboardRoute(fastify: FastifyInstance): Promise<void> {
|
|||||||
|
|
||||||
// Server-Sent Events endpoint for real-time request updates
|
// Server-Sent Events endpoint for real-time request updates
|
||||||
fastify.get('/api/stream/requests', async (request: FastifyRequest, reply: FastifyReply) => {
|
fastify.get('/api/stream/requests', async (request: FastifyRequest, reply: FastifyReply) => {
|
||||||
// Set SSE headers
|
// Use raw Node.js API to properly initialize HTTP/2 stream
|
||||||
reply.type('text/event-stream');
|
reply.raw.writeHead(200, {
|
||||||
reply.header('Cache-Control', 'no-cache');
|
'Content-Type': 'text/event-stream',
|
||||||
reply.header('Connection', 'keep-alive');
|
'Cache-Control': 'no-cache',
|
||||||
|
'Connection': 'keep-alive',
|
||||||
|
'Access-Control-Allow-Origin': '*',
|
||||||
|
'Access-Control-Allow-Methods': 'GET',
|
||||||
|
'Access-Control-Allow-Headers': 'Content-Type',
|
||||||
|
});
|
||||||
|
|
||||||
|
const clientIp = request.ip;
|
||||||
|
const clientId = `${clientIp}-${Date.now()}`;
|
||||||
|
|
||||||
|
logger.info({ clientId, clientIp, activeListeners: globalRequestStream.getListenerCount() }, 'SSE client connected to /api/stream/requests');
|
||||||
|
|
||||||
// Send initial connection message
|
// Send initial connection message
|
||||||
reply.raw.write(`data: ${JSON.stringify({ type: 'connected', timestamp: new Date().toISOString() })}\n\n`);
|
reply.raw.write('event: connected\n');
|
||||||
|
reply.raw.write(`data: ${JSON.stringify({ clientId, timestamp: new Date().toISOString() })}\n\n`);
|
||||||
|
|
||||||
// Subscribe to request events
|
// Subscribe to request events
|
||||||
const unsubscribe = globalRequestStream.onRequest((event) => {
|
const unsubscribe = globalRequestStream.onRequest((event) => {
|
||||||
reply.raw.write(`data: ${JSON.stringify(event)}\n\n`);
|
try {
|
||||||
|
reply.raw.write('event: request-update\n');
|
||||||
|
reply.raw.write(`data: ${JSON.stringify(event)}\n\n`);
|
||||||
|
} catch (err) {
|
||||||
|
logger.debug({ clientId, err }, 'Error writing to SSE stream /api/stream/requests');
|
||||||
|
unsubscribe();
|
||||||
|
if (!reply.raw.writableEnded) {
|
||||||
|
reply.raw.end();
|
||||||
|
}
|
||||||
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Keep connection alive with heartbeat every 30 seconds
|
||||||
|
const heartbeat = setInterval(() => {
|
||||||
|
try {
|
||||||
|
if (reply.raw.writable) {
|
||||||
|
reply.raw.write(': heartbeat\n\n');
|
||||||
|
} else {
|
||||||
|
clearInterval(heartbeat);
|
||||||
|
unsubscribe();
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
logger.debug({ clientId, err }, 'Heartbeat failed on /api/stream/requests');
|
||||||
|
clearInterval(heartbeat);
|
||||||
|
unsubscribe();
|
||||||
|
}
|
||||||
|
}, 30000);
|
||||||
|
|
||||||
// Handle client disconnect
|
// Handle client disconnect
|
||||||
reply.raw.on('close', () => {
|
reply.raw.on('close', () => {
|
||||||
|
logger.info({ clientId }, 'SSE client disconnected from /api/stream/requests');
|
||||||
|
clearInterval(heartbeat);
|
||||||
unsubscribe();
|
unsubscribe();
|
||||||
logger.info('SSE client disconnected from /api/stream/requests');
|
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Handle stream errors
|
||||||
reply.raw.on('error', (error) => {
|
reply.raw.on('error', (error) => {
|
||||||
logger.error({ error }, 'SSE stream error');
|
logger.error({ clientId, error }, 'SSE stream error on /api/stream/requests');
|
||||||
|
clearInterval(heartbeat);
|
||||||
unsubscribe();
|
unsubscribe();
|
||||||
});
|
});
|
||||||
|
|
||||||
logger.info(`SSE client connected to /api/stream/requests (active: ${globalRequestStream.getListenerCount()})`);
|
// Cleanup on reply finish
|
||||||
|
reply.raw.on('finish', () => {
|
||||||
|
logger.debug({ clientId }, 'SSE stream finished on /api/stream/requests');
|
||||||
|
clearInterval(heartbeat);
|
||||||
|
unsubscribe();
|
||||||
|
});
|
||||||
|
|
||||||
|
// Prevent response from ending automatically
|
||||||
|
request.raw.on('close', () => {
|
||||||
|
logger.debug({ clientId }, 'Request closed on /api/stream/requests');
|
||||||
|
clearInterval(heartbeat);
|
||||||
|
unsubscribe();
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
// Test endpoint
|
// Test endpoint
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user