implements sse events store, and update sse plugin and service

pages/cmf_licence/organization.vue

@@ -30,36 +30,39 @@
         </v-btn>
       </div>
     </v-form>
+    <!-- <<< for debugging purposes, TODO: remove before deploying -->
+    <ul>
+      <li v-for="event in events">{{ event }}</li>
+    </ul>
+    <!-- >>> for debugging purposes -->
   </div>
 </template>
 
 <script lang="ts">
-import {defineComponent, onBeforeUnmount, onMounted, Ref, ref, useContext} from "@nuxtjs/composition-api";
+import {defineComponent, Ref, ref, useContext} from "@nuxtjs/composition-api";
 import {HTTP_METHOD, QUERY_TYPE} from "~/types/enums";
 import DataPersister from "~/services/data/dataPersister";
 import {DataPersisterArgs} from "~/types/interfaces";
-import { EventSourcePolyfill } from "event-source-polyfill";
 
 export default defineComponent({
   name: 'OrganizationCmfLicence',
   setup() {
-    const { $config, store } = useContext()
+    const { store } = useContext()
 
     let pending: Ref<boolean> = ref(false)
     let fileUrl: Ref<string | null> = ref(null)
-    let eventSource: EventSource | null = null
 
-    const sseOpened = () => {
-      return eventSource !== null && eventSource.readyState === EventSource.OPEN
-    }
+    const async = () => { return store.state.sse.connected }
+
+    let events: Ref<Array<Array<any>>> = ref(store.state.sse.events)
 
     const submit = async () => {
       const response = await DataPersister.request(
         '/api/export/cmf-licence/organization',
         HTTP_METHOD.POST,
-        { type: QUERY_TYPE.DEFAULT, data: { format: 'pdf', async: sseOpened() }, withCredentials: true } as DataPersisterArgs
+        { type: QUERY_TYPE.DEFAULT, data: { format: 'pdf', async: true }, withCredentials: true } as DataPersisterArgs
       )
-      if (sseOpened()) {
+      if (async()) {
         pending.value = true
       } else {
         console.warn('File downloaded synchronously')
@@ -67,63 +70,21 @@ export default defineComponent({
       }
     }
 
-    const unsubscribe = async () => {
-      if (eventSource === null || eventSource.readyState === EventSource.CLOSED) {
-        return
-      }
-      console.log('SSE - Close subscription')
-      eventSource.close()
-    }
-
-    const subscribe = async () => {
-      if (sseOpened()) {
-        console.error('SSE - Already subscribed to the event source')
-        return
-      }
-      if (process.server) {
-        console.error('SSE - Cannot subscribe on server side')
-        return
-      }
-
-      const url = new URL($config.baseUrl_mercure)
-      url.searchParams.append('topic', "access/" + store.state.profile.access.id)
-
-      eventSource = new EventSourcePolyfill(
-        url.toString(),
-        { withCredentials: true }
-      );
-
-      eventSource.onerror = (event) => {
-        console.error('SSE - An error happened : ' + JSON.stringify(event))
-        eventSource?.close()
-      }
-      eventSource.onopen = () => {
-        console.log('SSE - Listening for events...')
-      }
-      eventSource.onmessage = event => {
-        const data = JSON.parse(event.data)
-        fileUrl.value = data.url
-        pending.value = false
-      }
-    }
-
-    // onMounted(() => {
-    //   subscribe()
-    // })
-    //
-    // onBeforeUnmount(() => {
-    //   unsubscribe()
-    // })
-    //
-    // if (process.browser) {
-    //   window.addEventListener('beforeunload', unsubscribe)
+    // const onMessage = (eventData) => {
+    //   fileUrl.value = eventData.url
+    //   pending.value = false
     // }
 
     return {
       submit,
       pending,
-      fileUrl
+      fileUrl,
+      events
     }
   }
 })
+
+function useSse(): { sseConnected: any; } {
+    throw new Error("Function not implemented.");
+}
 </script>

plugins/sse.ts

@@ -3,10 +3,18 @@ import SseSource from "~/services/sse/sseSource";
 
 /**
  * Setup SSE EventSource, allowing the app to listen for Mercure updates
+ * /!\ This has to be executed client side
+ *
  * @param ctx
  */
 const ssePlugin: Plugin = ({ $config, store }) => {
-  const sseSource = new SseSource($config.baseUrl_mercure, store.state.profile.access.id)
+  const sseSource = new SseSource(
+    $config.baseUrl_mercure,
+    "access/" + store.state.profile.access.id,
+    () => { store.commit('sse/setConnected', true) },
+    (eventData) => { store.commit('sse/addEvent', eventData) },
+    () => { store.commit('sse/setConnected', false) },
+  )
 
   sseSource.subscribe()
   window.addEventListener('beforeunload', () => { sseSource.unsubscribe() })

services/sse/sseSource.ts

@@ -1,26 +1,35 @@
 import { EventSourcePolyfill } from "event-source-polyfill";
 
 class SseSource {
-  private readonly accessId: number
   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(mercureBaseUrl: string, accessId: number) {
-    this.accessId = accessId
-    this.url = new URL(mercureBaseUrl)
-    this.url.searchParams.append('topic', "access/" + this.accessId)
+  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
   }
 
-  getUrl () {
-    return this.url
-  }
-
-  isOpen () {
+  isConnected () {
     return this.eventSource !== null && this.eventSource.readyState === EventSource.OPEN
   }
 
   public subscribe () {
-    if (this.isOpen()) {
+    if (this.isConnected()) {
       throw new Error('SSE - Already subscribed to this event source')
     }
     if (process.server) {
@@ -29,19 +38,20 @@ class SseSource {
 
     this.eventSource = new EventSourcePolyfill(
       this.url.toString(),
-      { withCredentials: true }
+      { withCredentials: this.withCredentials }
     );
 
     this.eventSource.onerror = (event) => {
       console.error('SSE - An error happened : ' + JSON.stringify(event))
-      this.eventSource?.close()
+      this.unsubscribe()
     }
     this.eventSource.onopen = () => {
       console.log('SSE - Listening for events...')
+      this.onOpen()
     }
     this.eventSource.onmessage = event => {
       const data = JSON.parse(event.data)
-      console.log(data)
+      this.onMessage(data)
     }
   }
 
@@ -50,6 +60,7 @@ class SseSource {
       return
     }
     this.eventSource.close()
+    this.onClose()
     console.log('SSE - Subscription closed')
   }
 }

store/sse.ts

@@ -0,0 +1,15 @@
+import { sseState } from "~/types/interfaces";
+
+export const state = () => ({
+  connected: false,
+  events: []
+})
+
+export const mutations = {
+  setConnected(state: sseState, connected: boolean) {
+    state.connected = connected
+  },
+  addEvent(state: sseState, event: Array<any>) {
+    state.events.push(event)
+  }
+}

types/interfaces.d.ts

@@ -289,3 +289,7 @@ interface HydraMetadata {
   type?: METADATA_TYPE
 }
 
+interface sseState {
+  connected: boolean,
+  events: Array<Array<any>>,
+}