feat(client, server)!: update event-source errors and decode behavior (#182)

* fix(client, server)!: throw an error if event source ends incompletely

* fix tests

* feat(server): not serialize ErrorEvent

* ignore UnknownEvent

* improve event-source compatibility

Author: dinwwwh

packages/client/src/index.ts

@@ -7,3 +7,5 @@ export * from './event-iterator'
 export * from './event-iterator-state'
 export * from './types'
 export * from './utils'
+
+export { ErrorEvent } from '@orpc/standard-server'

packages/client/src/openapi/serializer.test.ts

@@ -160,54 +160,6 @@ describe('openAPISerializer', () => {
         expect(serialize).toHaveBeenCalledWith(expect.objectContaining({ code: 'BAD_GATEWAY', data: { order: 3 } }))
         serialize.mockClear()
       })
-
-      it('on error with ErrorEvent', async () => {
-        const date = new Date()
-        const error = withEventMeta(new ErrorEvent({ data: { order: 3 } }), { id: '123456' })
-
-        const serialized = openapiSerializer.serialize((async function* () {
-          yield 1
-          yield withEventMeta({ order: 2, date }, { retry: 1000 })
-          throw error
-        })()) as any
-
-        await expect(serialized.next()).resolves.toSatisfy(({ value, done }) => {
-          expect(done).toBe(false)
-          expect(value).toBe(serialize.mock.results[0]!.value[0])
-          expect(getEventMeta(value)).toEqual(undefined)
-
-          return true
-        })
-
-        expect(serialize).toHaveBeenCalledOnce()
-        expect(serialize).toHaveBeenCalledWith(1)
-        serialize.mockClear()
-
-        await expect(serialized.next()).resolves.toSatisfy(({ value, done }) => {
-          expect(done).toBe(false)
-          expect(value).toEqual(serialize.mock.results[0]!.value[0])
-          expect(getEventMeta(value)).toEqual({ retry: 1000 })
-
-          return true
-        })
-
-        expect(serialize).toHaveBeenCalledOnce()
-        expect(serialize).toHaveBeenCalledWith({ order: 2, date })
-        serialize.mockClear()
-
-        await expect(serialized.next()).rejects.toSatisfy((e: any) => {
-          expect(e).toBeInstanceOf(ErrorEvent)
-          expect(e.data).toEqual(serialize.mock.results[0]!.value[0])
-          expect(e.cause).toBe(error)
-          expect(getEventMeta(e)).toEqual({ id: '123456' })
-
-          return true
-        })
-
-        expect(serialize).toHaveBeenCalledOnce()
-        expect(serialize).toHaveBeenCalledWith({ order: 3 })
-        serialize.mockClear()
-      })
     })
   })
 

packages/client/src/openapi/serializer.ts

@@ -17,13 +17,6 @@ export class OpenAPISerializer {
       return mapEventIterator(data, {
         value: async value => this.#serialize(value, false),
         error: async (e) => {
-          if (e instanceof ErrorEvent) {
-            return new ErrorEvent({
-              data: this.#serialize(e.data, false),
-              cause: e,
-            })
-          }
-
           return new ErrorEvent({
             data: this.#serialize(toORPCError(e).toJSON(), false),
             cause: e,

packages/client/src/rpc/serializer.test.ts

@@ -142,43 +142,6 @@ describe('rpcSerializer: event-source iterator', async () => {
     })
   })
 
-  it('on error with ErrorEvent', async () => {
-    const error = withEventMeta(new ErrorEvent({ data: { order: 3 } }), { id: '123456' })
-
-    const iterator = (async function* () {
-      yield 1
-      yield withEventMeta({ order: 2, date }, { retry: 1000 })
-      throw error
-    })()
-
-    const deserialized = serializeAndDeserialize(iterator) as any
-
-    expect(deserialized).toSatisfy(isAsyncIteratorObject)
-    await expect(deserialized.next()).resolves.toSatisfy(({ value, done }) => {
-      expect(done).toBe(false)
-      expect(value).toEqual(1)
-      expect(getEventMeta(value)).toEqual(undefined)
-
-      return true
-    })
-
-    await expect(deserialized.next()).resolves.toSatisfy(({ value, done }) => {
-      expect(done).toBe(false)
-      expect(value).toEqual({ order: 2, date })
-      expect(getEventMeta(value)).toEqual({ retry: 1000 })
-
-      return true
-    })
-
-    await expect(deserialized.next()).rejects.toSatisfy((e: any) => {
-      expect(e).toEqual(error)
-      expect(e).toBeInstanceOf(ErrorEvent)
-      expect(e.cause).toBeInstanceOf(ErrorEvent)
-
-      return true
-    })
-  })
-
   it('on error with unknown error when deserialize', async () => {
     const error = withEventMeta(new Error('UNKNOWN'), { id: '123456' })
 
@@ -213,4 +176,19 @@ describe('rpcSerializer: event-source iterator', async () => {
       return true
     })
   })
+
+  it('deserialize an invalid-ORPCError', async () => {
+    const iterator = serializer.deserialize((async function* () {
+      throw new ErrorEvent({
+        data: { json: { value: 1234 } },
+      })
+    })()) as any
+
+    await expect(iterator.next()).rejects.toSatisfy((e: any) => {
+      expect(e).toBeInstanceOf(ErrorEvent)
+      expect(e.data).toEqual({ value: 1234 })
+
+      return true
+    })
+  })
 })

packages/client/src/rpc/serializer.ts

@@ -14,13 +14,6 @@ export class RPCSerializer {
       return mapEventIterator(data, {
         value: async (value: unknown) => this.#serialize(value, false),
         error: async (e) => {
-          if (e instanceof ErrorEvent) {
-            return new ErrorEvent({
-              data: this.#serialize(e.data, false),
-              cause: e,
-            })
-          }
-
           return new ErrorEvent({
             data: this.#serialize(toORPCError(e).toJSON(), false),
             cause: e,

packages/standard-server-fetch/src/event-source.test.ts

@@ -1,5 +1,5 @@
 import { isAsyncIteratorObject } from '@orpc/shared'
-import { ErrorEvent, getEventMeta, UnknownEvent, withEventMeta } from '@orpc/standard-server'
+import { ErrorEvent, getEventMeta, withEventMeta } from '@orpc/standard-server'
 import { toEventIterator, toEventStream } from './event-source'
 
 describe('toEventIterator', () => {
@@ -8,7 +8,7 @@ describe('toEventIterator', () => {
       async pull(controller) {
         controller.enqueue('event: message\ndata: {"order": 1}\nid: id-1\nretry: 10000\n\n')
         controller.enqueue('event: message\ndata: {"order": 2}\nid: id-2\n\n')
-        controller.enqueue('event: done\ndata: {"order": 3}\nid: id-3\nretry: 30000')
+        controller.enqueue('event: done\ndata: {"order": 3}\nid: id-3\nretry: 30000\n\n')
         controller.close()
       },
     }).pipeThrough(new TextEncoderStream())
@@ -39,8 +39,6 @@ describe('toEventIterator', () => {
 
       return true
     })
-
-    await expect(stream.getReader().closed).resolves.toBe(undefined)
   })
 
   it('without dont event', async () => {
@@ -87,7 +85,7 @@ describe('toEventIterator', () => {
       async pull(controller) {
         controller.enqueue('event: message\ndata: {"order": 1}\nid: id-1\nretry: 10000\n\n')
         controller.enqueue('event: message\ndata: {"order": 2}\nid: id-2\n\n')
-        controller.enqueue('event: error\ndata: {"order": 3}\nid: id-3\nretry: 30000')
+        controller.enqueue('event: error\ndata: {"order": 3}\nid: id-3\nretry: 30000\n\n')
         controller.close()
       },
     }).pipeThrough(new TextEncoderStream())
@@ -122,46 +120,6 @@ describe('toEventIterator', () => {
     await expect(stream.getReader().closed).resolves.toBe(undefined)
   })
 
-  it('with unknown event', async () => {
-    const stream = new ReadableStream<string>({
-      async pull(controller) {
-        controller.enqueue('event: message\ndata: {"order": 1}\nid: id-1\nretry: 10000\n\n')
-        controller.enqueue('event: message\ndata: {"order": 2}\nid: id-2\n\n')
-        controller.enqueue('event: unknown\ndata: {"order": 3}\nid: id-3\nretry: 30000')
-        controller.close()
-      },
-    }).pipeThrough(new TextEncoderStream())
-
-    const generator = toEventIterator(stream)
-    expect(generator).toSatisfy(isAsyncIteratorObject)
-
-    expect(await generator.next()).toSatisfy(({ done, value }) => {
-      expect(done).toEqual(false)
-      expect(value).toEqual({ order: 1 })
-      expect(getEventMeta(value)).toEqual(expect.objectContaining({ id: 'id-1', retry: 10000 }))
-
-      return true
-    })
-
-    expect(await generator.next()).toSatisfy(({ done, value }) => {
-      expect(done).toEqual(false)
-      expect(value).toEqual({ order: 2 })
-      expect(getEventMeta(value)).toEqual(expect.objectContaining({ id: 'id-2', retry: undefined }))
-
-      return true
-    })
-
-    await expect(generator.next()).rejects.toSatisfy((error: any) => {
-      expect(error).toBeInstanceOf(UnknownEvent)
-      expect(error.data).toEqual({ order: 3 })
-      expect(getEventMeta(error)).toEqual(expect.objectContaining({ id: 'id-3', retry: 30000 }))
-
-      return true
-    })
-
-    await expect(stream.getReader().closed).resolves.toBe(undefined)
-  })
-
   it('when .return() before finish reading', async () => {
     const stream = new ReadableStream<string>({
       async pull(controller) {
@@ -205,7 +163,7 @@ describe('toEventStream', () => {
 
     expect((await reader.read()).value).toEqual('event: message\nid: id-1\ndata: {"order":1}\n\n')
     expect((await reader.read()).value).toEqual('event: message\nretry: 20000\ndata: {"order":2}\n\n')
-    expect((await reader.read()).value).toEqual('event: message\ndata: \n\n')
+    expect((await reader.read()).value).toEqual('event: message\n\n')
     expect((await reader.read()).value).toEqual('event: done\nretry: 40000\nid: id-4\ndata: {"order":4}\n\n')
     expect((await reader.read()).done).toEqual(true)
   })
@@ -224,8 +182,8 @@ describe('toEventStream', () => {
 
     expect((await reader.read()).value).toEqual('event: message\nid: id-1\ndata: {"order":1}\n\n')
     expect((await reader.read()).value).toEqual('event: message\nretry: 20000\ndata: {"order":2}\n\n')
-    expect((await reader.read()).value).toEqual('event: message\ndata: \n\n')
-    expect((await reader.read()).value).toEqual('event: error\nretry: 40000\nid: id-4\ndata: \n\n')
+    expect((await reader.read()).value).toEqual('event: message\n\n')
+    expect((await reader.read()).value).toEqual('event: error\nretry: 40000\nid: id-4\n\n')
     expect((await reader.read()).done).toEqual(true)
   })
 
@@ -243,7 +201,7 @@ describe('toEventStream', () => {
 
     expect((await reader.read()).value).toEqual('event: message\nid: id-1\ndata: {"order":1}\n\n')
     expect((await reader.read()).value).toEqual('event: message\nretry: 20000\ndata: {"order":2}\n\n')
-    expect((await reader.read()).value).toEqual('event: message\ndata: \n\n')
+    expect((await reader.read()).value).toEqual('event: message\n\n')
     expect((await reader.read()).value).toEqual('event: error\nretry: 40000\nid: id-4\ndata: {"order":4}\n\n')
     expect((await reader.read()).done).toEqual(true)
   })

packages/standard-server-fetch/src/event-source.ts

@@ -4,7 +4,6 @@ import {
   ErrorEvent,
   EventDecoderStream,
   getEventMeta,
-  UnknownEvent,
   withEventMeta,
 } from '@orpc/standard-server'
 
@@ -57,17 +56,6 @@ export function toEventIterator(
 
             return done
           }
-
-          default: {
-            let error = new UnknownEvent({
-              message: `Unknown event: ${value.event}`,
-              data: parseEmptyableJSON(value.data),
-            })
-
-            error = withEventMeta(error, value)
-
-            throw error
-          }
         }
       }
     }

packages/standard-server-node/src/event-source.test.ts

@@ -1,6 +1,6 @@
 import { Readable } from 'node:stream'
 import { isAsyncIteratorObject } from '@orpc/shared'
-import { ErrorEvent, getEventMeta, UnknownEvent, withEventMeta } from '@orpc/standard-server'
+import { ErrorEvent, getEventMeta, withEventMeta } from '@orpc/standard-server'
 import { toEventIterator, toEventStream } from './event-source'
 
 describe('toEventIterator', () => {
@@ -9,7 +9,7 @@ describe('toEventIterator', () => {
       async pull(controller) {
         controller.enqueue('event: message\ndata: {"order": 1}\nid: id-1\nretry: 10000\n\n')
         controller.enqueue('event: message\ndata: {"order": 2}\nid: id-2\n\n')
-        controller.enqueue('event: done\ndata: {"order": 3}\nid: id-3\nretry: 30000')
+        controller.enqueue('event: done\ndata: {"order": 3}\nid: id-3\nretry: 30000\n\n')
         controller.close()
       },
     }).pipeThrough(new TextEncoderStream()))
@@ -40,8 +40,6 @@ describe('toEventIterator', () => {
 
       return true
     })
-
-    await expect(Readable.toWeb(stream).getReader().closed).resolves.toBe(undefined)
   })
 
   it('without dont event', async () => {
@@ -88,7 +86,7 @@ describe('toEventIterator', () => {
       async pull(controller) {
         controller.enqueue('event: message\ndata: {"order": 1}\nid: id-1\nretry: 10000\n\n')
         controller.enqueue('event: message\ndata: {"order": 2}\nid: id-2\n\n')
-        controller.enqueue('event: error\ndata: {"order": 3}\nid: id-3\nretry: 30000')
+        controller.enqueue('event: error\ndata: {"order": 3}\nid: id-3\nretry: 30000\n\n')
         controller.close()
       },
     }).pipeThrough(new TextEncoderStream()))
@@ -123,46 +121,6 @@ describe('toEventIterator', () => {
     await expect(Readable.toWeb(stream).getReader().closed).resolves.toBe(undefined)
   })
 
-  it('with unknown event', async () => {
-    const stream = Readable.fromWeb(new ReadableStream<string>({
-      async pull(controller) {
-        controller.enqueue('event: message\ndata: {"order": 1}\nid: id-1\nretry: 10000\n\n')
-        controller.enqueue('event: message\ndata: {"order": 2}\nid: id-2\n\n')
-        controller.enqueue('event: unknown\ndata: {"order": 3}\nid: id-3\nretry: 30000')
-        controller.close()
-      },
-    }).pipeThrough(new TextEncoderStream()))
-
-    const generator = toEventIterator(stream)
-    expect(generator).toSatisfy(isAsyncIteratorObject)
-
-    expect(await generator.next()).toSatisfy(({ done, value }) => {
-      expect(done).toEqual(false)
-      expect(value).toEqual({ order: 1 })
-      expect(getEventMeta(value)).toEqual(expect.objectContaining({ id: 'id-1', retry: 10000 }))
-
-      return true
-    })
-
-    expect(await generator.next()).toSatisfy(({ done, value }) => {
-      expect(done).toEqual(false)
-      expect(value).toEqual({ order: 2 })
-      expect(getEventMeta(value)).toEqual(expect.objectContaining({ id: 'id-2', retry: undefined }))
-
-      return true
-    })
-
-    await expect(generator.next()).rejects.toSatisfy((error: any) => {
-      expect(error).toBeInstanceOf(UnknownEvent)
-      expect(error.data).toEqual({ order: 3 })
-      expect(getEventMeta(error)).toEqual(expect.objectContaining({ id: 'id-3', retry: 30000 }))
-
-      return true
-    })
-
-    await expect(Readable.toWeb(stream).getReader().closed).resolves.toBe(undefined)
-  })
-
   it('when .return() before finish reading', async () => {
     const stream = Readable.fromWeb(new ReadableStream<string>({
       async pull(controller) {
@@ -205,7 +163,7 @@ describe('toEventStream', () => {
 
     expect((await reader.read()).value).toEqual('event: message\nid: id-1\ndata: {"order":1}\n\n')
     expect((await reader.read()).value).toEqual('event: message\nretry: 20000\ndata: {"order":2}\n\n')
-    expect((await reader.read()).value).toEqual('event: message\ndata: \n\n')
+    expect((await reader.read()).value).toEqual('event: message\n\n')
     expect((await reader.read()).value).toEqual('event: done\nretry: 40000\nid: id-4\ndata: {"order":4}\n\n')
     expect((await reader.read()).done).toEqual(true)
   })
@@ -224,8 +182,8 @@ describe('toEventStream', () => {
 
     expect((await reader.read()).value).toEqual('event: message\nid: id-1\ndata: {"order":1}\n\n')
     expect((await reader.read()).value).toEqual('event: message\nretry: 20000\ndata: {"order":2}\n\n')
-    expect((await reader.read()).value).toEqual('event: message\ndata: \n\n')
-    expect((await reader.read()).value).toEqual('event: error\nretry: 40000\nid: id-4\ndata: \n\n')
+    expect((await reader.read()).value).toEqual('event: message\n\n')
+    expect((await reader.read()).value).toEqual('event: error\nretry: 40000\nid: id-4\n\n')
     expect((await reader.read()).done).toEqual(true)
   })
 
@@ -243,7 +201,7 @@ describe('toEventStream', () => {
 
     expect((await reader.read()).value).toEqual('event: message\nid: id-1\ndata: {"order":1}\n\n')
     expect((await reader.read()).value).toEqual('event: message\nretry: 20000\ndata: {"order":2}\n\n')
-    expect((await reader.read()).value).toEqual('event: message\ndata: \n\n')
+    expect((await reader.read()).value).toEqual('event: message\n\n')
     expect((await reader.read()).value).toEqual('event: error\nretry: 40000\nid: id-4\ndata: {"order":4}\n\n')
     expect((await reader.read()).done).toEqual(true)
   })