You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

tx.go 3.8KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148
  1. package redis
  2. import (
  3. "context"
  4. "github.com/go-redis/redis/v8/internal/pool"
  5. "github.com/go-redis/redis/v8/internal/proto"
  6. )
  7. // TxFailedErr transaction redis failed.
  8. const TxFailedErr = proto.RedisError("redis: transaction failed")
  9. // Tx implements Redis transactions as described in
  10. // http://redis.io/topics/transactions. It's NOT safe for concurrent use
  11. // by multiple goroutines, because Exec resets list of watched keys.
  12. // If you don't need WATCH it is better to use Pipeline.
  13. type Tx struct {
  14. baseClient
  15. cmdable
  16. statefulCmdable
  17. hooks
  18. ctx context.Context
  19. }
  20. func (c *Client) newTx(ctx context.Context) *Tx {
  21. tx := Tx{
  22. baseClient: baseClient{
  23. opt: c.opt,
  24. connPool: pool.NewStickyConnPool(c.connPool),
  25. },
  26. hooks: c.hooks.clone(),
  27. ctx: ctx,
  28. }
  29. tx.init()
  30. return &tx
  31. }
  32. func (c *Tx) init() {
  33. c.cmdable = c.Process
  34. c.statefulCmdable = c.Process
  35. }
  36. func (c *Tx) Context() context.Context {
  37. return c.ctx
  38. }
  39. func (c *Tx) WithContext(ctx context.Context) *Tx {
  40. if ctx == nil {
  41. panic("nil context")
  42. }
  43. clone := *c
  44. clone.init()
  45. clone.hooks.lock()
  46. clone.ctx = ctx
  47. return &clone
  48. }
  49. func (c *Tx) Process(ctx context.Context, cmd Cmder) error {
  50. return c.hooks.process(ctx, cmd, c.baseClient.process)
  51. }
  52. // Watch prepares a transaction and marks the keys to be watched
  53. // for conditional execution if there are any keys.
  54. //
  55. // The transaction is automatically closed when fn exits.
  56. func (c *Client) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
  57. tx := c.newTx(ctx)
  58. defer tx.Close(ctx)
  59. if len(keys) > 0 {
  60. if err := tx.Watch(ctx, keys...).Err(); err != nil {
  61. return err
  62. }
  63. }
  64. return fn(tx)
  65. }
  66. // Close closes the transaction, releasing any open resources.
  67. func (c *Tx) Close(ctx context.Context) error {
  68. _ = c.Unwatch(ctx).Err()
  69. return c.baseClient.Close()
  70. }
  71. // Watch marks the keys to be watched for conditional execution
  72. // of a transaction.
  73. func (c *Tx) Watch(ctx context.Context, keys ...string) *StatusCmd {
  74. args := make([]interface{}, 1+len(keys))
  75. args[0] = "watch"
  76. for i, key := range keys {
  77. args[1+i] = key
  78. }
  79. cmd := NewStatusCmd(ctx, args...)
  80. _ = c.Process(ctx, cmd)
  81. return cmd
  82. }
  83. // Unwatch flushes all the previously watched keys for a transaction.
  84. func (c *Tx) Unwatch(ctx context.Context, keys ...string) *StatusCmd {
  85. args := make([]interface{}, 1+len(keys))
  86. args[0] = "unwatch"
  87. for i, key := range keys {
  88. args[1+i] = key
  89. }
  90. cmd := NewStatusCmd(ctx, args...)
  91. _ = c.Process(ctx, cmd)
  92. return cmd
  93. }
  94. // Pipeline creates a pipeline. Usually it is more convenient to use Pipelined.
  95. func (c *Tx) Pipeline() Pipeliner {
  96. pipe := Pipeline{
  97. ctx: c.ctx,
  98. exec: func(ctx context.Context, cmds []Cmder) error {
  99. return c.hooks.processPipeline(ctx, cmds, c.baseClient.processPipeline)
  100. },
  101. }
  102. pipe.init()
  103. return &pipe
  104. }
  105. // Pipelined executes commands queued in the fn outside of the transaction.
  106. // Use TxPipelined if you need transactional behavior.
  107. func (c *Tx) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
  108. return c.Pipeline().Pipelined(ctx, fn)
  109. }
  110. // TxPipelined executes commands queued in the fn in the transaction.
  111. //
  112. // When using WATCH, EXEC will execute commands only if the watched keys
  113. // were not modified, allowing for a check-and-set mechanism.
  114. //
  115. // Exec always returns list of commands. If transaction fails
  116. // TxFailedErr is returned. Otherwise Exec returns an error of the first
  117. // failed command or nil.
  118. func (c *Tx) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
  119. return c.TxPipeline().Pipelined(ctx, fn)
  120. }
  121. // TxPipeline creates a pipeline. Usually it is more convenient to use TxPipelined.
  122. func (c *Tx) TxPipeline() Pipeliner {
  123. pipe := Pipeline{
  124. ctx: c.ctx,
  125. exec: func(ctx context.Context, cmds []Cmder) error {
  126. return c.hooks.processTxPipeline(ctx, cmds, c.baseClient.processTxPipeline)
  127. },
  128. }
  129. pipe.init()
  130. return &pipe
  131. }