I'm getting this onmessage error while implementing SSE

Question

I'm trying to implement SSE and running into this error regarding onmessage. This is my code:

import {Injectable} from "@angular/core";
import {Observable} from "rxjs";
import {AService} from "./a.service";
import {EventSourcePolyfill} from "event-source-polyfill";

@Injectable({
    providedIn: "root",
})
export class SseService {
    private eventSource!: EventSourcePolyfill | null;

    public constructor(private aService: AService) {}

    public getServerSentEvent(url: string): Observable<MessageEvent> {
        const token = this.aService.config.authorization;
        const headers = {Authorization: `Bearer ${token}`};

        return new Observable<MessageEvent>((observer) => {
            this.eventSource = new EventSourcePolyfill(url, {
                withCredentials: false,
                headers,
            });

            this.eventSource.onmessage = (event: MessageEvent): void => {
                observer.next(event);
            };

            this.eventSource.onerror = (error): void => {
                observer.error(error);
            };

            return () => {
                this.eventSource?.close();
            };
        });
    }

    public closeEventSource(): void {
        if (this.eventSource) {
            this.eventSource.close();
            this.eventSource = null;
        }
    }
}

I'm getting this error on line this.eventSource.onmessage = (event:MessageEvent)...

Type '(event: MessageEvent) => void' is not assignable to type '(this: EventSource, ev: MessageEvent) => any'. Types of parameters 'event' and 'ev' are incompatible. Type 'MessageEvent' is missing the following properties from type 'MessageEvent': origin, ports, source, initMessageEvent, and 20 more.ts(2322) (property) SseService.eventSource: EventSourcePolyfill

I tried adding a custom message event but that didn't fix my issue and typescript doesn't allow "any" type so that's out of the question. I'm not sure if this is of any relevance to the solution but I do need to pass the token as a header so passing it through as a query parameter is out of the question.

mat.hudak · Answer 1 · 2023-06-30T05:36:09.510

I believe you have to run event handling in the NgZone, since it's an event coming outside of Angular. Not sure about the void as a return value though, try to remove it, it should not be a problem. But check the zone event handling first.

import { NgZone } from "@angular/core";

constructor(private zone: NgZone) {}

this.eventSource.onmessage = (event: MessageEvent) => {
  this.zone.run(() => observer.next(event));
};

EDIT

Try maybe a different approach. Don't create the EventSource within the Observable but rather outside of it. But use the Observable to propagate the data outside of the service.

{
  // Open new channel, create new EventSource
  this.eventSource = new EventSource(this.sseChannelUrl);

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

// Processes custom event types
private processSseEvent(sseEvent: MessageEvent): void {
  const parsed = sseEvent.data ? JSON.parse(sseEvent.data) : {};
  switch (sseEvent.type) {
    case SSE_EVENTS.STATUS: {
      // PUSH TO A SUBJECT HERE AND HAVE IT EXPOSED AS AN OBSERVABLE
      this.someReplaySubject$.next(parsed);
      break;
      /*
      Class declarations:
      private someReplaySubject$: ReplaySubject<TYPE> = new ReplaySubject<TYPE>(1);
      someExposedData$: Observable<TYPE> = this.someReplaySubject$.asObservable();
     */
    }
    // Add others if neccessary
    default: {
      console.error('Unknown event:', sseEvent.type);
      break;
    }
  }
}

See the full code here.

this.eventSource.onmessage = (event: MessageEvent) => { this.zone.run(() => observer.next(event)); }; update it like you said, still getting the same error though — mksk23, Jun 29 '23 at 18:32

I'm getting this onmessage error while implementing SSE

1 Answers1