fix: move back to an array solution (#467)

Author: filipecabaco

src/RealtimeChannel.ts

@@ -279,6 +279,7 @@ export default class RealtimeChannel {
                 })
               } else {
                 this.unsubscribe()
+                this.state = CHANNEL_STATES.errored
                 callback?.(
                   REALTIME_SUBSCRIBE_STATES.CHANNEL_ERROR,
                   new Error(
@@ -296,6 +297,7 @@ export default class RealtimeChannel {
           }
         })
         .receive('error', (error: { [key: string]: any }) => {
+          this.state = CHANNEL_STATES.errored
           callback?.(
             REALTIME_SUBSCRIBE_STATES.CHANNEL_ERROR,
             new Error(

src/RealtimeClient.ts

@@ -91,7 +91,7 @@ const WORKER_SCRIPT = `
 export default class RealtimeClient {
   accessTokenValue: string | null = null
   apiKey: string | null = null
-  channels: Set<RealtimeChannel> = new Set()
+  channels: RealtimeChannel[] = new Array()
   endPoint: string = ''
   httpEndpoint: string = ''
   headers?: { [key: string]: string } = DEFAULT_HEADERS
@@ -262,7 +262,7 @@ export default class RealtimeClient {
    * Returns all created channels
    */
   getChannels(): RealtimeChannel[] {
-    return Array.from(this.channels)
+    return this.channels
   }
 
   /**
@@ -273,9 +273,11 @@ export default class RealtimeClient {
     channel: RealtimeChannel
   ): Promise<RealtimeRemoveChannelResponse> {
     const status = await channel.unsubscribe()
-    if (this.channels.size === 0) {
+    this.channels = this.channels.filter((c) => c._joinRef !== channel._joinRef)
+    if (this.channels.length === 0) {
       this.disconnect()
     }
+
     return status
   }
 
@@ -284,13 +286,10 @@ export default class RealtimeClient {
    */
   async removeAllChannels(): Promise<RealtimeRemoveChannelResponse[]> {
     const values_1 = await Promise.all(
-      Array.from(this.channels).map((channel) => {
-        this.channels.delete(channel)
-        return channel.unsubscribe()
-      })
+      this.channels.map((channel) => channel.unsubscribe())
     )
     this.disconnect()
-
+    this.channels = []
     return values_1
   }
 
@@ -337,7 +336,8 @@ export default class RealtimeClient {
 
     if (!exists) {
       const chan = new RealtimeChannel(`realtime:${topic}`, params, this)
-      this.channels.add(chan)
+      this.channels.push(chan)
+
       return chan
     } else {
       return exists
@@ -480,7 +480,7 @@ export default class RealtimeClient {
    * @internal
    */
   _leaveOpenTopic(topic: string): void {
-    let dupChannel = Array.from(this.channels).find(
+    let dupChannel = this.channels.find(
       (c) => c.topic === topic && (c._isJoined() || c._isJoining())
     )
     if (dupChannel) {
@@ -497,7 +497,7 @@ export default class RealtimeClient {
    * @internal
    */
   _remove(channel: RealtimeChannel) {
-    this.channels.delete(channel)
+    this.channels = this.channels.filter((c) => c.topic !== channel.topic)
   }
 
   /**

test/channel.test.ts

@@ -17,6 +17,7 @@ import RealtimeChannel from '../src/RealtimeChannel'
 import { Response } from '@supabase/node-fetch'
 import Worker from 'web-worker'
 import { Server, WebSocket } from 'mock-socket'
+import { CHANNEL_STATES } from '../src/lib/constants'
 
 const defaultRef = '1'
 const defaultTimeout = 1000
@@ -303,19 +304,19 @@ describe('subscribe', () => {
 
   test('unsubscribes to channel with incorrect server postgres_changes resp', () => {
     const unsubscribeSpy = sinon.spy(channel, 'unsubscribe')
-    const cbSpy = sinon.spy()
-    const func = () => {}
+    const callbackSpy = sinon.spy()
+    const dummyCallback = () => {}
 
     channel.bindings.postgres_changes = [
       {
         type: 'postgres_changes',
         filter: { event: '*', schema: '*' },
-        callback: func,
+        callback: dummyCallback,
       },
       {
         type: 'postgres_changes',
         filter: { event: 'INSERT', schema: 'public', table: 'test' },
-        callback: func,
+        callback: dummyCallback,
       },
       {
         type: 'postgres_changes',
@@ -325,27 +326,28 @@ describe('subscribe', () => {
           table: 'test',
           filter: 'id=eq.1',
         },
-        callback: func,
+        callback: dummyCallback,
       },
     ]
-    channel.subscribe(cbSpy)
 
-    const cb = channel.bindings['chan_reply_1'][0].callback
-    cb({
+    channel.subscribe(callbackSpy)
+    const replyCallback = channel.bindings['chan_reply_1'][0].callback
+    replyCallback({
       status: 'ok',
       response: { postgres_changes: [{ id: 'abc', event: '*', schema: '*' }] },
     })
 
     assert.ok(unsubscribeSpy.calledOnce)
     assert.ok(
-      cbSpy.calledWith(
+      callbackSpy.calledWith(
         'CHANNEL_ERROR',
         sinon.match({
           message:
             'mismatch between server and client bindings for postgres changes',
         })
       )
     )
+    assert.equal(channel.state, CHANNEL_STATES.errored)
   })
 
   test('can set timeout on joinPush', () => {
@@ -594,6 +596,7 @@ describe('joinPush', () => {
       helpers.receiveError()
 
       assert.ok(spyError.calledOnce)
+      assert.equal(channel.state, CHANNEL_STATES.errored)
     })
 
     test("triggers receive('error') callback if error response already received", () => {
@@ -1113,7 +1116,7 @@ describe('leave', () => {
   })
 
   test("closes channel on 'ok' from server", () => {
-    const anotherChannel = socket.channel('another', { three: 'four' })
+    const anotherChannel = socket.channel('another')
     assert.equal(socket.getChannels().length, 2)
 
     channel.unsubscribe()