| 750 | const cursor = client.query(new Cursor(sql, params as any)) |
| 751 | // @effect-diagnostics-next-line returnEffectInGen:off |
| 752 | return Effect.callback<Arr.NonEmptyReadonlyArray<any>, SqlError | Cause.Done>((resume) => { |
| 753 | cursor.read(128, (err, rows) => { |
| 754 | if (err) { |
| 755 | resume(Effect.fail(new SqlError({ reason: classifyError(err, "Failed to execute statement", "stream") }))) |
| 756 | } else if (Arr.isArrayNonEmpty(rows)) { |
| 757 | resume(Effect.succeed(transformRows ? transformRows(rows) as any : rows)) |
| 758 | } else { |
| 759 | resume(Cause.done()) |
| 760 | } |
| 761 | }) |
| 762 | }) |
| 763 | }))) |
| 764 | } |
| 765 | } |
| 766 | |
| 767 | const cancelEffects = new WeakMap<Pg.PoolClient, Effect.Effect<void> | undefined>() |
| 768 | const makeCancel = (pool: Pg.Pool, client: Pg.PoolClient) => { |
| 769 | if (cancelEffects.has(client)) { |
| 770 | return cancelEffects.get(client)! |
| 771 | } |
| 772 | const processId = (client as any).processID |
| 773 | const eff = processId !== undefined |
| 774 | // query cancelation is best-effort, so we don't fail if it doesn't work |