Class ConnectionImpl

packages/sql-pg/src/PgClient.ts:123–273

      undefined

    class ConnectionImpl implements Connection {
      readonly pg: Pg.PoolClient | undefined
      constructor(pg?: Pg.PoolClient) {
        this.pg = pg
      }

      private runWithClient<A>(f: (client: Pg.PoolClient, resume: (_: Effect.Effect<A, SqlError>) => void) => void) {
        if (this.pg !== undefined) {
          return Effect.async<A, SqlError>((resume) => {
            f(this.pg!, resume)
            return makeCancel(pool, this.pg!)
          })
        }
        return Effect.async<A, SqlError>((resume) => {
          let done = false
          let cancel: Effect.Effect<void> | undefined = undefined
          let client: Pg.PoolClient | undefined = undefined
          function onError(cause: Error) {
            cleanup(cause)
            resume(Effect.fail(new SqlError({ cause, message: "Connection error" })))
          }
          function cleanup(cause?: Error) {
            if (!done) client?.release(cause)
            done = true
            client?.off("error", onError)
          }
          pool.connect((cause, client_) => {
            if (cause) {
              return resume(Effect.fail(new SqlError({ cause, message: "Failed to acquire connection" })))
            } else if (!client_) {
              return resume(
                Effect.fail(
                  new SqlError({ message: "Failed to acquire connection", cause: new Error("No client returned") })
                )
              )
            } else if (done) {
              client_.release()
              return
            }
            client = client_
            client.once("error", onError)
            cancel = makeCancel(pool, client)
            f(client, (eff) => {
              cleanup()
              resume(eff)
            })
          })
          return Effect.suspend(() => {
            if (!cancel) {
              cleanup()
              return Effect.void
            }
            return Effect.ensuring(cancel, Effect.sync(cleanup))
          })
        })
      }

      private run(query: string, params: ReadonlyArray<unknown>) {