375 lines
7.8 KiB
Go
375 lines
7.8 KiB
Go
// Go MySQL Driver - A MySQL-Driver for Go's database/sql package
|
|
//
|
|
// Copyright 2012 The Go-MySQL-Driver Authors. All rights reserved.
|
|
//
|
|
// This Source Code Form is subject to the terms of the Mozilla Public
|
|
// License, v. 2.0. If a copy of the MPL was not distributed with this file,
|
|
// You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
|
|
package driver
|
|
|
|
import (
|
|
"context"
|
|
"database/sql/driver"
|
|
"net"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
type mysqlConn struct {
|
|
buf buffer
|
|
netConn net.Conn
|
|
affectedRows uint64
|
|
insertID uint64
|
|
cfg *Config
|
|
maxAllowedPacket int
|
|
maxWriteSize int
|
|
writeTimeout time.Duration
|
|
flags clientFlag
|
|
status statusFlag
|
|
sequence uint8
|
|
parseTime bool
|
|
|
|
// for context support (Go 1.8+)
|
|
watching bool
|
|
watcher chan<- context.Context
|
|
closech chan struct{}
|
|
finished chan<- struct{}
|
|
canceled atomicError // set non-nil if conn is canceled
|
|
closed atomicBool // set when conn is closed, before closech is closed
|
|
}
|
|
|
|
func newConnection(dsn string) (*mysqlConn, error) {
|
|
var err error
|
|
|
|
// New mysqlConn
|
|
mc := &mysqlConn{
|
|
maxAllowedPacket: maxPacketSize,
|
|
maxWriteSize: maxPacketSize - 1,
|
|
closech: make(chan struct{}),
|
|
}
|
|
mc.cfg, err = ParseDSN(dsn)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
mc.parseTime = mc.cfg.ParseTime
|
|
|
|
// Connect to Server
|
|
nd := net.Dialer{Timeout: mc.cfg.Timeout}
|
|
mc.netConn, err = nd.Dial(mc.cfg.Net, mc.cfg.Addr)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Enable TCP Keepalives on TCP connections
|
|
if tc, ok := mc.netConn.(*net.TCPConn); ok {
|
|
if err := tc.SetKeepAlive(true); err != nil {
|
|
// Don't send COM_QUIT before handshake.
|
|
mc.netConn.Close()
|
|
mc.netConn = nil
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
// Call startWatcher for context support (From Go 1.8)
|
|
mc.startWatcher()
|
|
|
|
mc.buf = newBuffer(mc.netConn)
|
|
|
|
// Set I/O timeouts
|
|
mc.buf.timeout = mc.cfg.ReadTimeout
|
|
mc.writeTimeout = mc.cfg.WriteTimeout
|
|
|
|
// Reading Handshake Initialization Packet
|
|
authData, plugin, err := mc.readHandshakePacket()
|
|
if err != nil {
|
|
mc.cleanup()
|
|
return nil, err
|
|
}
|
|
|
|
// Send Client Authentication Packet
|
|
authResp, addNUL, err := mc.auth(authData, plugin)
|
|
if err != nil {
|
|
// try the default auth plugin, if using the requested plugin failed
|
|
errLog.Print("could not use requested auth plugin '"+plugin+"': ", err.Error())
|
|
plugin = defaultAuthPlugin
|
|
authResp, addNUL, err = mc.auth(authData, plugin)
|
|
if err != nil {
|
|
mc.cleanup()
|
|
return nil, err
|
|
}
|
|
}
|
|
if err = mc.writeHandshakeResponsePacket(authResp, addNUL, plugin); err != nil {
|
|
mc.cleanup()
|
|
return nil, err
|
|
}
|
|
|
|
// Handle response to auth packet, switch methods if possible
|
|
if err = mc.handleAuthResult(authData, plugin); err != nil {
|
|
// Authentication failed and MySQL has already closed the connection
|
|
// (https://dev.mysql.com/doc/internals/en/authentication-fails.html).
|
|
// Do not send COM_QUIT, just cleanup and return the error.
|
|
mc.cleanup()
|
|
return nil, err
|
|
}
|
|
|
|
if mc.cfg.MaxAllowedPacket > 0 {
|
|
mc.maxAllowedPacket = mc.cfg.MaxAllowedPacket
|
|
} else {
|
|
// Get max allowed packet size
|
|
maxap, err := mc.getSystemVar("max_allowed_packet")
|
|
if err != nil {
|
|
mc.Close()
|
|
return nil, err
|
|
}
|
|
mc.maxAllowedPacket = stringToInt(maxap) - 1
|
|
}
|
|
if mc.maxAllowedPacket < maxPacketSize {
|
|
mc.maxWriteSize = mc.maxAllowedPacket
|
|
}
|
|
|
|
// Handle DSN Params
|
|
err = mc.handleParams()
|
|
if err != nil {
|
|
mc.Close()
|
|
return nil, err
|
|
}
|
|
|
|
return mc, nil
|
|
}
|
|
|
|
// Handles parameters set in DSN after the connection is established
|
|
func (mc *mysqlConn) handleParams() (err error) {
|
|
for param, val := range mc.cfg.Params {
|
|
switch param {
|
|
// Charset
|
|
case "charset":
|
|
charsets := strings.Split(val, ",")
|
|
for i := range charsets {
|
|
// ignore errors here - a charset may not exist
|
|
err = mc.exec("SET NAMES " + charsets[i])
|
|
if err == nil {
|
|
break
|
|
}
|
|
}
|
|
if err != nil {
|
|
return
|
|
}
|
|
|
|
// System Vars
|
|
default:
|
|
err = mc.exec("SET " + param + "=" + val)
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
func (mc *mysqlConn) markBadConn(err error) error {
|
|
if mc == nil {
|
|
return err
|
|
}
|
|
if err != errBadConnNoWrite {
|
|
return err
|
|
}
|
|
return driver.ErrBadConn
|
|
}
|
|
|
|
func (mc *mysqlConn) Close() (err error) {
|
|
// Makes Close idempotent
|
|
if !mc.closed.IsSet() {
|
|
err = mc.writeCommandPacket(comQuit)
|
|
}
|
|
|
|
mc.cleanup()
|
|
|
|
return
|
|
}
|
|
|
|
// Closes the network connection and unsets internal variables. Do not call this
|
|
// function after successfully authentication, call Close instead. This function
|
|
// is called before auth or on auth failure because MySQL will have already
|
|
// closed the network connection.
|
|
func (mc *mysqlConn) cleanup() {
|
|
if !mc.closed.TrySet(true) {
|
|
return
|
|
}
|
|
|
|
// Makes cleanup idempotent
|
|
close(mc.closech)
|
|
if mc.netConn == nil {
|
|
return
|
|
}
|
|
if err := mc.netConn.Close(); err != nil {
|
|
errLog.Print(err)
|
|
}
|
|
}
|
|
|
|
func (mc *mysqlConn) error() error {
|
|
if mc.closed.IsSet() {
|
|
if err := mc.canceled.Value(); err != nil {
|
|
return err
|
|
}
|
|
return ErrInvalidConn
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Internal function to execute commands
|
|
func (mc *mysqlConn) exec(query string) error {
|
|
// Send command
|
|
if err := mc.writeCommandPacketStr(comQuery, query); err != nil {
|
|
return mc.markBadConn(err)
|
|
}
|
|
|
|
// Read Result
|
|
resLen, err := mc.readResultSetHeaderPacket()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if resLen > 0 {
|
|
// columns
|
|
if err := mc.readUntilEOF(); err != nil {
|
|
return err
|
|
}
|
|
|
|
// rows
|
|
if err := mc.readUntilEOF(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return mc.discardResults()
|
|
}
|
|
|
|
// Gets the value of the given MySQL System Variable
|
|
// The returned byte slice is only valid until the next read
|
|
func (mc *mysqlConn) getSystemVar(name string) ([]byte, error) {
|
|
// Send command
|
|
if err := mc.writeCommandPacketStr(comQuery, "SELECT @@"+name); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Read Result
|
|
resLen, err := mc.readResultSetHeaderPacket()
|
|
if err == nil {
|
|
rows := new(textRows)
|
|
rows.mc = mc
|
|
rows.rs.columns = []mysqlField{{fieldType: fieldTypeVarChar}}
|
|
|
|
if resLen > 0 {
|
|
// Columns
|
|
if err := mc.readUntilEOF(); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
dest := make([]driver.Value, resLen)
|
|
if err = rows.readRow(dest); err == nil {
|
|
return dest[0].([]byte), mc.readUntilEOF()
|
|
}
|
|
}
|
|
return nil, err
|
|
}
|
|
|
|
// finish is called when the query has canceled.
|
|
func (mc *mysqlConn) cancel(err error) {
|
|
mc.canceled.Set(err)
|
|
mc.cleanup()
|
|
}
|
|
|
|
// finish is called when the query has succeeded.
|
|
func (mc *mysqlConn) finish() {
|
|
if !mc.watching || mc.finished == nil {
|
|
return
|
|
}
|
|
select {
|
|
case mc.finished <- struct{}{}:
|
|
mc.watching = false
|
|
case <-mc.closech:
|
|
}
|
|
}
|
|
|
|
// Ping implements driver.Pinger interface
|
|
func (mc *mysqlConn) Ping(ctx context.Context) (err error) {
|
|
if mc.closed.IsSet() {
|
|
errLog.Print(ErrInvalidConn)
|
|
return driver.ErrBadConn
|
|
}
|
|
|
|
if err = mc.watchCancel(ctx); err != nil {
|
|
return
|
|
}
|
|
defer mc.finish()
|
|
|
|
if err = mc.writeCommandPacket(comPing); err != nil {
|
|
return
|
|
}
|
|
|
|
return mc.readResultOK()
|
|
}
|
|
|
|
func (mc *mysqlConn) watchCancel(ctx context.Context) error {
|
|
if mc.watching {
|
|
// Reach here if canceled,
|
|
// so the connection is already invalid
|
|
mc.cleanup()
|
|
return nil
|
|
}
|
|
if ctx.Done() == nil {
|
|
return nil
|
|
}
|
|
|
|
mc.watching = true
|
|
select {
|
|
default:
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
if mc.watcher == nil {
|
|
return nil
|
|
}
|
|
|
|
mc.watcher <- ctx
|
|
|
|
return nil
|
|
}
|
|
|
|
func (mc *mysqlConn) startWatcher() {
|
|
watcher := make(chan context.Context, 1)
|
|
mc.watcher = watcher
|
|
finished := make(chan struct{})
|
|
mc.finished = finished
|
|
go func() {
|
|
for {
|
|
var ctx context.Context
|
|
select {
|
|
case ctx = <-watcher:
|
|
case <-mc.closech:
|
|
return
|
|
}
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
mc.cancel(ctx.Err())
|
|
case <-finished:
|
|
case <-mc.closech:
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
// ResetSession implements driver.SessionResetter.
|
|
// (From Go 1.10)
|
|
func (mc *mysqlConn) ResetSession(ctx context.Context) error {
|
|
if mc.closed.IsSet() {
|
|
return driver.ErrBadConn
|
|
}
|
|
return nil
|
|
}
|