mirror of https://github.com/go-redis/redis.git
147 lines
2.9 KiB
Go
147 lines
2.9 KiB
Go
package redis
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
)
|
|
|
|
var errDiscard = errors.New("redis: Discard can be used only inside Exec")
|
|
|
|
// Multi implements Redis transactions as described in
|
|
// http://redis.io/topics/transactions.
|
|
type Multi struct {
|
|
commandable
|
|
|
|
base *baseClient
|
|
cmds []Cmder
|
|
}
|
|
|
|
func (c *Client) Multi() *Multi {
|
|
multi := &Multi{
|
|
base: &baseClient{
|
|
opt: c.opt,
|
|
connPool: newSingleConnPool(c.connPool, true),
|
|
},
|
|
}
|
|
multi.commandable.process = multi.process
|
|
return multi
|
|
}
|
|
|
|
func (c *Multi) process(cmd Cmder) {
|
|
if c.cmds == nil {
|
|
c.base.process(cmd)
|
|
} else {
|
|
c.cmds = append(c.cmds, cmd)
|
|
}
|
|
}
|
|
|
|
func (c *Multi) Close() error {
|
|
if err := c.Unwatch().Err(); err != nil {
|
|
log.Printf("redis: Unwatch failed: %s", err)
|
|
}
|
|
return c.base.Close()
|
|
}
|
|
|
|
func (c *Multi) Watch(keys ...string) *StatusCmd {
|
|
args := append([]string{"WATCH"}, keys...)
|
|
cmd := NewStatusCmd(args...)
|
|
c.Process(cmd)
|
|
return cmd
|
|
}
|
|
|
|
func (c *Multi) Unwatch(keys ...string) *StatusCmd {
|
|
args := append([]string{"UNWATCH"}, keys...)
|
|
cmd := NewStatusCmd(args...)
|
|
c.Process(cmd)
|
|
return cmd
|
|
}
|
|
|
|
func (c *Multi) Discard() error {
|
|
if c.cmds == nil {
|
|
return errDiscard
|
|
}
|
|
c.cmds = c.cmds[:1]
|
|
return nil
|
|
}
|
|
|
|
// Exec always returns list of commands. If transaction fails
|
|
// TxFailedErr is returned. Otherwise Exec returns error of the first
|
|
// failed command or nil.
|
|
func (c *Multi) Exec(f func() error) ([]Cmder, error) {
|
|
c.cmds = []Cmder{NewStatusCmd("MULTI")}
|
|
if err := f(); err != nil {
|
|
return nil, err
|
|
}
|
|
c.cmds = append(c.cmds, NewSliceCmd("EXEC"))
|
|
|
|
cmds := c.cmds
|
|
c.cmds = nil
|
|
|
|
if len(cmds) == 2 {
|
|
return []Cmder{}, nil
|
|
}
|
|
|
|
cn, err := c.base.conn()
|
|
if err != nil {
|
|
setCmdsErr(cmds[1:len(cmds)-1], err)
|
|
return cmds[1 : len(cmds)-1], err
|
|
}
|
|
|
|
err = c.execCmds(cn, cmds)
|
|
c.base.putConn(cn, err)
|
|
return cmds[1 : len(cmds)-1], err
|
|
}
|
|
|
|
func (c *Multi) execCmds(cn *conn, cmds []Cmder) error {
|
|
err := cn.writeCmds(cmds...)
|
|
if err != nil {
|
|
setCmdsErr(cmds[1:len(cmds)-1], err)
|
|
return err
|
|
}
|
|
|
|
statusCmd := NewStatusCmd()
|
|
|
|
// Omit last command (EXEC).
|
|
cmdsLen := len(cmds) - 1
|
|
|
|
// Parse queued replies.
|
|
for i := 0; i < cmdsLen; i++ {
|
|
if err := statusCmd.parseReply(cn.rd); err != nil {
|
|
setCmdsErr(cmds[1:len(cmds)-1], err)
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Parse number of replies.
|
|
line, err := readLine(cn.rd)
|
|
if err != nil {
|
|
setCmdsErr(cmds[1:len(cmds)-1], err)
|
|
return err
|
|
}
|
|
if line[0] != '*' {
|
|
err := fmt.Errorf("redis: expected '*', but got line %q", line)
|
|
setCmdsErr(cmds[1:len(cmds)-1], err)
|
|
return err
|
|
}
|
|
if len(line) == 3 && line[1] == '-' && line[2] == '1' {
|
|
setCmdsErr(cmds[1:len(cmds)-1], TxFailedErr)
|
|
return TxFailedErr
|
|
}
|
|
|
|
var firstCmdErr error
|
|
|
|
// Parse replies.
|
|
// Loop starts from 1 to omit MULTI cmd.
|
|
for i := 1; i < cmdsLen; i++ {
|
|
cmd := cmds[i]
|
|
if err := cmd.parseReply(cn.rd); err != nil {
|
|
if firstCmdErr == nil {
|
|
firstCmdErr = err
|
|
}
|
|
}
|
|
}
|
|
|
|
return firstCmdErr
|
|
}
|