Function makeParser

repos/effect/packages/effect/src/unstable/encoding/Sse.ts:178–328

(onParse: (event: AnyEvent) => void)

  S extends EventCodec,
  IE,
  Done
>(
  schema: S,
  options?: DecodeOptions
): Channel.Channel<
  NonEmptyReadonlyArray<S["Type"]>,
  IE | Retry | SseError | Schema.SchemaError,
  Done,
  NonEmptyReadonlyArray<string>,
  IE,
  Done,
  S["DecodingServices"]
> =>
  Channel.pipeTo(
    decode<IE, Done>(options),
    ChannelSchema.decode(EventEncoded.pipe(
      Schema.decodeTo(schema)
    ))()
  )

/**
 * Creates an SSE decoder channel that JSON-decodes each event `data` field with a schema.
 *
 * **Details**
 *
 * The output preserves the SSE `event` name and optional `id` while replacing
 * `data` with the decoded value.
 *
 * @category decoding
 * @since 4.0.0
 */
export const decodeDataSchema = <Type, DecodingServices, IE, Done>(
  schema: Schema.ConstraintDecoder<Type, DecodingServices>,
  options?: DecodeOptions
): Channel.Channel<
  NonEmptyReadonlyArray<{
    readonly event: string
    readonly id: string | undefined
    readonly data: Type
  }>,
  IE | Retry | SseError | Schema.SchemaError,
  Done,
  NonEmptyReadonlyArray<string>,
  IE,
  Done,
  DecodingServices
> => {
  const eventSchema = Schema.Struct({
    ...EventEncoded.fields,
    data: Schema.fromJsonString(schema)
  })
  return Channel.pipeTo(
    decode<IE, Done>(options),
    Channel.map(
      ChannelSchema.decode(eventSchema)(),
      Arr.map((event) => ({ ...event, id: event.id }))
    )
  )