add: Worker for Healthcheck (#419)

Added capacity to use Worker for Healthchecks. We also allow users to override the worker URL so they are able to modify it as they see fit.

Author: filipecabaco

package-lock.json

@@ -56,6 +56,7 @@
     "sinon": "^9.0.2",
     "typedoc": "^0.22.16",
     "typescript": "^4.0.3",
-    "vitest": "^2.0.5"
+    "vitest": "^2.0.5",
+    "web-worker": "1.2.0"
   }
 }

package.json

@@ -38,6 +38,8 @@ export type RealtimeClientOptions = {
   params?: { [key: string]: any }
   log_level?: 'info' | 'debug' | 'warn' | 'error'
   fetch?: Fetch
+  worker?: boolean
+  workerUrl?: string
 }
 
 export type RealtimeMessage = {
@@ -104,6 +106,9 @@ export default class RealtimeClient {
     message: [],
   }
   fetch: Fetch
+  worker?: boolean
+  workerUrl?: string
+  workerRef?: Worker
 
   /**
    * Initializes the Socket.
@@ -119,6 +124,8 @@ export default class RealtimeClient {
    * @param options.encode The function to encode outgoing messages. Defaults to JSON: (payload, callback) => callback(JSON.stringify(payload))
    * @param options.decode The function to decode incoming messages. Defaults to Serializer's decode.
    * @param options.reconnectAfterMs he optional function that returns the millsec reconnect interval. Defaults to stepped backoff off.
+   * @param options.worker Use Web Worker to set a side flow. Defaults to false.
+   * @param options.workerUrl The URL of the worker script. Defaults to https://realtime.supabase.com/worker.js that includes a heartbeat event call to keep the connection alive.
    */
   constructor(endPoint: string, options?: RealtimeClientOptions) {
     this.endPoint = `${endPoint}/${TRANSPORTS.websocket}`
@@ -160,6 +167,14 @@ export default class RealtimeClient {
     }, this.reconnectAfterMs)
 
     this.fetch = this._resolveFetch(options?.fetch)
+    if (options?.worker) {
+      if (typeof window !== 'undefined' && !window.Worker) {
+        throw new Error('Web Worker is not supported')
+      }
+      this.worker = options?.worker || false
+      this.workerUrl =
+        options?.workerUrl || 'https://realtime.supabase.com/worker.js'
+    }
   }
 
   /**
@@ -452,11 +467,31 @@ export default class RealtimeClient {
     this.log('transport', `connected to ${this._endPointURL()}`)
     this._flushSendBuffer()
     this.reconnectTimer.reset()
-    this.heartbeatTimer && clearInterval(this.heartbeatTimer)
-    this.heartbeatTimer = setInterval(
-      () => this._sendHeartbeat(),
-      this.heartbeatIntervalMs
-    )
+    if (!this.worker) {
+      this.heartbeatTimer && clearInterval(this.heartbeatTimer)
+      this.heartbeatTimer = setInterval(
+        () => this._sendHeartbeat(),
+        this.heartbeatIntervalMs
+      )
+    } else {
+      this.log('worker', `starting worker for from ${this.workerUrl!}`)
+
+      this.workerRef = new Worker(this.workerUrl!)
+      this.workerRef.onerror = (error) => {
+        this.log('worker', 'worker error', error.message)
+        this.workerRef!.terminate()
+      }
+      this.workerRef.onmessage = (event) => {
+        if (event.data.event === 'keepAlive') {
+          this._sendHeartbeat()
+        }
+      }
+      this.workerRef.postMessage({
+        event: 'start',
+        interval: this.heartbeatIntervalMs,
+      })
+    }
+
     this.stateChangeCallbacks.open.forEach((callback) => callback())!
   }
 

src/RealtimeClient.ts

@@ -1,10 +1,19 @@
 import assert from 'assert'
 import sinon from 'sinon'
-import { describe, beforeEach, afterEach, test } from 'vitest'
+import {
+  describe,
+  beforeEach,
+  afterEach,
+  test,
+  beforeAll,
+  afterAll,
+} from 'vitest'
 
 import RealtimeClient from '../src/RealtimeClient'
 import RealtimeChannel from '../src/RealtimeChannel'
 import { Response } from '@supabase/node-fetch'
+import { WebSocketServer } from 'ws'
+import Worker from 'web-worker'
 
 let channel, socket
 const defaultRef = '1'
@@ -1418,3 +1427,34 @@ describe('trigger', () => {
     assert.equal(client.accessToken, '123')
   })
 })
+
+describe('worker', () => {
+  let client
+  let mockServer
+
+  beforeAll(() => {
+    window.Worker = Worker
+    mockServer = new WebSocketServer({ port: 8080 })
+  })
+
+  afterAll(() => {
+    window.Worker = undefined
+    mockServer.close()
+  })
+
+  beforeEach(() => {
+    client = new RealtimeClient('ws://localhost:8080/socket', {
+      worker: true,
+      workerUrl: 'https://realtime.supabase.com/worker.js',
+      heartbeatIntervalMs: 10,
+    })
+  })
+
+  test('sets worker flag', () => {
+    assert.ok(client.worker)
+  })
+
+  test('sets worker URL', () => {
+    assert.equal(client.workerUrl, 'https://realtime.supabase.com/worker.js')
+  })
+})

test/channel.test.ts

@@ -13,6 +13,5 @@
     "moduleResolution": "Node",
     "forceConsistentCasingInFileNames": true,
     "stripInternal": true,
-    "types": ["vitest/globals"]
   }
 }