Merge pull request #353 from supabase/feat/realtime-subscription

feat: refactor Realtime subscription methods

Author: w3b6x9

src/SupabaseClient.ts

@@ -131,82 +131,65 @@ export default class SupabaseClient {
   }
 
   /**
-   * Remove all subscriptions.
+   * Closes and removes all subscriptions and returns a list of removed
+   * subscriptions and their errors.
    */
   async removeAllSubscriptions(): Promise<
-    (
-      | {
-          status: 'fulfilled'
-          value: {
-            error: null
-          }
-        }
-      | { status: 'rejected'; reason: { error: Error } }
-    )[]
+    { data: { subscription: RealtimeSubscription }; error: Error | null }[]
   > {
-    const subs: RealtimeSubscription[] = this.realtime.channels.slice()
-    const removeSubPromises = subs.map((sub: RealtimeSubscription) =>
-      this.removeSubscription(sub)
-        .then((): { status: 'fulfilled'; value: { error: null } } => ({
-          status: 'fulfilled',
-          value: { error: null },
-        }))
-        .catch((reason: { error: Error }): { status: 'rejected'; reason: { error: Error } } => ({
-          status: 'rejected',
-          reason,
-        }))
-    )
-    return Promise.all(removeSubPromises)
+    const allSubs: RealtimeSubscription[] = this.getSubscriptions().slice()
+    const allSubPromises = allSubs.map((sub) => this.removeSubscription(sub))
+    const allRemovedSubs = await Promise.all(allSubPromises)
+
+    return allRemovedSubs.map(({ error }, i) => {
+      return {
+        data: { subscription: allSubs[i] },
+        error,
+      }
+    })
   }
 
   /**
-   * Removes an active subscription and returns the number of open connections.
+   * Closes and removes a subscription and returns the number of open subscriptions.
    *
-   * @param subscription The subscription you want to remove.
+   * @param subscription The subscription you want to close and remove.
    */
-  removeSubscription(subscription: RealtimeSubscription): Promise<
-    | {
-        data: { openSubscriptions: number }
-        error: null
-      }
-    | { error: Error }
-  > {
-    return new Promise(async (resolve, reject) => {
-      const { error } = await this._closeSubscription(subscription)
-
-      if (error) {
-        return reject({ error })
-      }
-
-      const allSubscriptions = this.getSubscriptions()
-
-      if (allSubscriptions.length === 0) {
-        const { error } = await this.realtime.disconnect()
-
-        if (error) {
-          return reject({ error })
-        }
-      }
+  async removeSubscription(
+    subscription: RealtimeSubscription
+  ): Promise<{ data: { openSubscriptions: number }; error: Error | null }> {
+    const { error } = await this._closeSubscription(subscription)
+    const allSubs: RealtimeSubscription[] = this.getSubscriptions()
+    const openSubCount = allSubs.filter((chan) => chan.isJoined()).length
 
-      const openSubscriptionsCount = allSubscriptions.filter((chan) => chan.isJoined()).length
+    if (allSubs.length === 0) await this.realtime.disconnect()
 
-      return resolve({
-        data: { openSubscriptions: openSubscriptionsCount },
-        error: null,
-      })
-    })
+    return { data: { openSubscriptions: openSubCount }, error }
   }
 
   private async _closeSubscription(
     subscription: RealtimeSubscription
-  ): Promise<{ error: null | Error }> {
+  ): Promise<{ error: Error | null }> {
+    let error = null
+
     if (!subscription.isClosed()) {
-      return await this._closeChannel(subscription)
+      const { error: unsubError } = await this._unsubscribeSubscription(subscription)
+      error = unsubError
     }
 
+    this.realtime.remove(subscription)
+
+    return { error }
+  }
+
+  private _unsubscribeSubscription(
+    subscription: RealtimeSubscription
+  ): Promise<{ error: Error | null }> {
     return new Promise((resolve) => {
-      this.realtime.remove(subscription)
-      return resolve({ error: null })
+      subscription
+        .unsubscribe()
+        .receive('ok', () => resolve({ error: null }))
+        .receive('error', (error: Error) => resolve({ error }))
+        .receive('timeout', () => resolve({ error: new Error('timed out') }))
     })
   }
 
@@ -263,16 +246,6 @@ export default class SupabaseClient {
     return headers
   }
 
-  private _closeChannel(subscription: RealtimeSubscription): Promise<{ error: null | Error }> {
-    return new Promise((resolve, reject) => {
-      subscription
-        .unsubscribe()
-        .receive('ok', () => resolve({ error: null }))
-        .receive('error', (error: Error) => reject({ error }))
-        .receive('timeout', () => reject({ error: 'timed out' }))
-    })
-  }
-
   private _listenForMultiTabEvents() {
     if (!this.multiTab || !isBrowser() || !window?.addEventListener) {
       return null
@@ -325,7 +298,7 @@ export default class SupabaseClient {
       this.changedAccessToken = token
     } else if (event === 'SIGNED_OUT' || event === 'USER_DELETED') {
       // Token is removed
-      this.removeAllSubscriptions()
+      this.realtime.setAuth(this.supabaseKey)
       if (source == 'STORAGE') this.auth.signOut()
     }
   }