fix: check storage.subscribe return value before calling onSubscribe callback

Author: RomainLanz

src/stream_manager.ts

@@ -105,7 +105,12 @@ export class StreamManager<Context extends unknown> {
       }
     }
 
-    this.#storage.subscribe(uid, channel)
+    const subscribed = this.#storage.subscribe(uid, channel)
+
+    if (!subscribed) {
+      return false
+    }
+
     onSubscribe?.({ uid, channel, context: context! })
 
     return true

tests/stream_manager.spec.ts

@@ -110,9 +110,41 @@ test.group('StreamManager', () => {
   })
 
   test('should subscribe to a channel', async ({ assert }) => {
+    const socket = new Socket()
+    const manager = new StreamManager()
+    const request = new IncomingMessage(socket)
+    const response = new ServerResponse(request)
+
+    const stream = manager.createStream({
+      uid: randomUUID(),
+      request,
+      response,
+      context: {},
+    })
+
+    assert.isTrue(await manager.subscribe({ uid: stream.getUid(), channel: 'foo', context: {} }))
+  })
+
+  test('should return false when subscribing with non-existent uid', async ({ assert }) => {
+    const manager = new StreamManager()
+
+    assert.isFalse(await manager.subscribe({ uid: randomUUID(), channel: 'foo', context: {} }))
+  })
+
+  test('should not call onSubscribe when uid does not exist', async ({ assert }) => {
     const manager = new StreamManager()
+    let subscribed = false
 
-    assert.isTrue(await manager.subscribe({ uid: randomUUID(), channel: 'foo', context: {} }))
+    await manager.subscribe({
+      uid: randomUUID(),
+      channel: 'foo',
+      context: {},
+      onSubscribe() {
+        subscribed = true
+      },
+    })
+
+    assert.isFalse(subscribed)
   })
 
   test('should not subscribe to a channel if not authorized', async ({ assert }) => {
@@ -124,11 +156,21 @@ test.group('StreamManager', () => {
   })
 
   test('should call onSubscribe callback', async ({ assert }) => {
+    const socket = new Socket()
     const manager = new StreamManager()
+    const request = new IncomingMessage(socket)
+    const response = new ServerResponse(request)
     let subscribed = false
 
-    await manager.subscribe({
+    const stream = manager.createStream({
       uid: randomUUID(),
+      request,
+      response,
+      context: {},
+    })
+
+    await manager.subscribe({
+      uid: stream.getUid(),
       channel: 'foo',
       context: {},
       onSubscribe() {

tests/transmit.spec.ts

@@ -62,13 +62,13 @@ test.group('Transmit', () => {
       transport: null,
     })
 
-    const uid = randomUUID()
+    const stream = makeStream(transmit)
 
     transmit.authorize('channel1', () => {
       return true
     })
 
-    const authorized = await transmit.subscribe({ channel: 'channel1', uid })
+    const authorized = await transmit.subscribe({ channel: 'channel1', uid: stream.getUid() })
 
     assert.isTrue(authorized)
   })
@@ -94,14 +94,14 @@ test.group('Transmit', () => {
       transport: null,
     })
 
-    const uid = randomUUID()
+    const stream = makeStream(transmit)
 
     transmit.authorize<{ id: string }>('channel/:id', (_context, params) => {
       return params.id === '1'
     })
 
-    const authorized = await transmit.subscribe({ channel: 'channel/1', uid })
-    const refused = await transmit.subscribe({ channel: 'channel/2', uid })
+    const authorized = await transmit.subscribe({ channel: 'channel/1', uid: stream.getUid() })
+    const refused = await transmit.subscribe({ channel: 'channel/2', uid: stream.getUid() })
 
     assert.isTrue(authorized)
     assert.isFalse(refused)
@@ -138,18 +138,18 @@ test.group('Transmit', () => {
       transport: null,
     })
 
-    const uid = randomUUID()
+    const stream = makeStream(transmit)
     let subscribed = false
 
     transmit.on('subscribe', (params) => {
       subscribed = true
 
-      assert.equal(params.uid, uid)
+      assert.equal(params.uid, stream.getUid())
       assert.equal(params.channel, 'users/1')
     })
 
     await transmit.subscribe({
-      uid,
+      uid: stream.getUid(),
       channel: 'users/1',
     })