typescript - Angular'da SSE'yi uygulama

Test uygulamamda SSE uygulamaya çalışıyorum. Sunucu tarafı kuruldu ve sadece uç noktayı (api/v1/sse/document) kullanıyorum. Senaryo şu: Bir tarama yaptığımda, taramanın test istemcimde ve ana uygulamamda görünmesi gerekiyor. Sorun şu ki, sadece sayfayı yenilediğimde test istemcimde görünüyor. Biraz kod yazdım ama test istemcimdeki olayları otomatik olarak güncelleyebilmeyi hâlâ sağlayamıyorum.

bu benim yazdığım kod

sse.servis.ts

[document-list.component.ts]

public ngOnInit(): void {
        this.getDocuments();
        this.registerServerSentEvent();
    }

    public ngOnDestroy(): void {
        this.closeServerSentEvent();
    }

    /**
     * getDocuments: call the api service to get the list of documents.
     */
    public getDocuments(): void {
        this.aService.getDocuments().subscribe((documents: Documents[]) => {
            this.documents = documents;
        });
    }

    /**
     * markDocumentAsProcessed: call the api service to mark the document as processed.
     */
    public markDocumentAsProcessed(document: Documents): void {
        this.aService.markDocumentProcessed(document.document.id).subscribe({
            next: () => {
                // Remove the processed document from the list
                this.documents = this.documents.filter((doc) => doc.document.id !== document.document.id);
                this.closeDialog();
            },
            error: (error) => {
                console.log("markDocumentProcessed Error:", error);
                // Handle the error here
            },
        });
    }

    /**
     * showDetails: call the api service to get the document image and display it in a dialog.
     */
    public showDetails(document: Documents): void {
        this.aService.getDocumentImage(document.document.id).subscribe((image: Blob) => {
            const url = window.URL.createObjectURL(image);
            const safeUrl: SafeUrl = this.sanitizer.bypassSecurityTrustUrl(url);
            this.selectedDocument = {...document, imageDataURL: safeUrl};
            this.displayDialog = true;
        });
    }

    /**
     * closeDialog: close the dialog after pressing the process button.
     */
    private closeDialog(): void {
        this.displayDialog = false;
        this.selectedDocument = null;
    }

    private registerServerSentEvent(): void {
        const sseUrl = `${this.aService.config.url}api/v1/sse/document`;
        this.sseSubscription = this.sseService.getServerSentEvent(sseUrl).subscribe((event: MessageEvent) => {
            const documentEvent = JSON.parse(event.data);
            const eventType = documentEvent.type;
            const eventData = documentEvent.data;

            switch (eventType) {
                case "NewDocument": {
                    // Process new document event
                    break;
                }

                case "ViewDocument": {
                    // Process view document event
                    break;
                }

                case "WatchlistStatusUpdate": {
                    // Process watchlist status update event
                    break;
                }

                case "DocumentProcessed": {
                    // Process document processed event
                    const processedDocumentId = eventData.documentId;
                    this.updateProcessedDocument(processedDocumentId);
                    break;
                }
            }
        });
    }

    private updateProcessedDocument(processedDocumentId: string): void {
        // Find the processed document in the documents list
        const processedDocumentIndex = this.documents.findIndex((doc) => doc.document.id === processedDocumentId);
        if (processedDocumentIndex !== -1) {
            // Remove the processed document from the list
            this.documents.splice(processedDocumentIndex, 1);
            // Update any other UI-related logic or perform additional actions as needed
        }
    }

    private closeServerSentEvent(): void {
        if (this.sseSubscription) {
            this.sseSubscription.unsubscribe();
            this.sseService.closeEventSource();
        }
    }
}

ve bu benim belgeyi vs. alan servisım

[a.service.ts]

public getDocuments(): Observable<Documents[]> {
        let url = this.config.url;
        if (!url.endsWith("/")) {
            url += "/";
        }

        return this.http
            .get<Documents[]>(`${url}api/v1/document/`, {
                headers: {
                    authorization: this.config.authorization,
                },
            })
            .pipe(
                map((data) => {
                    return data;
                }),
                catchError((error) => {
                    throw error;
                })
            );
    }

    /**
     * markDocumentProcessed: calls the api service to mark the document as processed.
     *
     */
    public markDocumentProcessed(documentId: string): Observable<Documents[]> {
        let url = this.config.url;
        if (!url.endsWith("/")) {
            url += "/";
        }

        const requestBody = {
            DocumentId: documentId,
        };

        return this.http
            .post<Documents[]>(`${url}api/v1/document/processed`, requestBody, {
                headers: {
                    authorization: this.config.authorization,
                },
            })
            .pipe(
                map((data) => {
                    return data;
                }),
                catchError((error) => {
                    throw error;
                })
            );
    }

    /**
     * getDocumentImage: calls the api service to get the document image.
     *
     */
    public getDocumentImage(documentId: string): Observable<Blob> {
        let url = this.config.url;
        if (!url.endsWith("/")) {
            url += "/";
        }

        return this.http.get(`${url}api/v1/document/${documentId}/image/Photo`, {
            responseType: "blob",
            headers: {
                authorization: this.config.authorization,
            },
        });
    }

Konsolda hiçbir hata almıyorum, endpoint ile 200 alıyorum ama test istemcim olayları otomatik olarak almıyor ve değişiklikleri görmek için yenilemem gerekiyor.

Cevap #1

Sorun şu ki SSE uç noktasını normal bir GET uç noktası olarak kullanıyorsunuz. Elbette güncellemeleri yalnızca siz talep ederseniz alırsınız, sizin durumunuzda sayfayı yeniden yükleyerek.

Bunu yapıyorsunuz (SSE servis ekran görüntünüzden alınmıştır):

public getServerSentEvent(url: string): Observable<MessageEvent> {
  const token = this.aService.config.authorization;
  // WARNING - SSE doesn't have an API for Headers, this won't work if used as proper SSE
  const headers = new HttpHeaders({
    Authorization: token,
  });

  return new Observable<MessageEvent>((observer) => {
    // HERE - you are not creating the SSE Event Source, just getting from it
    this.http.get(url, {headers, responseType: "text"}).subscribe({
      //...
    })
  })
}

Yani SSE uç noktanızda HTTP GET çağrısı yapıyorsunuz ve sonra bunu bir SSE olayı gibi göstermeye çalışıyorsunuz. Gördüğünüz gibi bu işe yaramıyor.

Yeni bir EventSource açmalısınız ve bu aslında SSE'nin olması gerektiği gibi çalışacaktır:

// Creates SSE event source, handles SSE events
protected createSseEventSource(): void {
  // Close event source if current instance of SSE service has some
  if (this.eventSource) {
    this.closeSseConnection();
    this.eventSource = null;
  }
  // Open new channel, create new EventSource - provide your own URL
  this.eventSource = new EventSource(this.yourSSEurl);

  // Process default event
  this.eventSource.onmessage = (event: MessageEvent) => {
    this.zone.run(() => this.processSseEvent(event));
  };

  // Add your own EVENTS handling...
  /*
    enum SSE_EVENTS [
      NewDocument: 'NewDocument',
      //...
    ]
  */
  Object.keys(SSE_EVENTS).forEach(key => {
    this.eventSource.addEventListener(SSE_EVENTS[key], event => {
      this.zone.run(() => this.processSseEvent(event));
    });
  });

  // Process connection opened
  this.eventSource.onopen = () => {
    this.reconnectFrequencySec = 1;
  };

  // Process error
  this.eventSource.onerror = (error: any) => {
    this.reconnectOnError();
  };
}

Eski cevabıma bak: Hata durumunda SSE yeniden bağlanmayı yönetme. Orada ihtiyacınız olan her şeyle neredeyse çalışan bir SSE servisi bulabilirsiniz.

Cevap #2

import { Injectable, NgZone } from '@angular/core';
import { Observable, ReplaySubject } from 'rxjs';
import {
  EventSourceController,
  EventSourcePlus,
  SseMessage,
} from 'event-source-plus';
import { CompanyService } from './company.service';

@Injectable({
  providedIn: 'root',
})
export class SseService {
  private eventSource: EventSourcePlus | null = null;
  companyId!: string;
  private isConnected = false;
  private controller!: EventSourceController;

  constructor(private zone: NgZone, companyService: CompanyService) {
    companyService.selectedCompany$.subscribe((company) => {
      this.companyId =
        company?.companyProfileId || localStorage.getItem('companyId') || '';
    });
  }

  /**
   * Connects to an SSE endpoint and provides an observable for events.
   * @param url - The URL for the SSE endpoint.
   * @param headers - Optional headers for the request.
   * @returns An Observable emitting MessageEvent data.
   */
  connect(
    url: string,
    headers?: Record<string, string>
  ): Observable<MessageEvent> {
    const subject = new ReplaySubject<MessageEvent>(1); // Buffer the last emitted event

    // Close any existing connection
    this.close();

    // Initialize EventSourcePlus with custom headers if provided
    this.eventSource = new EventSourcePlus(url, {
      retryStrategy: 'always',
      maxRetryCount: 4,
      method: 'get',
      headers: {
        'Content-Type': 'application/json',
        'x-company-id': `${
          this.companyId || localStorage.getItem('companyId') || ''
        }`,
        Authorization: `Bearer ${localStorage.getItem('token')}`,
        refreshToken: `Bearer ${localStorage.getItem('refresh')}`,
      },
    });

    this.isConnected = true;
    this.eventSource.retryInterval = 3000;
    // Listen for messages
    this.controller = this.eventSource.listen({
      onMessage: (res: SseMessage) => {
        try {
          const message = res.data; // Attempt to parse JSON
          this.zone.run(() => {
            subject.next(new MessageEvent('message', { data: message }));
          });
        } catch (error) {
          console.error('Failed to parse message:', res.data, error); // Log the raw data and error
        }
      },

      // Log request errors
      onRequestError: ({ request, error }) => {
        console.error('[Request Error]', request, error);
        this.reconnect(url, headers, subject);
      },

      // Handle response errors (e.g., server-side issues)
      onResponseError: ({ response }) => {
        console.error('[Response Error]', response.status, response.body);
        if (response.status >= 400) {
          this.reconnect(url, headers, subject);
        }
      },
    });

    return subject.asObservable();
  }

    
  close(): void {
    if (this.eventSource) {
      this.controller.abort();  

      this.eventSource = null;
      this.isConnected = false;
      console.log('SSE Connection closed');
    }
  }

  /**
   * Reconnects to the SSE endpoint after a delay.
   * @param url - The URL for the SSE endpoint.
   * @param headers - Optional headers for the request.
   * @param subject - The subject to continue emitting events.
   */
  private reconnect(
    url: string,
    headers: Record<string, string> | undefined,
    subject: ReplaySubject<MessageEvent>
  ): void {
    if (this.isConnected) return; // Prevent duplicate connections

    console.log('Reconnecting to SSE in 1 second...');
    setTimeout(() => {
      this.zone.run(() => {
        this.connect(url, headers).subscribe(subject);
      });
    }, 1000);
  }
}

   // To implement it,

    const url = `https://example.com/v1/notification/subscribe`;
  
    this.sseSubscription = this.sseService.connect(url).subscribe({
      next: (event) => {
        const data = JSON.parse(event.data);
        // TOAST SERVICE
        this.toast.success(data.data, data.timestamp);
      },
      error: (err) => {
        console.error('SSE Error:', err);
      },
    });`