fix(ws): subscription unsubscribing

Author: denny-il

packages/application/lib/events.ts

@@ -105,7 +105,7 @@ export class EventManager<
     const subscription = subscriptions.get(eventKey)
     if (!subscription) return false
     await this.subManager.unsubscribe(subscription)
-    subscription.emit('unsubscribe')
+    subscription.end()
     subscriptions.delete(eventKey)
   }
 

packages/application/lib/subscription.ts

@@ -14,7 +14,6 @@ export class Subscription<E extends Event = Event> extends PassThrough {
     readonly unsubscribe: () => Promise<any>,
   ) {
     super({ writableObjectMode: true, readableObjectMode: true })
-    this.once('unsubscribe', () => this.end())
   }
 }
 

packages/client-websockets/index.ts

@@ -439,7 +439,8 @@ class WebsocketsClient<
   [KEY[MessageType.ServerUnsubscribe]](buffer: ArrayBuffer) {
     const [key] = JSON.parse(decodeText(buffer))
     const subscription = this.subscriptions.get(key)
-    subscription?.unsubscribe()
+    subscription?.emit('end')
+    this.subscriptions.delete(key)
   }
 }
 

packages/transport-websockets/lib/server.ts

@@ -338,12 +338,21 @@ export class WebsocketsTransportServer extends BaseHttpTransportServer {
       },
       close: async (ws, code, message) => {
         const { id, container, streams, subscriptions } = ws.getUserData()
+        this.logger.debug(
+          'Close websocket [%s] with code [%s]: %s',
+          id,
+          code,
+          decodeText(message),
+        )
         this.sockets.delete(ws)
         this.application.connections.remove(id)
-        for (const _streams of [streams.up, streams.down, subscriptions]) {
+        for (const _streams of [streams.up, streams.down]) {
           for (const stream of _streams.values()) stream.destroy()
           _streams.clear()
         }
+        for (const subscription of subscriptions.values()) {
+          subscription.unsubscribe()
+        }
         this.handleContainerDisposal(container)
       },
     })