connection_go18.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204
  1. // Go MySQL Driver - A MySQL-Driver for Go's database/sql package
  2. //
  3. // Copyright 2012 The Go-MySQL-Driver Authors. All rights reserved.
  4. //
  5. // This Source Code Form is subject to the terms of the Mozilla Public
  6. // License, v. 2.0. If a copy of the MPL was not distributed with this file,
  7. // You can obtain one at http://mozilla.org/MPL/2.0/.
  8. // +build go1.8
  9. package mysql
  10. import (
  11. "context"
  12. "database/sql"
  13. "database/sql/driver"
  14. "errors"
  15. )
  16. // Ping implements driver.Pinger interface
  17. func (mc *mysqlConn) Ping(ctx context.Context) error {
  18. if mc.closed.IsSet() {
  19. errLog.Print(ErrInvalidConn)
  20. return driver.ErrBadConn
  21. }
  22. if err := mc.watchCancel(ctx); err != nil {
  23. return err
  24. }
  25. defer mc.finish()
  26. if err := mc.writeCommandPacket(comPing); err != nil {
  27. return err
  28. }
  29. if _, err := mc.readResultOK(); err != nil {
  30. return err
  31. }
  32. return nil
  33. }
  34. // BeginTx implements driver.ConnBeginTx interface
  35. func (mc *mysqlConn) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
  36. if opts.ReadOnly {
  37. // TODO: support read-only transactions
  38. return nil, errors.New("mysql: read-only transactions not supported")
  39. }
  40. if err := mc.watchCancel(ctx); err != nil {
  41. return nil, err
  42. }
  43. defer mc.finish()
  44. if sql.IsolationLevel(opts.Isolation) != sql.LevelDefault {
  45. level, err := mapIsolationLevel(opts.Isolation)
  46. if err != nil {
  47. return nil, err
  48. }
  49. err = mc.exec("SET TRANSACTION ISOLATION LEVEL " + level)
  50. if err != nil {
  51. return nil, err
  52. }
  53. }
  54. return mc.Begin()
  55. }
  56. func (mc *mysqlConn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
  57. dargs, err := namedValueToValue(args)
  58. if err != nil {
  59. return nil, err
  60. }
  61. if err := mc.watchCancel(ctx); err != nil {
  62. return nil, err
  63. }
  64. rows, err := mc.query(query, dargs)
  65. if err != nil {
  66. mc.finish()
  67. return nil, err
  68. }
  69. rows.finish = mc.finish
  70. return rows, err
  71. }
  72. func (mc *mysqlConn) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
  73. dargs, err := namedValueToValue(args)
  74. if err != nil {
  75. return nil, err
  76. }
  77. if err := mc.watchCancel(ctx); err != nil {
  78. return nil, err
  79. }
  80. defer mc.finish()
  81. return mc.Exec(query, dargs)
  82. }
  83. func (mc *mysqlConn) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
  84. if err := mc.watchCancel(ctx); err != nil {
  85. return nil, err
  86. }
  87. stmt, err := mc.Prepare(query)
  88. mc.finish()
  89. if err != nil {
  90. return nil, err
  91. }
  92. select {
  93. default:
  94. case <-ctx.Done():
  95. stmt.Close()
  96. return nil, ctx.Err()
  97. }
  98. return stmt, nil
  99. }
  100. func (stmt *mysqlStmt) QueryContext(ctx context.Context, args []driver.NamedValue) (driver.Rows, error) {
  101. dargs, err := namedValueToValue(args)
  102. if err != nil {
  103. return nil, err
  104. }
  105. if err := stmt.mc.watchCancel(ctx); err != nil {
  106. return nil, err
  107. }
  108. rows, err := stmt.query(dargs)
  109. if err != nil {
  110. stmt.mc.finish()
  111. return nil, err
  112. }
  113. rows.finish = stmt.mc.finish
  114. return rows, err
  115. }
  116. func (stmt *mysqlStmt) ExecContext(ctx context.Context, args []driver.NamedValue) (driver.Result, error) {
  117. dargs, err := namedValueToValue(args)
  118. if err != nil {
  119. return nil, err
  120. }
  121. if err := stmt.mc.watchCancel(ctx); err != nil {
  122. return nil, err
  123. }
  124. defer stmt.mc.finish()
  125. return stmt.Exec(dargs)
  126. }
  127. func (mc *mysqlConn) watchCancel(ctx context.Context) error {
  128. if mc.watching {
  129. // Reach here if canceled,
  130. // so the connection is already invalid
  131. mc.cleanup()
  132. return nil
  133. }
  134. if ctx.Done() == nil {
  135. return nil
  136. }
  137. mc.watching = true
  138. select {
  139. default:
  140. case <-ctx.Done():
  141. return ctx.Err()
  142. }
  143. if mc.watcher == nil {
  144. return nil
  145. }
  146. mc.watcher <- ctx
  147. return nil
  148. }
  149. func (mc *mysqlConn) startWatcher() {
  150. watcher := make(chan mysqlContext, 1)
  151. mc.watcher = watcher
  152. finished := make(chan struct{})
  153. mc.finished = finished
  154. go func() {
  155. for {
  156. var ctx mysqlContext
  157. select {
  158. case ctx = <-watcher:
  159. case <-mc.closech:
  160. return
  161. }
  162. select {
  163. case <-ctx.Done():
  164. mc.cancel(ctx.Err())
  165. case <-finished:
  166. case <-mc.closech:
  167. return
  168. }
  169. }
  170. }()
  171. }