如何区分Flux数据流结束与异常？基于Angular5与Spring5 WebFlux

Great question—this is a super common gotcha with SSE and EventSource, since the browser's native EventSource treats both normal stream completion and unexpected errors (like app crashes) the same way: it fires an error event. The fix relies on adding a custom "completion signal" from your Spring WebFlux backend, which your Angular app can listen for to tell the two scenarios apart.

Here's the step-by-step solution: #

1. Modify the Spring WebFlux Backend to Send a Completion Event #

When your Flux stream finishes normally (not due to an exception), append a custom Server-Sent Event (SSE) with a unique event type (e.g., complete). This signals to the frontend that the stream ended intentionally.

@GetMapping(value = "/sse-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> streamData() {
    // Replace this with your actual business data stream
    Flux<String> businessData = Flux.interval(Duration.ofSeconds(1))
            .map(i -> "Business data point " + i)
            .take(5); // Simulates a stream that ends after 5 items

    // Append a custom "complete" event to mark normal termination
    return businessData
            .map(data -> ServerSentEvent.builder(data).build())
            .concatWith(Mono.just(ServerSentEvent.<String>builder()
                    .event("complete") // Custom event type
                    .data("Stream finished successfully")
                    .build()));
}

Key details:

• Use concatWith to ensure the completion event is sent only after the main data stream finishes.
• The event("complete") flag lets the frontend distinguish this from regular message events.

2. Update the Angular 5 Frontend to Listen for the Completion Signal #

In your Angular component, add a listener for the custom complete event, and track whether the stream ended normally. Use this flag to interpret the error event correctly.

import { Component, OnInit, OnDestroy } from '@angular/core';

@Component({
  selector: 'app-sse-listener',
  template: `
    <h3>SSE Stream Status</h3>
    <ul>
      <li *ngFor="let msg of messages">{{msg}}</li>
    </ul>
  `
})
export class SseListenerComponent implements OnInit, OnDestroy {
  private eventSource: EventSource | null = null;
  public messages: string[] = [];
  private isStreamCompleted = false;

  ngOnInit(): void {
    this.eventSource = new EventSource('/sse-stream');

    // Listen for regular data messages
    this.eventSource.onmessage = (event) => {
      this.messages.push(`Received: ${event.data}`);
    };

    // Listen for the custom "complete" event
    this.eventSource.addEventListener('complete', (event) => {
      this.isStreamCompleted = true;
      this.messages.push(`✅ Normal completion: ${event.data}`);
      this.eventSource?.close(); // Clean up the connection
    });

    // Handle error events with context
    this.eventSource.onerror = (error) => {
      if (this.isStreamCompleted) {
        // Ignore post-completion cleanup errors (some browsers fire this)
        this.messages.push('ℹ️ Connection closed after normal finish');
      } else {
        // This is an unexpected failure
        this.messages.push(`❌ Unexpected disconnection: ${error.message || 'Unknown error'}`);
        this.eventSource?.close();
      }
    };
  }

  ngOnDestroy(): void {
    this.eventSource?.close();
  }
}

Why This Works: #

• If the stream ends normally: The backend sends the complete event first. The frontend marks isStreamCompleted = true, then any subsequent error event (from the browser closing the connection) is ignored.
• If the stream crashes unexpectedly: The backend never sends the complete event. The error event fires with isStreamCompleted = false, so you know it's a failure.

Extra Notes: #

• For backend exceptions: If your Flux throws an error, Spring WebFlux will close the connection without sending the complete event, so the frontend's error event will correctly flag it as an unexpected issue.
• Browser behavior: Some browsers may fire an error event immediately after you call eventSource.close(), hence the need for the isStreamCompleted flag to filter those out.

Hope this helps you reliably tell normal SSE closures apart from unexpected crashes! Let me know if you need to adjust this for your specific business logic.

内容的提问来源于stack exchange，提问作者Endolf