Function queryStream

repos/effect/packages/sql/mysql2/src/MysqlClient.ts:475–520

(
  conn: Mysql.PoolConnection,
  sql: string,
  params?: ReadonlyArray<any>
)

      return ["", []]
    },
    onRecordUpdate() {
      return ["", []]
    }
  })

const escape = Statement.defaultEscape("`")

function queryStream(
  conn: Mysql.PoolConnection,
  sql: string,
  params?: ReadonlyArray<any>
) {
  return asyncPauseResume<any, SqlError>(Effect.fnUntraced(function*(emit) {
    const query = (conn as any).query(sql, params).stream()
    yield* Effect.addFinalizer(() => Effect.sync(() => query.destroy() as void))

    let buffer: Array<any> = []
    let taskPending = false
    query.on(
      "error",
      (cause: unknown) =>
        emit.fail(new SqlError({ reason: classifyError(cause, "Failed to stream statement", "stream") }))
    )
    query.on("data", (row: any) => {
      buffer.push(row)
      if (!taskPending) {
        taskPending = true
        queueMicrotask(() => {
          const items = buffer
          buffer = []
          emit.array(items)
          taskPending = false
        })
      }
    })
    query.on("end", () => {
      if (buffer.length > 0) {
        emit.array(buffer)
        buffer = []
      }
      emit.end()
    })

    return {
      onPause() {
        query.pause()
      },
      onResume() {
        query.resume()