prepare_stmt.go 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157
  1. package gorm
  2. import (
  3. "context"
  4. "database/sql"
  5. "sync"
  6. )
  7. type Stmt struct {
  8. *sql.Stmt
  9. Transaction bool
  10. }
  11. type PreparedStmtDB struct {
  12. Stmts map[string]Stmt
  13. PreparedSQL []string
  14. Mux *sync.RWMutex
  15. ConnPool
  16. }
  17. func (db *PreparedStmtDB) Close() {
  18. db.Mux.Lock()
  19. for _, query := range db.PreparedSQL {
  20. if stmt, ok := db.Stmts[query]; ok {
  21. delete(db.Stmts, query)
  22. stmt.Close()
  23. }
  24. }
  25. db.Mux.Unlock()
  26. }
  27. func (db *PreparedStmtDB) prepare(ctx context.Context, conn ConnPool, isTransaction bool, query string) (Stmt, error) {
  28. db.Mux.RLock()
  29. if stmt, ok := db.Stmts[query]; ok && (!stmt.Transaction || isTransaction) {
  30. db.Mux.RUnlock()
  31. return stmt, nil
  32. }
  33. db.Mux.RUnlock()
  34. db.Mux.Lock()
  35. // double check
  36. if stmt, ok := db.Stmts[query]; ok && (!stmt.Transaction || isTransaction) {
  37. db.Mux.Unlock()
  38. return stmt, nil
  39. } else if ok {
  40. stmt.Close()
  41. }
  42. stmt, err := conn.PrepareContext(ctx, query)
  43. if err == nil {
  44. db.Stmts[query] = Stmt{Stmt: stmt, Transaction: isTransaction}
  45. db.PreparedSQL = append(db.PreparedSQL, query)
  46. }
  47. db.Mux.Unlock()
  48. return db.Stmts[query], err
  49. }
  50. func (db *PreparedStmtDB) BeginTx(ctx context.Context, opt *sql.TxOptions) (ConnPool, error) {
  51. if beginner, ok := db.ConnPool.(TxBeginner); ok {
  52. tx, err := beginner.BeginTx(ctx, opt)
  53. return &PreparedStmtTX{PreparedStmtDB: db, Tx: tx}, err
  54. }
  55. return nil, ErrInvalidTransaction
  56. }
  57. func (db *PreparedStmtDB) ExecContext(ctx context.Context, query string, args ...interface{}) (result sql.Result, err error) {
  58. stmt, err := db.prepare(ctx, db.ConnPool, false, query)
  59. if err == nil {
  60. result, err = stmt.ExecContext(ctx, args...)
  61. if err != nil {
  62. db.Mux.Lock()
  63. stmt.Close()
  64. delete(db.Stmts, query)
  65. db.Mux.Unlock()
  66. }
  67. }
  68. return result, err
  69. }
  70. func (db *PreparedStmtDB) QueryContext(ctx context.Context, query string, args ...interface{}) (rows *sql.Rows, err error) {
  71. stmt, err := db.prepare(ctx, db.ConnPool, false, query)
  72. if err == nil {
  73. rows, err = stmt.QueryContext(ctx, args...)
  74. if err != nil {
  75. db.Mux.Lock()
  76. stmt.Close()
  77. delete(db.Stmts, query)
  78. db.Mux.Unlock()
  79. }
  80. }
  81. return rows, err
  82. }
  83. func (db *PreparedStmtDB) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row {
  84. stmt, err := db.prepare(ctx, db.ConnPool, false, query)
  85. if err == nil {
  86. return stmt.QueryRowContext(ctx, args...)
  87. }
  88. return &sql.Row{}
  89. }
  90. type PreparedStmtTX struct {
  91. *sql.Tx
  92. PreparedStmtDB *PreparedStmtDB
  93. }
  94. func (tx *PreparedStmtTX) Commit() error {
  95. if tx.Tx != nil {
  96. return tx.Tx.Commit()
  97. }
  98. return ErrInvalidTransaction
  99. }
  100. func (tx *PreparedStmtTX) Rollback() error {
  101. if tx.Tx != nil {
  102. return tx.Tx.Rollback()
  103. }
  104. return ErrInvalidTransaction
  105. }
  106. func (tx *PreparedStmtTX) ExecContext(ctx context.Context, query string, args ...interface{}) (result sql.Result, err error) {
  107. stmt, err := tx.PreparedStmtDB.prepare(ctx, tx.Tx, true, query)
  108. if err == nil {
  109. result, err = tx.Tx.StmtContext(ctx, stmt.Stmt).ExecContext(ctx, args...)
  110. if err != nil {
  111. tx.PreparedStmtDB.Mux.Lock()
  112. stmt.Close()
  113. delete(tx.PreparedStmtDB.Stmts, query)
  114. tx.PreparedStmtDB.Mux.Unlock()
  115. }
  116. }
  117. return result, err
  118. }
  119. func (tx *PreparedStmtTX) QueryContext(ctx context.Context, query string, args ...interface{}) (rows *sql.Rows, err error) {
  120. stmt, err := tx.PreparedStmtDB.prepare(ctx, tx.Tx, true, query)
  121. if err == nil {
  122. rows, err = tx.Tx.Stmt(stmt.Stmt).QueryContext(ctx, args...)
  123. if err != nil {
  124. tx.PreparedStmtDB.Mux.Lock()
  125. stmt.Close()
  126. delete(tx.PreparedStmtDB.Stmts, query)
  127. tx.PreparedStmtDB.Mux.Unlock()
  128. }
  129. }
  130. return rows, err
  131. }
  132. func (tx *PreparedStmtTX) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row {
  133. stmt, err := tx.PreparedStmtDB.prepare(ctx, tx.Tx, true, query)
  134. if err == nil {
  135. return tx.Tx.StmtContext(ctx, stmt.Stmt).QueryRowContext(ctx, args...)
  136. }
  137. return &sql.Row{}
  138. }