fix(stream): notify all callers after a join callback unsubscribes

src/stream/transport/phoenix.ts

@@ -508,22 +508,28 @@ export class PhoenixChannelsTransport implements StreamTransport {
           subscription.joined = true
           subscription.rejoinTries = 0
           this.log(`subscribed to "${subscription.topic}"`)
-          for (const callbacks of subscription.callbackList) {
-            this.safeInvoke(() => callbacks.onSubscribed?.())
+          for (const callbacks of [...subscription.callbackList]) {
+            if (subscription.callbackList.includes(callbacks)) {
+              this.safeInvoke(() => callbacks.onSubscribed?.())
+            }
           }
         } else {
           this.log(`failed to subscribe to "${subscription.topic}"`)
           const error = new Error(
             `Subscribe to "${subscription.topic}" failed: ${status}`,
           )
-          for (const callbacks of subscription.callbackList) {
-            this.safeInvoke(() => callbacks.onSubscribeError?.(error))
+          for (const callbacks of [...subscription.callbackList]) {
+            if (subscription.callbackList.includes(callbacks)) {
+              this.safeInvoke(() => callbacks.onSubscribeError?.(error))
+            }
           }
         }
       },
       reason => {
-        for (const callbacks of subscription.callbackList) {
-          this.safeInvoke(() => callbacks.onSubscribeError?.(reason))
+        for (const callbacks of [...subscription.callbackList]) {
+          if (subscription.callbackList.includes(callbacks)) {
+            this.safeInvoke(() => callbacks.onSubscribeError?.(reason))
+          }
         }
       },
     )

test/stream/transport.spec.ts

@@ -394,6 +394,98 @@ describe("subscribe acknowledgement", () => {
     transport.disconnect()
   })
 
+  test("removing a handler in its join callback does not skip the next caller", async () => {
+    const transport = makeTransport()
+    transport.connect()
+    await flushMicrotasks()
+
+    let offFirst = () => {}
+    const first = vi.fn(() => offFirst())
+    const second = vi.fn()
+    offFirst = transport
+      .subscribe("collection:c1", undefined, { onSubscribed: first })
+      .on(EventType.ITEM_SOLD, vi.fn())
+    transport
+      .subscribe("collection:c1", undefined, { onSubscribed: second })
+      .on(EventType.ITEM_SOLD, vi.fn())
+
+    const [, ref, topic] = server.framesOfType("phx_join")[0]
+    server.send(encodeReply({ ref: ref as string, topic }))
+
+    expect(first).toHaveBeenCalledTimes(1)
+    expect(second).toHaveBeenCalledTimes(1)
+    transport.disconnect()
+  })
+
+  test("a caller removed by another join callback is not notified", async () => {
+    const transport = makeTransport()
+    transport.connect()
+    await flushMicrotasks()
+
+    let offSecond = () => {}
+    const first = vi.fn(() => offSecond())
+    const second = vi.fn()
+    transport
+      .subscribe("collection:c1", undefined, { onSubscribed: first })
+      .on(EventType.ITEM_SOLD, vi.fn())
+    offSecond = transport
+      .subscribe("collection:c1", undefined, { onSubscribed: second })
+      .on(EventType.ITEM_SOLD, vi.fn())
+
+    const [, ref, topic] = server.framesOfType("phx_join")[0]
+    server.send(encodeReply({ ref: ref as string, topic }))
+
+    expect(first).toHaveBeenCalledTimes(1)
+    expect(second).not.toHaveBeenCalled()
+    transport.disconnect()
+  })
+
+  test("removing a handler in its join error callback does not skip the next caller", async () => {
+    const transport = makeTransport()
+    transport.connect()
+    await flushMicrotasks()
+
+    let offFirst = () => {}
+    const first = vi.fn(() => offFirst())
+    const second = vi.fn()
+    offFirst = transport
+      .subscribe("collection:c1", undefined, { onSubscribeError: first })
+      .on(EventType.ITEM_SOLD, vi.fn())
+    transport
+      .subscribe("collection:c1", undefined, { onSubscribeError: second })
+      .on(EventType.ITEM_SOLD, vi.fn())
+
+    const [, ref, topic] = server.framesOfType("phx_join")[0]
+    server.send(encodeReply({ ref: ref as string, topic, status: "error" }))
+
+    expect(first).toHaveBeenCalledTimes(1)
+    expect(second).toHaveBeenCalledTimes(1)
+    transport.disconnect()
+  })
+
+  test("removing a handler on join timeout does not skip the next caller", async () => {
+    vi.useFakeTimers()
+    const transport = makeTransport({ timeout: 500 })
+    transport.connect()
+    await flushMicrotasks()
+
+    let offFirst = () => {}
+    const first = vi.fn(() => offFirst())
+    const second = vi.fn()
+    offFirst = transport
+      .subscribe("collection:c1", undefined, { onSubscribeError: first })
+      .on(EventType.ITEM_SOLD, vi.fn())
+    transport
+      .subscribe("collection:c1", undefined, { onSubscribeError: second })
+      .on(EventType.ITEM_SOLD, vi.fn())
+
+    vi.advanceTimersByTime(500)
+
+    expect(first).toHaveBeenCalledTimes(1)
+    expect(second).toHaveBeenCalledTimes(1)
+    transport.disconnect()
+  })
+
   test("a second subscriber to a joined topic is told immediately", async () => {
     const transport = makeTransport()
     transport.connect()