refactor: Improve SSE connection and event handler management
- Add flag to prevent re-registering SSE event handlers - Enhance SSE connection setup with connection state checks - Implement safer event handler registration mechanism - Add additional connection cleanup and reset logic for reconnection attempts - Improve debug logging for SSE connection and handler registration
- commit
- 3f69ff24ebe95e8dc5e277b28116ce902a5d3b9b
- author
- Kit · · 2025-02-15 07:52 UTC
- parent
- cee1fd8
- signature
- SSH Unverified
SHA256:H91DFuEG4jpk5edzLA4aUKkpv4zv9Q339ipRmkph6TkThe key is unknown
1 file · +37 -9
M src/lib/states/chat.svelte.ts +37 -9
| @@ -28,6 +28,8 @@ class ChatState { | |||
| 28 | 28 | private connectionAttemptTimeout: ReturnType<typeof setTimeout> | null = null; | |
| 29 | 29 | private _memoizedEnrichedMessages: EnrichedMessage[] | null = null; | |
| 30 | 30 | private _lastEnrichmentKey: string = ''; | |
| 31 | + | // Flag to avoid re-registering SSE event handlers | |
| 32 | + | private sseHandlersRegistered = false; | |
| 31 | 33 | ||
| 32 | 34 | public async reinitialize() { | |
| 33 | 35 | if (this.isInitializing) { | |
| @@ -473,8 +475,11 @@ class ChatState { | |||
| 473 | 475 | } | |
| 474 | 476 | ||
| 475 | 477 | private setupSSE() { | |
| 476 | - | console.debug('Setting up SSE connection for chat messages'); | |
| 477 | - | ||
| 478 | + | // Prevent creating multiple SSE connections if one is already active. | |
| 479 | + | if (this.eventSource && this.eventSource.readyState !== EventSource.CLOSED) { | |
| 480 | + | console.debug('setupSSE: Active SSE connection exists, skipping setup.'); | |
| 481 | + | return; | |
| 482 | + | } | |
| 478 | 483 | // Clear any existing timeouts | |
| 479 | 484 | if (this.connectionTimeout) { | |
| 480 | 485 | clearTimeout(this.connectionTimeout); | |
| @@ -520,6 +525,8 @@ class ChatState { | |||
| 520 | 525 | ||
| 521 | 526 | console.debug('Connecting to SSE...'); | |
| 522 | 527 | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); | |
| 528 | + | // Reset the handlers flag for the new connection | |
| 529 | + | this.sseHandlersRegistered = false; | |
| 523 | 530 | ||
| 524 | 531 | this.eventSource.onopen = () => { | |
| 525 | 532 | console.debug('SSE connection established'); | |
| @@ -538,8 +545,8 @@ class ChatState { | |||
| 538 | 545 | } | |
| 539 | 546 | }; | |
| 540 | 547 | ||
| 541 | - | // Set up message handling | |
| 542 | - | this.setupMessageHandlers(this.eventSource); | |
| 548 | + | // Register event handlers (only once) | |
| 549 | + | this.setupMessageHandlers(); | |
| 543 | 550 | ||
| 544 | 551 | } catch (error) { | |
| 545 | 552 | console.debug('Error setting up SSE connection:', error); | |
| @@ -549,15 +556,25 @@ class ChatState { | |||
| 549 | 556 | } | |
| 550 | 557 | } | |
| 551 | 558 | ||
| 552 | - | private setupMessageHandlers(eventSource: EventSource) { | |
| 559 | + | // Register SSE event handlers only once per connection. | |
| 560 | + | private setupMessageHandlers() { | |
| 561 | + | if (this.sseHandlersRegistered) { | |
| 562 | + | console.debug('SSE message handlers already registered, skipping.'); | |
| 563 | + | return; | |
| 564 | + | } | |
| 565 | + | ||
| 566 | + | if (!this.eventSource) { | |
| 567 | + | console.debug('No SSE connection available for registering handlers, skipping.'); | |
| 568 | + | return; | |
| 569 | + | } | |
| 570 | + | ||
| 553 | 571 | // Listen for chat messages | |
| 554 | - | eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { | |
| 572 | + | this.eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { | |
| 555 | 573 | try { | |
| 556 | 574 | const messageData = JSON.parse(event.data) as Message; | |
| 557 | 575 | console.debug('Received chat message via SSE:', messageData); | |
| 558 | 576 | // Deduplicate and update messages array with the new message | |
| 559 | 577 | this.messages = Array.from(new Map([...this.messages, messageData].map(m => [m.id, m])).values()); | |
| 560 | - | ||
| 561 | 578 | // Ensure the sender's data is available | |
| 562 | 579 | await this.ensureUserData(messageData.senderId); | |
| 563 | 580 | } catch (error) { | |
| @@ -566,16 +583,17 @@ class ChatState { | |||
| 566 | 583 | }); | |
| 567 | 584 | ||
| 568 | 585 | // Listen for user status updates to update the buddy list dynamically | |
| 569 | - | eventSource.addEventListener('userStatusUpdate', async (event: MessageEvent) => { | |
| 586 | + | this.eventSource.addEventListener('userStatusUpdate', async (event: MessageEvent) => { | |
| 570 | 587 | try { | |
| 571 | 588 | const statusUpdate = JSON.parse(event.data) as { userId: string; status: User['status']; lastSeen: number }; | |
| 572 | 589 | console.debug('Received userStatusUpdate via SSE:', statusUpdate); | |
| 573 | - | // Update the user status and ensure buddy list is updated appropriately | |
| 574 | 590 | this.updateUserStatus(statusUpdate.userId, statusUpdate.status, statusUpdate.lastSeen); | |
| 575 | 591 | } catch (error) { | |
| 576 | 592 | console.debug('Error handling userStatusUpdate via SSE:', error); | |
| 577 | 593 | } | |
| 578 | 594 | }); | |
| 595 | + | ||
| 596 | + | this.sseHandlersRegistered = true; | |
| 579 | 597 | } | |
| 580 | 598 | ||
| 581 | 599 | private async ensureUserData(userId: string) { | |
| @@ -606,10 +624,20 @@ class ChatState { | |||
| 606 | 624 | this.reconnectAttempts++; | |
| 607 | 625 | if (this.reconnectAttempts > this.maxReconnectAttempts) { | |
| 608 | 626 | console.debug('Max reconnect attempts reached, giving up'); | |
| 627 | + | if (this.eventSource) { | |
| 628 | + | this.eventSource.close(); | |
| 629 | + | } | |
| 609 | 630 | this.eventSource = null; | |
| 610 | 631 | return; | |
| 611 | 632 | } | |
| 612 | 633 | this.reconnectDelay *= 2; // Exponential backoff | |
| 634 | + | // Ensure any existing connection is properly closed before reconnecting | |
| 635 | + | if (this.eventSource) { | |
| 636 | + | this.eventSource.close(); | |
| 637 | + | this.eventSource = null; | |
| 638 | + | } | |
| 639 | + | // Reset the handlers flag so that the new connection gets its event listeners registered fresh. | |
| 640 | + | this.sseHandlersRegistered = false; | |
| 613 | 641 | this.setupSSE(); | |
| 614 | 642 | } | |
| 615 | 643 | ||
| @@ -28,6 +28,8 @@ class ChatState { | |||||
| 28 | private connectionAttemptTimeout: ReturnType<typeof setTimeout> | null = null; | 28 | private connectionAttemptTimeout: ReturnType<typeof setTimeout> | null = null; | ||
| 29 | private _memoizedEnrichedMessages: EnrichedMessage[] | null = null; | 29 | private _memoizedEnrichedMessages: EnrichedMessage[] | null = null; | ||
| 30 | private _lastEnrichmentKey: string = ''; | 30 | private _lastEnrichmentKey: string = ''; | ||
| 31 | + | // Flag to avoid re-registering SSE event handlers | |||
| 32 | + | private sseHandlersRegistered = false; | |||
| 31 | 33 | ||||
| 32 | public async reinitialize() { | 34 | public async reinitialize() { | ||
| 33 | if (this.isInitializing) { | 35 | if (this.isInitializing) { | ||
| @@ -473,8 +475,11 @@ class ChatState { | |||||
| 473 | } | 475 | } | ||
| 474 | 476 | ||||
| 475 | private setupSSE() { | 477 | private setupSSE() { | ||
| 476 | - | console.debug('Setting up SSE connection for chat messages'); | 478 | + | // Prevent creating multiple SSE connections if one is already active. |
| 477 | - | 479 | + | if (this.eventSource && this.eventSource.readyState !== EventSource.CLOSED) { | |
| 480 | + | console.debug('setupSSE: Active SSE connection exists, skipping setup.'); | |||
| 481 | + | return; | |||
| 482 | + | } | |||
| 478 | // Clear any existing timeouts | 483 | // Clear any existing timeouts | ||
| 479 | if (this.connectionTimeout) { | 484 | if (this.connectionTimeout) { | ||
| 480 | clearTimeout(this.connectionTimeout); | 485 | clearTimeout(this.connectionTimeout); | ||
| @@ -520,6 +525,8 @@ class ChatState { | |||||
| 520 | 525 | ||||
| 521 | console.debug('Connecting to SSE...'); | 526 | console.debug('Connecting to SSE...'); | ||
| 522 | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); | 527 | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); | ||
| 528 | + | // Reset the handlers flag for the new connection | |||
| 529 | + | this.sseHandlersRegistered = false; | |||
| 523 | 530 | ||||
| 524 | this.eventSource.onopen = () => { | 531 | this.eventSource.onopen = () => { | ||
| 525 | console.debug('SSE connection established'); | 532 | console.debug('SSE connection established'); | ||
| @@ -538,8 +545,8 @@ class ChatState { | |||||
| 538 | } | 545 | } | ||
| 539 | }; | 546 | }; | ||
| 540 | 547 | ||||
| 541 | - | // Set up message handling | 548 | + | // Register event handlers (only once) |
| 542 | - | this.setupMessageHandlers(this.eventSource); | 549 | + | this.setupMessageHandlers(); |
| 543 | 550 | ||||
| 544 | } catch (error) { | 551 | } catch (error) { | ||
| 545 | console.debug('Error setting up SSE connection:', error); | 552 | console.debug('Error setting up SSE connection:', error); | ||
| @@ -549,15 +556,25 @@ class ChatState { | |||||
| 549 | } | 556 | } | ||
| 550 | } | 557 | } | ||
| 551 | 558 | ||||
| 552 | - | private setupMessageHandlers(eventSource: EventSource) { | 559 | + | // Register SSE event handlers only once per connection. |
| 560 | + | private setupMessageHandlers() { | |||
| 561 | + | if (this.sseHandlersRegistered) { | |||
| 562 | + | console.debug('SSE message handlers already registered, skipping.'); | |||
| 563 | + | return; | |||
| 564 | + | } | |||
| 565 | + | ||||
| 566 | + | if (!this.eventSource) { | |||
| 567 | + | console.debug('No SSE connection available for registering handlers, skipping.'); | |||
| 568 | + | return; | |||
| 569 | + | } | |||
| 570 | + | ||||
| 553 | // Listen for chat messages | 571 | // Listen for chat messages | ||
| 554 | - | eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { | 572 | + | this.eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { |
| 555 | try { | 573 | try { | ||
| 556 | const messageData = JSON.parse(event.data) as Message; | 574 | const messageData = JSON.parse(event.data) as Message; | ||
| 557 | console.debug('Received chat message via SSE:', messageData); | 575 | console.debug('Received chat message via SSE:', messageData); | ||
| 558 | // Deduplicate and update messages array with the new message | 576 | // Deduplicate and update messages array with the new message | ||
| 559 | this.messages = Array.from(new Map([...this.messages, messageData].map(m => [m.id, m])).values()); | 577 | this.messages = Array.from(new Map([...this.messages, messageData].map(m => [m.id, m])).values()); | ||
| 560 | - | ||||
| 561 | // Ensure the sender's data is available | 578 | // Ensure the sender's data is available | ||
| 562 | await this.ensureUserData(messageData.senderId); | 579 | await this.ensureUserData(messageData.senderId); | ||
| 563 | } catch (error) { | 580 | } catch (error) { | ||
| @@ -566,16 +583,17 @@ class ChatState { | |||||
| 566 | }); | 583 | }); | ||
| 567 | 584 | ||||
| 568 | // Listen for user status updates to update the buddy list dynamically | 585 | // Listen for user status updates to update the buddy list dynamically | ||
| 569 | - | eventSource.addEventListener('userStatusUpdate', async (event: MessageEvent) => { | 586 | + | this.eventSource.addEventListener('userStatusUpdate', async (event: MessageEvent) => { |
| 570 | try { | 587 | try { | ||
| 571 | const statusUpdate = JSON.parse(event.data) as { userId: string; status: User['status']; lastSeen: number }; | 588 | const statusUpdate = JSON.parse(event.data) as { userId: string; status: User['status']; lastSeen: number }; | ||
| 572 | console.debug('Received userStatusUpdate via SSE:', statusUpdate); | 589 | console.debug('Received userStatusUpdate via SSE:', statusUpdate); | ||
| 573 | - | // Update the user status and ensure buddy list is updated appropriately | |||
| 574 | this.updateUserStatus(statusUpdate.userId, statusUpdate.status, statusUpdate.lastSeen); | 590 | this.updateUserStatus(statusUpdate.userId, statusUpdate.status, statusUpdate.lastSeen); | ||
| 575 | } catch (error) { | 591 | } catch (error) { | ||
| 576 | console.debug('Error handling userStatusUpdate via SSE:', error); | 592 | console.debug('Error handling userStatusUpdate via SSE:', error); | ||
| 577 | } | 593 | } | ||
| 578 | }); | 594 | }); | ||
| 595 | + | ||||
| 596 | + | this.sseHandlersRegistered = true; | |||
| 579 | } | 597 | } | ||
| 580 | 598 | ||||
| 581 | private async ensureUserData(userId: string) { | 599 | private async ensureUserData(userId: string) { | ||
| @@ -606,10 +624,20 @@ class ChatState { | |||||
| 606 | this.reconnectAttempts++; | 624 | this.reconnectAttempts++; | ||
| 607 | if (this.reconnectAttempts > this.maxReconnectAttempts) { | 625 | if (this.reconnectAttempts > this.maxReconnectAttempts) { | ||
| 608 | console.debug('Max reconnect attempts reached, giving up'); | 626 | console.debug('Max reconnect attempts reached, giving up'); | ||
| 627 | + | if (this.eventSource) { | |||
| 628 | + | this.eventSource.close(); | |||
| 629 | + | } | |||
| 609 | this.eventSource = null; | 630 | this.eventSource = null; | ||
| 610 | return; | 631 | return; | ||
| 611 | } | 632 | } | ||
| 612 | this.reconnectDelay *= 2; // Exponential backoff | 633 | this.reconnectDelay *= 2; // Exponential backoff | ||
| 634 | + | // Ensure any existing connection is properly closed before reconnecting | |||
| 635 | + | if (this.eventSource) { | |||
| 636 | + | this.eventSource.close(); | |||
| 637 | + | this.eventSource = null; | |||
| 638 | + | } | |||
| 639 | + | // Reset the handlers flag so that the new connection gets its event listeners registered fresh. | |||
| 640 | + | this.sseHandlersRegistered = false; | |||
| 613 | this.setupSSE(); | 641 | this.setupSSE(); | ||
| 614 | } | 642 | } | ||
| 615 | 643 | ||||