summaryrefslogtreecommitdiffstats
path: root/vendor/github.com/lib/pq/conn_go18.go
blob: 43cc35f7b4305d871cac8ba10fd35a81b652292a (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
// +build go1.8

package pq

import (
	"context"
	"database/sql/driver"
	"errors"
)

// Implement the "QueryerContext" interface
func (cn *conn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
	list := make([]driver.Value, len(args))
	for i, nv := range args {
		list[i] = nv.Value
	}
	closed := cn.watchCancel(ctx)
	r, err := cn.query(query, list)
	if err != nil {
		return nil, err
	}
	r.closed = closed
	return r, nil
}

// Implement the "ExecerContext" interface
func (cn *conn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
	list := make([]driver.Value, len(args))
	for i, nv := range args {
		list[i] = nv.Value
	}

	if closed := cn.watchCancel(ctx); closed != nil {
		defer close(closed)
	}

	return cn.Exec(query, list)
}

// Implement the "ConnBeginTx" interface
func (cn *conn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
	if opts.Isolation != 0 {
		return nil, errors.New("isolation levels not supported")
	}
	if opts.ReadOnly {
		return nil, errors.New("read-only transactions not supported")
	}
	tx, err := cn.Begin()
	if err != nil {
		return nil, err
	}
	cn.txnClosed = cn.watchCancel(ctx)
	return tx, nil
}

func (cn *conn) watchCancel(ctx context.Context) chan<- struct{} {
	if done := ctx.Done(); done != nil {
		closed := make(chan struct{})
		go func() {
			select {
			case <-done:
				cn.cancel()
			case <-closed:
			}
		}()
		return closed
	}
	return nil
}

func (cn *conn) cancel() {
	var err error
	can := &conn{}
	can.c, err = dial(cn.dialer, cn.opts)
	if err != nil {
		return
	}
	can.ssl(cn.opts)

	defer can.errRecover(&err)

	w := can.writeBuf(0)
	w.int32(80877102) // cancel request code
	w.int32(cn.processID)
	w.int32(cn.secretKey)

	can.sendStartupPacket(w)
	_ = can.c.Close()
}