Method executeStream

packages/sql/pg/src/PgClient.ts:740–764

(
    sql: string,
    params: ReadonlyArray<unknown>,
    transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
  )

    return this.execute(sql, params, transformRows)
  }
  executeStream(
    sql: string,
    params: ReadonlyArray<unknown>,
    transformRows: (<A extends object>(row: ReadonlyArray<A>) => ReadonlyArray<A>) | undefined
  ) {
    // oxlint-disable-next-line @typescript-eslint/no-this-alias
    const self = this
    return Stream.fromChannel(Channel.fromTransform(Effect.fnUntraced(function*(_, scope) {
      const client = yield* Scope.provide(self.reserve, scope)
      yield* Scope.addFinalizer(scope, Effect.promise(() => cursor.close()))
      const cursor = client.query(new Cursor(sql, params as any))
      // @effect-diagnostics-next-line returnEffectInGen:off
      return Effect.callback<Arr.NonEmptyReadonlyArray<any>, SqlError | Cause.Done>((resume) => {
        cursor.read(128, (err, rows) => {
          if (err) {
            resume(Effect.fail(new SqlError({ reason: classifyError(err, "Failed to execute statement", "stream") })))
          } else if (Arr.isArrayNonEmpty(rows)) {
            resume(Effect.succeed(transformRows ? transformRows(rows) as any : rows))
          } else {
            resume(Cause.done())
          }
        })
      })
    })))
  }
}

const cancelEffects = new WeakMap<Pg.PoolClient, Effect.Effect<void> | undefined>()