Angular 8 - 处理 SSE 错误重新连接

### Angular 8 中处理 SSE 错误并重新连接

typescript
import { Injectable } from '@angular/core';
import { Observable } from 'rxjs';
@Injectable({
  providedIn: 'root'
})
export class SseService {
  private eventSource: EventSource | undefined;
  constructor() { }
  initSse(): Observable {
    return new Observable((observer) => {
      this.eventSource = new EventSource('YOUR_SSE_ENDPOINT');
      this.eventSource.onmessage = (event: MessageEvent) => {
        observer.next(event);
      };
      this.eventSource.onerror = (error: Event) => {
        if (this.eventSource) {
          this.eventSource.close();
        }
        observer.error(error);
      };
    });
  }
}

typescript
import { Injectable } from '@angular/core';
import { Observable, throwError, timer } from 'rxjs';
import { catchError, switchMap } from 'rxjs/operators';
@Injectable({
  providedIn: 'root'
})
export class SseService {
  private eventSource: EventSource | undefined;
  constructor() { }
  initSse(): Observable {
    return new Observable((observer) => {
      this.establishConnection(observer);
    });
  }
  private establishConnection(observer: any) {
    this.eventSource = new EventSource('YOUR_SSE_ENDPOINT');
    this.eventSource.onmessage = (event: MessageEvent) => {
      observer.next(event);
    };
    this.eventSource.onerror = (error: Event) => {
      if (this.eventSource) {
        this.eventSource.close();
      }
      observer.error(error);
      // Retry connection after a delay (e.g., 5 seconds)
      this.retryConnection(observer);
    };
  }
  private retryConnection(observer: any) {
    timer(5000).pipe(
      switchMap(() => {
        this.establishConnection(observer);
        return throwError('Error occurred, retrying...');
      }),
      catchError((err) => {
        console.error(err);
        return throwError(err);
      })
    ).subscribe();
  }
}