123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172 |
- package mysql
- import (
- "database/sql"
- "database/sql/driver"
- "net"
- "sync"
- )
- type watcher interface {
- startWatcher()
- }
- type MySQLDriver struct{}
- type DialFunc func(addr string) (net.Conn, error)
- var (
- dialsLock sync.RWMutex
- dials map[string]DialFunc
- )
- func RegisterDial(net string, dial DialFunc) {
- dialsLock.Lock()
- defer dialsLock.Unlock()
- if dials == nil {
- dials = make(map[string]DialFunc)
- }
- dials[net] = dial
- }
- func (d MySQLDriver) Open(dsn string) (driver.Conn, error) {
- var err error
-
- 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
-
- dialsLock.RLock()
- dial, ok := dials[mc.cfg.Net]
- dialsLock.RUnlock()
- if ok {
- mc.netConn, err = dial(mc.cfg.Addr)
- } else {
- nd := net.Dialer{Timeout: mc.cfg.Timeout}
- mc.netConn, err = nd.Dial(mc.cfg.Net, mc.cfg.Addr)
- }
- if err != nil {
- return nil, err
- }
-
- if tc, ok := mc.netConn.(*net.TCPConn); ok {
- if err := tc.SetKeepAlive(true); err != nil {
-
- mc.netConn.Close()
- mc.netConn = nil
- return nil, err
- }
- }
-
- if s, ok := interface{}(mc).(watcher); ok {
- s.startWatcher()
- }
- mc.buf = newBuffer(mc.netConn)
-
- mc.buf.timeout = mc.cfg.ReadTimeout
- mc.writeTimeout = mc.cfg.WriteTimeout
-
- authData, plugin, err := mc.readHandshakePacket()
- if err != nil {
- mc.cleanup()
- return nil, err
- }
- if plugin == "" {
- plugin = defaultAuthPlugin
- }
-
- authResp, err := mc.auth(authData, plugin)
- if err != nil {
-
- errLog.Print("could not use requested auth plugin '"+plugin+"': ", err.Error())
- plugin = defaultAuthPlugin
- authResp, err = mc.auth(authData, plugin)
- if err != nil {
- mc.cleanup()
- return nil, err
- }
- }
- if err = mc.writeHandshakeResponsePacket(authResp, plugin); err != nil {
- mc.cleanup()
- return nil, err
- }
-
- if err = mc.handleAuthResult(authData, plugin); err != nil {
-
-
-
- mc.cleanup()
- return nil, err
- }
- if mc.cfg.MaxAllowedPacket > 0 {
- mc.maxAllowedPacket = mc.cfg.MaxAllowedPacket
- } else {
-
- 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
- }
-
- err = mc.handleParams()
- if err != nil {
- mc.Close()
- return nil, err
- }
- return mc, nil
- }
- func init() {
- sql.Register("mysql", &MySQLDriver{})
- }
|