Function makeNdjson

packages/effect/src/unstable/rpc/RpcSerialization.ts:117–164

(options?: StreamOptions)

 * @since 4.0.0
 */
export const makeNdjson = (options?: StreamOptions): RpcSerialization["Service"] => {
  const maxBufferSize = options?.maxBufferSize ?? defaultMaxBufferSize
  return RpcSerialization.of({
    contentType: "application/ndjson",
    includesFraming: true,
    makeUnsafe: () => {
      const decoder = new TextDecoder()
      let buffer = ""
      const failMaxBufferSize = (maxBufferSize: number): never => {
        buffer = ""
        throw new MaxBufferSizeExceeded({ maxBufferSize })
      }
      return ({
        decode: (bytes) => {
          buffer += typeof bytes === "string" ? bytes : decoder.decode(bytes, { stream: true })
          let position = 0
          let nlIndex = buffer.indexOf("\n", position)
          const items: Array<unknown> = []
          while (nlIndex !== -1) {
            if (isBufferSizeExceeded(nlIndex - position, maxBufferSize)) {
              failMaxBufferSize(maxBufferSize)
            }
            const item = JSON.parse(buffer.slice(position, nlIndex))
            items.push(item)
            position = nlIndex + 1
            nlIndex = buffer.indexOf("\n", position)
          }
          buffer = buffer.slice(position)
          if (isBufferSizeExceeded(buffer.length, maxBufferSize)) {
            failMaxBufferSize(maxBufferSize)
          }
          return items
        },
        encode: (response) => {
          if (Array.isArray(response)) {
            if (response.length === 0) return undefined
            let data = ""
            for (let i = 0; i < response.length; i++) {
              data += JSON.stringify(response[i]) + "\n"
            }
            return data
          }
          return JSON.stringify(response) + "\n"
        }
      })
    }
  })
}

/**
 * Default newline-delimited JSON RPC serialization.