stream.ts 2.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. import {createParser} from './parse.ts'
  2. import type {EventSourceMessage, EventSourceParser} from './types.ts'
  3. /**
  4. * Options for the EventSourceParserStream.
  5. *
  6. * @public
  7. */
  8. export interface StreamOptions {
  9. /**
  10. * Behavior when a parsing error occurs.
  11. *
  12. * - A custom function can be provided to handle the error.
  13. * - `'terminate'` will error the stream and stop parsing.
  14. * - Any other value will ignore the error and continue parsing.
  15. *
  16. * @defaultValue `undefined`
  17. */
  18. onError?: ('terminate' | ((error: Error) => void)) | undefined
  19. /**
  20. * Callback for when a reconnection interval is sent from the server.
  21. *
  22. * @param retry - The number of milliseconds to wait before reconnecting.
  23. */
  24. onRetry?: ((retry: number) => void) | undefined
  25. /**
  26. * Callback for when a comment is encountered in the stream.
  27. *
  28. * @param comment - The comment encountered in the stream.
  29. */
  30. onComment?: ((comment: string) => void) | undefined
  31. }
  32. /**
  33. * A TransformStream that ingests a stream of strings and produces a stream of `EventSourceMessage`.
  34. *
  35. * @example Basic usage
  36. * ```
  37. * const eventStream =
  38. * response.body
  39. * .pipeThrough(new TextDecoderStream())
  40. * .pipeThrough(new EventSourceParserStream())
  41. * ```
  42. *
  43. * @example Terminate stream on parsing errors
  44. * ```
  45. * const eventStream =
  46. * response.body
  47. * .pipeThrough(new TextDecoderStream())
  48. * .pipeThrough(new EventSourceParserStream({terminateOnError: true}))
  49. * ```
  50. *
  51. * @public
  52. */
  53. export class EventSourceParserStream extends TransformStream<string, EventSourceMessage> {
  54. constructor({onError, onRetry, onComment}: StreamOptions = {}) {
  55. let parser!: EventSourceParser
  56. super({
  57. start(controller) {
  58. parser = createParser({
  59. onEvent: (event) => {
  60. controller.enqueue(event)
  61. },
  62. onError(error) {
  63. if (onError === 'terminate') {
  64. controller.error(error)
  65. } else if (typeof onError === 'function') {
  66. onError(error)
  67. }
  68. // Ignore by default
  69. },
  70. onRetry,
  71. onComment,
  72. })
  73. },
  74. transform(chunk) {
  75. parser.feed(chunk)
  76. },
  77. })
  78. }
  79. }
  80. export {type ErrorType, ParseError} from './errors.ts'
  81. export type {EventSourceMessage} from './types.ts'