diff options
Diffstat (limited to 'vendor/github.com/jackc/pgx/v4/stdlib/sql.go')
| -rw-r--r-- | vendor/github.com/jackc/pgx/v4/stdlib/sql.go | 879 |
1 files changed, 0 insertions, 879 deletions
diff --git a/vendor/github.com/jackc/pgx/v4/stdlib/sql.go b/vendor/github.com/jackc/pgx/v4/stdlib/sql.go deleted file mode 100644 index f43ae3249..000000000 --- a/vendor/github.com/jackc/pgx/v4/stdlib/sql.go +++ /dev/null @@ -1,879 +0,0 @@ -// Package stdlib is the compatibility layer from pgx to database/sql. -// -// A database/sql connection can be established through sql.Open. -// -// db, err := sql.Open("pgx", "postgres://pgx_md5:secret@localhost:5432/pgx_test?sslmode=disable") -// if err != nil { -// return err -// } -// -// Or from a DSN string. -// -// db, err := sql.Open("pgx", "user=postgres password=secret host=localhost port=5432 database=pgx_test sslmode=disable") -// if err != nil { -// return err -// } -// -// Or a pgx.ConnConfig can be used to set configuration not accessible via connection string. In this case the -// pgx.ConnConfig must first be registered with the driver. This registration returns a connection string which is used -// with sql.Open. -// -// connConfig, _ := pgx.ParseConfig(os.Getenv("DATABASE_URL")) -// connConfig.Logger = myLogger -// connStr := stdlib.RegisterConnConfig(connConfig) -// db, _ := sql.Open("pgx", connStr) -// -// pgx uses standard PostgreSQL positional parameters in queries. e.g. $1, $2. -// It does not support named parameters. -// -// db.QueryRow("select * from users where id=$1", userID) -// -// In Go 1.13 and above (*sql.Conn) Raw() can be used to get a *pgx.Conn from the standard -// database/sql.DB connection pool. This allows operations that use pgx specific functionality. -// -// // Given db is a *sql.DB -// conn, err := db.Conn(context.Background()) -// if err != nil { -// // handle error from acquiring connection from DB pool -// } -// -// err = conn.Raw(func(driverConn interface{}) error { -// conn := driverConn.(*stdlib.Conn).Conn() // conn is a *pgx.Conn -// // Do pgx specific stuff with conn -// conn.CopyFrom(...) -// return nil -// }) -// if err != nil { -// // handle error that occurred while using *pgx.Conn -// } -package stdlib - -import ( - "context" - "database/sql" - "database/sql/driver" - "errors" - "fmt" - "io" - "math" - "math/rand" - "reflect" - "strconv" - "strings" - "sync" - "time" - - "github.com/jackc/pgconn" - "github.com/jackc/pgtype" - "github.com/jackc/pgx/v4" -) - -// Only intrinsic types should be binary format with database/sql. -var databaseSQLResultFormats pgx.QueryResultFormatsByOID - -var pgxDriver *Driver - -type ctxKey int - -var ctxKeyFakeTx ctxKey = 0 - -var ErrNotPgx = errors.New("not pgx *sql.DB") - -func init() { - pgxDriver = &Driver{ - configs: make(map[string]*pgx.ConnConfig), - } - fakeTxConns = make(map[*pgx.Conn]*sql.Tx) - - // if pgx driver was already registered by different pgx major version then we - // skip registration under the default name. - if !contains(sql.Drivers(), "pgx") { - sql.Register("pgx", pgxDriver) - } - sql.Register("pgx/v4", pgxDriver) - - databaseSQLResultFormats = pgx.QueryResultFormatsByOID{ - pgtype.BoolOID: 1, - pgtype.ByteaOID: 1, - pgtype.CIDOID: 1, - pgtype.DateOID: 1, - pgtype.Float4OID: 1, - pgtype.Float8OID: 1, - pgtype.Int2OID: 1, - pgtype.Int4OID: 1, - pgtype.Int8OID: 1, - pgtype.OIDOID: 1, - pgtype.TimestampOID: 1, - pgtype.TimestamptzOID: 1, - pgtype.XIDOID: 1, - } -} - -// TODO replace by slices.Contains when experimental package will be merged to stdlib -// https://pkg.go.dev/golang.org/x/exp/slices#Contains -func contains(list []string, y string) bool { - for _, x := range list { - if x == y { - return true - } - } - return false -} - -var ( - fakeTxMutex sync.Mutex - fakeTxConns map[*pgx.Conn]*sql.Tx -) - -// OptionOpenDB options for configuring the driver when opening a new db pool. -type OptionOpenDB func(*connector) - -// OptionBeforeConnect provides a callback for before connect. It is passed a shallow copy of the ConnConfig that will -// be used to connect, so only its immediate members should be modified. -func OptionBeforeConnect(bc func(context.Context, *pgx.ConnConfig) error) OptionOpenDB { - return func(dc *connector) { - dc.BeforeConnect = bc - } -} - -// OptionAfterConnect provides a callback for after connect. -func OptionAfterConnect(ac func(context.Context, *pgx.Conn) error) OptionOpenDB { - return func(dc *connector) { - dc.AfterConnect = ac - } -} - -// OptionResetSession provides a callback that can be used to add custom logic prior to executing a query on the -// connection if the connection has been used before. -// If ResetSessionFunc returns ErrBadConn error the connection will be discarded. -func OptionResetSession(rs func(context.Context, *pgx.Conn) error) OptionOpenDB { - return func(dc *connector) { - dc.ResetSession = rs - } -} - -// RandomizeHostOrderFunc is a BeforeConnect hook that randomizes the host order in the provided connConfig, so that a -// new host becomes primary each time. This is useful to distribute connections for multi-master databases like -// CockroachDB. If you use this you likely should set https://golang.org/pkg/database/sql/#DB.SetConnMaxLifetime as well -// to ensure that connections are periodically rebalanced across your nodes. -func RandomizeHostOrderFunc(ctx context.Context, connConfig *pgx.ConnConfig) error { - if len(connConfig.Fallbacks) == 0 { - return nil - } - - newFallbacks := append([]*pgconn.FallbackConfig{&pgconn.FallbackConfig{ - Host: connConfig.Host, - Port: connConfig.Port, - TLSConfig: connConfig.TLSConfig, - }}, connConfig.Fallbacks...) - - rand.Shuffle(len(newFallbacks), func(i, j int) { - newFallbacks[i], newFallbacks[j] = newFallbacks[j], newFallbacks[i] - }) - - // Use the one that sorted last as the primary and keep the rest as the fallbacks - newPrimary := newFallbacks[len(newFallbacks)-1] - connConfig.Host = newPrimary.Host - connConfig.Port = newPrimary.Port - connConfig.TLSConfig = newPrimary.TLSConfig - connConfig.Fallbacks = newFallbacks[:len(newFallbacks)-1] - return nil -} - -func GetConnector(config pgx.ConnConfig, opts ...OptionOpenDB) driver.Connector { - c := connector{ - ConnConfig: config, - BeforeConnect: func(context.Context, *pgx.ConnConfig) error { return nil }, // noop before connect by default - AfterConnect: func(context.Context, *pgx.Conn) error { return nil }, // noop after connect by default - ResetSession: func(context.Context, *pgx.Conn) error { return nil }, // noop reset session by default - driver: pgxDriver, - } - - for _, opt := range opts { - opt(&c) - } - return c -} - -func OpenDB(config pgx.ConnConfig, opts ...OptionOpenDB) *sql.DB { - c := GetConnector(config, opts...) - return sql.OpenDB(c) -} - -type connector struct { - pgx.ConnConfig - BeforeConnect func(context.Context, *pgx.ConnConfig) error // function to call before creation of every new connection - AfterConnect func(context.Context, *pgx.Conn) error // function to call after creation of every new connection - ResetSession func(context.Context, *pgx.Conn) error // function is called before a connection is reused - driver *Driver -} - -// Connect implement driver.Connector interface -func (c connector) Connect(ctx context.Context) (driver.Conn, error) { - var ( - err error - conn *pgx.Conn - ) - - // Create a shallow copy of the config, so that BeforeConnect can safely modify it - connConfig := c.ConnConfig - if err = c.BeforeConnect(ctx, &connConfig); err != nil { - return nil, err - } - - if conn, err = pgx.ConnectConfig(ctx, &connConfig); err != nil { - return nil, err - } - - if err = c.AfterConnect(ctx, conn); err != nil { - return nil, err - } - - return &Conn{conn: conn, driver: c.driver, connConfig: connConfig, resetSessionFunc: c.ResetSession}, nil -} - -// Driver implement driver.Connector interface -func (c connector) Driver() driver.Driver { - return c.driver -} - -// GetDefaultDriver returns the driver initialized in the init function -// and used when the pgx driver is registered. -func GetDefaultDriver() driver.Driver { - return pgxDriver -} - -type Driver struct { - configMutex sync.Mutex - configs map[string]*pgx.ConnConfig - sequence int -} - -func (d *Driver) Open(name string) (driver.Conn, error) { - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) // Ensure eventual timeout - defer cancel() - - connector, err := d.OpenConnector(name) - if err != nil { - return nil, err - } - return connector.Connect(ctx) -} - -func (d *Driver) OpenConnector(name string) (driver.Connector, error) { - return &driverConnector{driver: d, name: name}, nil -} - -func (d *Driver) registerConnConfig(c *pgx.ConnConfig) string { - d.configMutex.Lock() - connStr := fmt.Sprintf("registeredConnConfig%d", d.sequence) - d.sequence++ - d.configs[connStr] = c - d.configMutex.Unlock() - return connStr -} - -func (d *Driver) unregisterConnConfig(connStr string) { - d.configMutex.Lock() - delete(d.configs, connStr) - d.configMutex.Unlock() -} - -type driverConnector struct { - driver *Driver - name string -} - -func (dc *driverConnector) Connect(ctx context.Context) (driver.Conn, error) { - var connConfig *pgx.ConnConfig - - dc.driver.configMutex.Lock() - connConfig = dc.driver.configs[dc.name] - dc.driver.configMutex.Unlock() - - if connConfig == nil { - var err error - connConfig, err = pgx.ParseConfig(dc.name) - if err != nil { - return nil, err - } - } - - conn, err := pgx.ConnectConfig(ctx, connConfig) - if err != nil { - return nil, err - } - - c := &Conn{ - conn: conn, - driver: dc.driver, - connConfig: *connConfig, - resetSessionFunc: func(context.Context, *pgx.Conn) error { return nil }, - } - - return c, nil -} - -func (dc *driverConnector) Driver() driver.Driver { - return dc.driver -} - -// RegisterConnConfig registers a ConnConfig and returns the connection string to use with Open. -func RegisterConnConfig(c *pgx.ConnConfig) string { - return pgxDriver.registerConnConfig(c) -} - -// UnregisterConnConfig removes the ConnConfig registration for connStr. -func UnregisterConnConfig(connStr string) { - pgxDriver.unregisterConnConfig(connStr) -} - -type Conn struct { - conn *pgx.Conn - psCount int64 // Counter used for creating unique prepared statement names - driver *Driver - connConfig pgx.ConnConfig - resetSessionFunc func(context.Context, *pgx.Conn) error // Function is called before a connection is reused -} - -// Conn returns the underlying *pgx.Conn -func (c *Conn) Conn() *pgx.Conn { - return c.conn -} - -func (c *Conn) Prepare(query string) (driver.Stmt, error) { - return c.PrepareContext(context.Background(), query) -} - -func (c *Conn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) { - if c.conn.IsClosed() { - return nil, driver.ErrBadConn - } - - name := fmt.Sprintf("pgx_%d", c.psCount) - c.psCount++ - - sd, err := c.conn.Prepare(ctx, name, query) - if err != nil { - return nil, err - } - - return &Stmt{sd: sd, conn: c}, nil -} - -func (c *Conn) Close() error { - ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) - defer cancel() - return c.conn.Close(ctx) -} - -func (c *Conn) Begin() (driver.Tx, error) { - return c.BeginTx(context.Background(), driver.TxOptions{}) -} - -func (c *Conn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) { - if c.conn.IsClosed() { - return nil, driver.ErrBadConn - } - - if pconn, ok := ctx.Value(ctxKeyFakeTx).(**pgx.Conn); ok { - *pconn = c.conn - return fakeTx{}, nil - } - - var pgxOpts pgx.TxOptions - switch sql.IsolationLevel(opts.Isolation) { - case sql.LevelDefault: - case sql.LevelReadUncommitted: - pgxOpts.IsoLevel = pgx.ReadUncommitted - case sql.LevelReadCommitted: - pgxOpts.IsoLevel = pgx.ReadCommitted - case sql.LevelRepeatableRead, sql.LevelSnapshot: - pgxOpts.IsoLevel = pgx.RepeatableRead - case sql.LevelSerializable: - pgxOpts.IsoLevel = pgx.Serializable - default: - return nil, fmt.Errorf("unsupported isolation: %v", opts.Isolation) - } - - if opts.ReadOnly { - pgxOpts.AccessMode = pgx.ReadOnly - } - - tx, err := c.conn.BeginTx(ctx, pgxOpts) - if err != nil { - return nil, err - } - - return wrapTx{ctx: ctx, tx: tx}, nil -} - -func (c *Conn) ExecContext(ctx context.Context, query string, argsV []driver.NamedValue) (driver.Result, error) { - if c.conn.IsClosed() { - return nil, driver.ErrBadConn - } - - args := namedValueToInterface(argsV) - - commandTag, err := c.conn.Exec(ctx, query, args...) - // if we got a network error before we had a chance to send the query, retry - if err != nil { - if pgconn.SafeToRetry(err) { - return nil, driver.ErrBadConn - } - } - return driver.RowsAffected(commandTag.RowsAffected()), err -} - -func (c *Conn) QueryContext(ctx context.Context, query string, argsV []driver.NamedValue) (driver.Rows, error) { - if c.conn.IsClosed() { - return nil, driver.ErrBadConn - } - - args := []interface{}{databaseSQLResultFormats} - args = append(args, namedValueToInterface(argsV)...) - - rows, err := c.conn.Query(ctx, query, args...) - if err != nil { - if pgconn.SafeToRetry(err) { - return nil, driver.ErrBadConn - } - return nil, err - } - - // Preload first row because otherwise we won't know what columns are available when database/sql asks. - more := rows.Next() - if err = rows.Err(); err != nil { - rows.Close() - return nil, err - } - return &Rows{conn: c, rows: rows, skipNext: true, skipNextMore: more}, nil -} - -func (c *Conn) Ping(ctx context.Context) error { - if c.conn.IsClosed() { - return driver.ErrBadConn - } - - err := c.conn.Ping(ctx) - if err != nil { - // A Ping failure implies some sort of fatal state. The connection is almost certainly already closed by the - // failure, but manually close it just to be sure. - c.Close() - return driver.ErrBadConn - } - - return nil -} - -func (c *Conn) CheckNamedValue(*driver.NamedValue) error { - // Underlying pgx supports sql.Scanner and driver.Valuer interfaces natively. So everything can be passed through directly. - return nil -} - -func (c *Conn) ResetSession(ctx context.Context) error { - if c.conn.IsClosed() { - return driver.ErrBadConn - } - - return c.resetSessionFunc(ctx, c.conn) -} - -type Stmt struct { - sd *pgconn.StatementDescription - conn *Conn -} - -func (s *Stmt) Close() error { - ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) - defer cancel() - return s.conn.conn.Deallocate(ctx, s.sd.Name) -} - -func (s *Stmt) NumInput() int { - return len(s.sd.ParamOIDs) -} - -func (s *Stmt) Exec(argsV []driver.Value) (driver.Result, error) { - return nil, errors.New("Stmt.Exec deprecated and not implemented") -} - -func (s *Stmt) ExecContext(ctx context.Context, argsV []driver.NamedValue) (driver.Result, error) { - return s.conn.ExecContext(ctx, s.sd.Name, argsV) -} - -func (s *Stmt) Query(argsV []driver.Value) (driver.Rows, error) { - return nil, errors.New("Stmt.Query deprecated and not implemented") -} - -func (s *Stmt) QueryContext(ctx context.Context, argsV []driver.NamedValue) (driver.Rows, error) { - return s.conn.QueryContext(ctx, s.sd.Name, argsV) -} - -type rowValueFunc func(src []byte) (driver.Value, error) - -type Rows struct { - conn *Conn - rows pgx.Rows - valueFuncs []rowValueFunc - skipNext bool - skipNextMore bool - - columnNames []string -} - -func (r *Rows) Columns() []string { - if r.columnNames == nil { - fields := r.rows.FieldDescriptions() - r.columnNames = make([]string, len(fields)) - for i, fd := range fields { - r.columnNames[i] = string(fd.Name) - } - } - - return r.columnNames -} - -// ColumnTypeDatabaseTypeName returns the database system type name. If the name is unknown the OID is returned. -func (r *Rows) ColumnTypeDatabaseTypeName(index int) string { - if dt, ok := r.conn.conn.ConnInfo().DataTypeForOID(r.rows.FieldDescriptions()[index].DataTypeOID); ok { - return strings.ToUpper(dt.Name) - } - - return strconv.FormatInt(int64(r.rows.FieldDescriptions()[index].DataTypeOID), 10) -} - -const varHeaderSize = 4 - -// ColumnTypeLength returns the length of the column type if the column is a -// variable length type. If the column is not a variable length type ok -// should return false. -func (r *Rows) ColumnTypeLength(index int) (int64, bool) { - fd := r.rows.FieldDescriptions()[index] - - switch fd.DataTypeOID { - case pgtype.TextOID, pgtype.ByteaOID: - return math.MaxInt64, true - case pgtype.VarcharOID, pgtype.BPCharArrayOID: - return int64(fd.TypeModifier - varHeaderSize), true - default: - return 0, false - } -} - -// ColumnTypePrecisionScale should return the precision and scale for decimal -// types. If not applicable, ok should be false. -func (r *Rows) ColumnTypePrecisionScale(index int) (precision, scale int64, ok bool) { - fd := r.rows.FieldDescriptions()[index] - - switch fd.DataTypeOID { - case pgtype.NumericOID: - mod := fd.TypeModifier - varHeaderSize - precision = int64((mod >> 16) & 0xffff) - scale = int64(mod & 0xffff) - return precision, scale, true - default: - return 0, 0, false - } -} - -// ColumnTypeScanType returns the value type that can be used to scan types into. -func (r *Rows) ColumnTypeScanType(index int) reflect.Type { - fd := r.rows.FieldDescriptions()[index] - - switch fd.DataTypeOID { - case pgtype.Float8OID: - return reflect.TypeOf(float64(0)) - case pgtype.Float4OID: - return reflect.TypeOf(float32(0)) - case pgtype.Int8OID: - return reflect.TypeOf(int64(0)) - case pgtype.Int4OID: - return reflect.TypeOf(int32(0)) - case pgtype.Int2OID: - return reflect.TypeOf(int16(0)) - case pgtype.BoolOID: - return reflect.TypeOf(false) - case pgtype.NumericOID: - return reflect.TypeOf(float64(0)) - case pgtype.DateOID, pgtype.TimestampOID, pgtype.TimestamptzOID: - return reflect.TypeOf(time.Time{}) - case pgtype.ByteaOID: - return reflect.TypeOf([]byte(nil)) - default: - return reflect.TypeOf("") - } -} - -func (r *Rows) Close() error { - r.rows.Close() - return r.rows.Err() -} - -func (r *Rows) Next(dest []driver.Value) error { - ci := r.conn.conn.ConnInfo() - fieldDescriptions := r.rows.FieldDescriptions() - - if r.valueFuncs == nil { - r.valueFuncs = make([]rowValueFunc, len(fieldDescriptions)) - - for i, fd := range fieldDescriptions { - dataTypeOID := fd.DataTypeOID - format := fd.Format - - switch fd.DataTypeOID { - case pgtype.BoolOID: - var d bool - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - return d, err - } - case pgtype.ByteaOID: - var d []byte - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - return d, err - } - case pgtype.CIDOID: - var d pgtype.CID - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - if err != nil { - return nil, err - } - return d.Value() - } - case pgtype.DateOID: - var d pgtype.Date - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - if err != nil { - return nil, err - } - return d.Value() - } - case pgtype.Float4OID: - var d float32 - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - return float64(d), err - } - case pgtype.Float8OID: - var d float64 - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - return d, err - } - case pgtype.Int2OID: - var d int16 - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - return int64(d), err - } - case pgtype.Int4OID: - var d int32 - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - return int64(d), err - } - case pgtype.Int8OID: - var d int64 - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - return d, err - } - case pgtype.JSONOID: - var d pgtype.JSON - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - if err != nil { - return nil, err - } - return d.Value() - } - case pgtype.JSONBOID: - var d pgtype.JSONB - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - if err != nil { - return nil, err - } - return d.Value() - } - case pgtype.OIDOID: - var d pgtype.OIDValue - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - if err != nil { - return nil, err - } - return d.Value() - } - case pgtype.TimestampOID: - var d pgtype.Timestamp - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - if err != nil { - return nil, err - } - return d.Value() - } - case pgtype.TimestamptzOID: - var d pgtype.Timestamptz - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - if err != nil { - return nil, err - } - return d.Value() - } - case pgtype.XIDOID: - var d pgtype.XID - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - if err != nil { - return nil, err - } - return d.Value() - } - default: - var d string - scanPlan := ci.PlanScan(dataTypeOID, format, &d) - r.valueFuncs[i] = func(src []byte) (driver.Value, error) { - err := scanPlan.Scan(ci, dataTypeOID, format, src, &d) - return d, err - } - } - } - } - - var more bool - if r.skipNext { - more = r.skipNextMore - r.skipNext = false - } else { - more = r.rows.Next() - } - - if !more { - if r.rows.Err() == nil { - return io.EOF - } else { - return r.rows.Err() - } - } - - for i, rv := range r.rows.RawValues() { - if rv != nil { - var err error - dest[i], err = r.valueFuncs[i](rv) - if err != nil { - return fmt.Errorf("convert field %d failed: %v", i, err) - } - } else { - dest[i] = nil - } - } - - return nil -} - -func valueToInterface(argsV []driver.Value) []interface{} { - args := make([]interface{}, 0, len(argsV)) - for _, v := range argsV { - if v != nil { - args = append(args, v.(interface{})) - } else { - args = append(args, nil) - } - } - return args -} - -func namedValueToInterface(argsV []driver.NamedValue) []interface{} { - args := make([]interface{}, 0, len(argsV)) - for _, v := range argsV { - if v.Value != nil { - args = append(args, v.Value.(interface{})) - } else { - args = append(args, nil) - } - } - return args -} - -type wrapTx struct { - ctx context.Context - tx pgx.Tx -} - -func (wtx wrapTx) Commit() error { return wtx.tx.Commit(wtx.ctx) } - -func (wtx wrapTx) Rollback() error { return wtx.tx.Rollback(wtx.ctx) } - -type fakeTx struct{} - -func (fakeTx) Commit() error { return nil } - -func (fakeTx) Rollback() error { return nil } - -// AcquireConn acquires a *pgx.Conn from database/sql connection pool. It must be released with ReleaseConn. -// -// In Go 1.13 this functionality has been incorporated into the standard library in the db.Conn.Raw() method. -func AcquireConn(db *sql.DB) (*pgx.Conn, error) { - var conn *pgx.Conn - ctx := context.WithValue(context.Background(), ctxKeyFakeTx, &conn) - tx, err := db.BeginTx(ctx, nil) - if err != nil { - return nil, err - } - if conn == nil { - tx.Rollback() - return nil, ErrNotPgx - } - - fakeTxMutex.Lock() - fakeTxConns[conn] = tx - fakeTxMutex.Unlock() - - return conn, nil -} - -// ReleaseConn releases a *pgx.Conn acquired with AcquireConn. -func ReleaseConn(db *sql.DB, conn *pgx.Conn) error { - var tx *sql.Tx - var ok bool - - if conn.PgConn().IsBusy() || conn.PgConn().TxStatus() != 'I' { - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - conn.Close(ctx) - } - - fakeTxMutex.Lock() - tx, ok = fakeTxConns[conn] - if ok { - delete(fakeTxConns, conn) - fakeTxMutex.Unlock() - } else { - fakeTxMutex.Unlock() - return fmt.Errorf("can't release conn that is not acquired") - } - - return tx.Rollback() -} |