Align subscription error handling (#2296)

Author: jribbink

packages/fcl-core/src/events/index.test.ts

@@ -76,7 +76,8 @@ describe("events", () => {
 
   test("subscribe should pipe the events to the callback", async () => {
     const filter = {eventTypes: ["A"]}
-    const callback = jest.fn()
+    const onData = jest.fn()
+    const onError = jest.fn()
 
     const mockEvents = [{type: "A"}, {type: "B"}] as any[]
 
@@ -85,59 +86,82 @@ describe("events", () => {
       return mockSubscription
     })
 
-    events(filter).subscribe(callback)
+    events(filter).subscribe(onData, onError)
 
     // Flush the event loop
     await new Promise(resolve => setTimeout(resolve, 0))
 
-    expect(callback.mock.calls).toEqual([
-      [mockEvents[0], null],
-      [mockEvents[1], null],
-    ])
+    expect(onData.mock.calls).toEqual([[{type: "A"}], [{type: "B"}]])
+    expect(onError).not.toHaveBeenCalled()
   })
 
   test("subscribe should pipe the errors to the callback", async () => {
     const filter = {eventTypes: ["A"]}
-    const callback = jest.fn()
+    const onData = jest.fn()
+    const onError = jest.fn()
 
     const mockError = new Error("mock error")
 
-    jest.mocked(subscribe).mockImplementation(({onError}) => {
-      onError(mockError)
+    jest.mocked(subscribe).mockImplementation(({onError: _onError}) => {
+      _onError(mockError)
       return mockSubscription
     })
 
-    events(filter).subscribe(callback)
+    events(filter).subscribe(onData, onError)
 
     // Flush the event loop
     await new Promise(resolve => setTimeout(resolve, 0))
 
-    expect(callback.mock.calls).toEqual([[null, mockError]])
+    expect(onData).not.toHaveBeenCalled()
+    expect(onError).toHaveBeenCalledTimes(1)
+    expect(onError).toHaveBeenCalledWith(mockError)
   })
 
   test("fallback to legacy polling if subscriptions are not supported", async () => {
     const filter = "A"
-    const callback = jest.fn()
+    const onData = jest.fn()
+    const onError = jest.fn()
 
-    const mockError = new SubscriptionsNotSupportedError()
+    const mockNotSupportedError = new SubscriptionsNotSupportedError()
 
-    jest.mocked(subscribe).mockImplementation(({onError}) => {
-      onError(mockError)
+    jest.mocked(subscribe).mockImplementation(({onError: _onError}) => {
+      _onError(mockNotSupportedError)
       return mockSubscription
     })
 
-    const unsubscribe = events(filter).subscribe(callback)
+    const unsubscribe = events(filter).subscribe(onData, onError)
 
     // Flush the event loop
     await new Promise(resolve => setTimeout(resolve, 0))
 
-    // Check that the callback was called
-    expect(callback).toHaveBeenCalledTimes(0)
-    expect(subscribe).toHaveBeenCalledTimes(1)
+    // Check that the error did not propagate to the onError callback
+    expect(onData).not.toHaveBeenCalled()
+    expect(onError).not.toHaveBeenCalled()
 
-    // Check that legacy polling was called
+    // Check that the legacy subscribe was called
+    expect(mockLegacySubscribeObject.subscribe).toHaveBeenCalledTimes(1)
+    expect(mockLegacyUnsubscribe).not.toHaveBeenCalled()
+
+    // Check that the legacy events were called with the correct filter
     expect(legacyEvents).toHaveBeenCalledWith(filter)
-    expect(mockLegacySubscribeObject.subscribe).toHaveBeenCalledWith(callback)
+    expect(mockLegacySubscribeObject.subscribe).toHaveBeenCalledWith(
+      expect.any(Function)
+    )
+
+    // Mock the legacy subscribe to call the onData callback
+    const legacyOnData = (
+      mockLegacySubscribeObject.subscribe.mock.calls as any
+    )[0][0] as jest.Mock
+    const mockLegacyEvents = [{type: "A"}, {type: "B"}] as any[]
+    mockLegacyEvents.forEach(event => legacyOnData(event))
+
+    // Flush the event loop
+    await new Promise(resolve => setTimeout(resolve, 0))
+
+    // Check that the onData callback was called with the legacy events
+    expect(onData.mock.calls).toEqual([[{type: "A"}], [{type: "B"}]])
+    expect(onError).not.toHaveBeenCalled()
+    expect(mockLegacyUnsubscribe).not.toHaveBeenCalled()
 
     // Unsubscribe
     unsubscribe()
@@ -151,16 +175,19 @@ describe("events", () => {
 
   test("emulator should fallback to legacy polling", async () => {
     const filter = "A"
-    const callback = jest.fn()
+    const onData = jest.fn()
+    const onError = jest.fn()
 
     jest.mocked(getChainId).mockResolvedValue("local")
 
-    events(filter).subscribe(callback)
+    events(filter).subscribe(onData, onError)
 
     // Flush the event loop
     await new Promise(resolve => setTimeout(resolve, 10))
 
     expect(legacyEvents).toHaveBeenCalledWith(filter)
-    expect(mockLegacySubscribeObject.subscribe).toHaveBeenCalledWith(callback)
+    expect(mockLegacySubscribeObject.subscribe).toHaveBeenCalledWith(
+      expect.any(Function)
+    )
   })
 })

packages/fcl-core/src/events/index.ts

@@ -25,8 +25,11 @@ export function events(filterOrType?: EventFilter | string) {
 
   return {
     subscribe: (
-      callback: (events: Event | null, error: Error | null) => void
-    ) => {
+      onData: (event: Event) => void,
+      onError: (error: Error) => void = (error: Error) => {
+        console.error("Unhandled error in event subscription:", error)
+      }
+    ): (() => void) => {
       let unsubscribeFn = () => {}
       let unsubscribeFnLegacy = () => {}
 
@@ -35,8 +38,9 @@ export function events(filterOrType?: EventFilter | string) {
         const {unsubscribe} = subscribe({
           topic: SubscriptionTopic.EVENTS,
           args: filter,
-          onData: data => {
-            callback(data, null)
+          onData: event => {
+            // Emit the event
+            onData(event)
           },
           onError: (error: Error) => {
             // If subscriptions are not supported, fallback to legacy polling, otherwise return the error
@@ -46,7 +50,7 @@ export function events(filterOrType?: EventFilter | string) {
               )
               fallbackLegacyPolling()
             } else {
-              callback(null, error)
+              onError(error)
             }
           },
         })
@@ -60,7 +64,15 @@ export function events(filterOrType?: EventFilter | string) {
             "Legacy fcl.events fallback only supports string filters (single event type)"
           )
         }
-        unsubscribeFnLegacy = legacyEvents(filterOrType).subscribe(callback)
+        unsubscribeFnLegacy = legacyEvents(filterOrType).subscribe(
+          (event: Event, error?: Error) => {
+            if (error) {
+              onError(error)
+            } else {
+              onData(event)
+            }
+          }
+        )
       }
 
       async function subscribeToEvents() {
@@ -80,7 +92,7 @@ export function events(filterOrType?: EventFilter | string) {
 
       // Subscribe to events
       const initPromise = subscribeToEvents().catch(error => {
-        callback(null, error)
+        onError(error)
       })
 
       // Return an unsubscribe function