Function decode

packages/effect/src/unstable/encoding/Sse.ts:102–143

(options?: DecodeOptions)

 * @since 4.0.0
 */
export const decode = <IE, Done>(options?: DecodeOptions): Channel.Channel<
  NonEmptyReadonlyArray<Event>,
  IE | Retry | SseError,
  Done,
  NonEmptyReadonlyArray<string>,
  IE,
  Done
> =>
  Channel.fromTransform((upstream, _scope) =>
    Effect.sync(() => {
      let buffer: Array<Event> = []
      let retry: Retry | undefined
      const parser = makeParser((event) => {
        if (event._tag === "Retry") {
          retry = event
        } else {
          buffer.push(event)
        }
      }, options)

      const pump = Effect.flatMap(upstream, (arr) => {
        for (let i = 0; i < arr.length; i++) {
          const error = parser.feed(arr[i])
          if (error !== undefined) {
            return Effect.fail(error)
          }
        }
        return Effect.void
      })

      return Effect.suspend(function loop(): Pull.Pull<NonEmptyReadonlyArray<Event>, IE | Retry | SseError, Done> {
        if (Arr.isArrayNonEmpty(buffer)) {
          const out = buffer
          buffer = []
          return Effect.succeed(out)
        } else if (retry) {
          return Effect.fail(retry)
        }
        return Effect.flatMap(pump, loop)
      })
    })
  )

/**
 * A constraint for schemas that can decode SSE events.