Function queryStream

packages/sql-mysql2/src/MysqlClient.ts:328–357

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

const escape = Statement.defaultEscape("`")

function queryStream(
  conn: Mysql.PoolConnection,
  sql: string,
  params?: ReadonlyArray<any>
) {
  return asyncPauseResume<any, SqlError>((emit) => {
    const query = (conn as any).query(sql, params).stream()
    let buffer: Array<any> = []
    let taskPending = false
    query.on("error", (cause: unknown) => emit.fail(new SqlError({ cause, message: "Failed to stream statement" })))
    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", () => emit.end())
    return {
      onInterrupt: Effect.sync(() => query.destroy()),
      onPause: Effect.sync(() => query.pause()),
      onResume: Effect.sync(() => query.resume())
    }
  })
}