fix(source): triggering onmessage event.

Author: razshare

package.json

@@ -1,6 +1,6 @@
 {
   "name": "sveltekit-sse",
-  "version": "0.14.2",
+  "version": "0.14.3",
   "license": "MIT",
   "scripts": {
     "dev": "vite dev",

src/lib/consume.js

@@ -48,6 +48,11 @@ export function consume({
     const id = uuid({ short: true })
     const isLocal = localAbort
     localAbort = false
+
+    if (options.onclose) {
+      options.onclose()
+    }
+
     for (const onClose of events.onClose) {
       onClose({
         id,
@@ -117,17 +122,22 @@ export function consume({
       openWhenHidden,
       method: 'POST',
       ...rest,
-      async onopen({
-        headers: headersLocal,
-        status: statusLocal,
-        statusText: statusTextLocal,
-        ok,
-      }) {
+      async onopen(response) {
+        const {
+          headers: headersLocal,
+          status: statusLocal,
+          statusText: statusTextLocal,
+          ok,
+        } = response
         status = statusLocal
         statusText = statusTextLocal
         headers = headersLocal
 
         if (ok && headers.get('content-type') === 'text/event-stream') {
+          if (options.onopen) {
+            options.onopen(response)
+          }
+
           for (const onOpen of events.onOpen) {
             onOpen({
               id: '',
@@ -148,6 +158,10 @@ export function consume({
         }
       },
       onmessage({ id, event, data }) {
+        if (options.onmessage) {
+          options.onmessage({ id, event, data })
+        }
+
         for (const onMessage of events.onMessage) {
           onMessage({
             id,
@@ -163,6 +177,10 @@ export function consume({
         }
       },
       onclose() {
+        if (options.onclose) {
+          options.onclose()
+        }
+
         for (const onClose of events.onClose) {
           onClose({
             id: '',
@@ -178,6 +196,9 @@ export function consume({
         }
       },
       onerror(error) {
+        if (options.onerror) {
+          options.onerror(error)
+        }
         for (const onError of events.onError) {
           onError({
             id: '',

src/lib/fetchEventSource.js

@@ -205,7 +205,7 @@ export function fetchEventSource(input, init = {}) {
 
       try {
         const response = await fetch(input, init)
-        await onopen(response)
+        onopen(response)
         if (!response.body) {
           reject(new Error(`empty response from source ${input}.`))
           return

src/lib/playwright/playwright.js

@@ -34,6 +34,12 @@ import { IS_BROWSER } from '../constants'
  * @property {string} message2
  */
 
+/**
+ * @typedef Issue70
+ * @property {number} status
+ * @property {import('$lib/types.external').FetchEventSourceMessage[]} onmessage
+ */
+
 /**
  * @typedef PlaywrightState
  * @property {number} counter
@@ -43,6 +49,7 @@ import { IS_BROWSER } from '../constants'
  * @property {Issue65} issue65
  * @property {Issue68} issue68
  * @property {PullRequest69} pullRequest69
+ * @property {Issue70} issue70
  */
 
 /** @type {PlaywrightState} */
@@ -70,6 +77,10 @@ let state = {
     message1: '',
     message2: '',
   },
+  issue70: {
+    onmessage: [],
+    status: -1,
+  },
 }
 
 if (IS_BROWSER) {

src/lib/source.js

@@ -24,11 +24,7 @@ function connectable({ resource, options, onClose, onError, onOpen }) {
       const consumedStream = consume({
         resource,
         options,
-        onClose(e) {
-          if (onClose) {
-            onClose(e)
-          }
-        },
+        onClose,
         onOpen,
         onError,
         onMessage(e) {

src/lib/types.external.js

@@ -203,7 +203,7 @@ export {}
  * @property {string} [method]
  * @property {any} [body]
  * @property {Record<string,string>} [headers]
- * @property {function(Response):Promise<void>} [onopen]
+ * @property {function(Response):void} [onopen]
  * @property {function(FetchEventSourceMessage):void} [onmessage]
  * @property {function():void} [onclose]
  * @property {function(Error):void} [onerror]

src/routes/issue-70/+page.svelte

@@ -0,0 +1,20 @@
+<!-- +page.svelte -->
+<script>
+  import { source } from '$lib'
+  import { playwright } from '$lib/playwright/playwright'
+
+  const value = source(`/issue-70/events`, {
+    options: {
+      onmessage(e) {
+        console.log({ onmessage: e })
+        playwright.state.issue70.onmessage.push(e)
+      },
+      onopen(e) {
+        console.log({ onopen: e })
+        playwright.state.issue70.status = e.status
+      },
+    },
+  }).select('message')
+</script>
+
+<span>Value:{$value}</span>

src/routes/issue-70/events/+server.js

@@ -0,0 +1,14 @@
+import { produce } from '$lib'
+import { delay } from '$lib/delay.js'
+
+export async function POST() {
+  return produce(async function start({ emit }) {
+    while (true) {
+      const { error } = emit('message', `hello ${Date.now()}`)
+      if (error) {
+        return
+      }
+      await delay(1000)
+    }
+  })
+}

tests/main.test.js

@@ -74,3 +74,16 @@ test('Making sure connection gets cached using non Latin1 characters.', async fu
   expect(pullRequest69.message1).toBe(pullRequest69.message2)
   expect(pullRequest69.message1).toContain('hello 猫')
 })
+
+test('Making sure onmessage event is triggered on the source object.', async function run({
+  page,
+}) {
+  // Testing changes made in response to issue 70 https://github.com/razshare/sveltekit-sse/issues/70
+  await page.goto('/issue-70')
+  await delay(1000)
+  const { issue70 } = await getPlaywrightState({ page })
+  expect(issue70.status).toBe(200)
+  expect(issue70.onmessage.length).toBeGreaterThan(0)
+  expect(issue70.onmessage[0].event).toBe('message')
+  expect(issue70.onmessage[0].id).toBe('1')
+})