app_vue2/services/sse/sseSource.ts

import { EventSourcePolyfill } from "event-source-polyfill";

class SseSource {
  private readonly url: URL
  private readonly onOpen: (() => void)
  private readonly onMessage: ((eventData: Array<any>) => void)
  private readonly onClose: (() => void)
  private readonly withCredentials: boolean
  private eventSource: EventSource | null = null

  constructor(
    mercureHubBaseUrl: string,
    topic: string,
    onOpen: (() => void),
    onMessage: ((eventData: Array<any>) => void),
    onClose: (() => void),
    withCredentials: boolean = true
  ) {
    this.url = new URL(mercureHubBaseUrl)
    this.url.searchParams.append('topic', topic)
    this.onOpen = onOpen
    this.onMessage = onMessage
    this.onClose = onClose
    this.withCredentials = withCredentials
  }

  isConnected () {
    return this.eventSource !== null && this.eventSource.readyState === EventSource.OPEN
  }

  public subscribe () {
    if (this.isConnected()) {
      throw new Error('SSE - Already subscribed to this event source')
    }
    if (process.server) {
      throw new Error('SSE - Cannot subscribe on server side')
    }

    this.eventSource = new EventSourcePolyfill(
      this.url.toString(),
      {
        withCredentials: this.withCredentials,
        heartbeatTimeout: 45 * 1000 // in ms, timeout can not be disabled yet, so I set it very large instead
      }
    );

    this.eventSource.onerror = (event) => {
      console.error('SSE - An error happened')
    }
    this.eventSource.onopen = () => {
      console.log('SSE - Listening for events...')
      this.onOpen()
    }
    this.eventSource.onmessage = event => {
      const data = JSON.parse(event.data)
      this.onMessage(data)
    }
  }

  public unsubscribe () {
    if (this.eventSource === null || this.eventSource.readyState === EventSource.CLOSED) {
      return
    }
    this.eventSource.close()
    this.onClose()
    console.log('SSE - Subscription closed')
  }
}

export default SseSource