Skip to content

Commit

Permalink
Add ErrTxFailed. Small tweaks.
Browse files Browse the repository at this point in the history
  • Loading branch information
vmihailenco committed Sep 29, 2013
1 parent 4c504ab commit 39c42d4
Show file tree
Hide file tree
Showing 7 changed files with 52 additions and 28 deletions.
6 changes: 6 additions & 0 deletions v2/command.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,12 @@ type Cmder interface {
Err() error
}

func setCmdsErr(cmds []Cmder, e error) {
for _, cmd := range cmds {
cmd.setErr(e)
}
}

//------------------------------------------------------------------------------

type baseCmd struct {
Expand Down
3 changes: 1 addition & 2 deletions v2/example_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -108,8 +108,7 @@ func ExampleMulti() {

for {
cmds, err := incr(tx)
if err == redis.Nil {
// Transaction failed. Repeat.
if err == redis.TxFailedErr {
continue
} else if err != nil {
panic(err)
Expand Down
33 changes: 24 additions & 9 deletions v2/multi.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,9 @@ func (c *Multi) Discard() error {
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()) ([]Cmder, error) {
c.cmds = []Cmder{NewStatusCmd("MULTI")}
f()
Expand All @@ -64,62 +67,74 @@ func (c *Multi) Exec(f func()) ([]Cmder, error) {

cn, err := c.conn()
if err != nil {
return nil, err
setCmdsErr(cmds[1:len(cmds)-1], err)
return cmds[1 : len(cmds)-1], err
}

// Synchronize writes and reads to the connection using mutex.
err = c.execCmds(cmds, cn)
err = c.execCmds(cn, cmds)
if err != nil {
c.removeConn(cn)
return nil, err
c.freeConn(cn, err)
return cmds[1 : len(cmds)-1], err
}

c.putConn(cn)
return cmds[1 : len(cmds)-1], nil
}

func (c *Multi) execCmds(cmds []Cmder, cn *conn) error {
func (c *Multi) execCmds(cn *conn, cmds []Cmder) error {
err := c.writeCmd(cn, cmds...)
if err != nil {
setCmdsErr(cmds[1:len(cmds)-1], err)
return err
}

statusCmd := NewStatusCmd()

// Omit last cmduest (EXEC).
// Omit last command (EXEC).
cmdsLen := len(cmds) - 1

// Parse queued replies.
for i := 0; i < cmdsLen; i++ {
_, err = statusCmd.parseReply(cn.rd)
if 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] != '*' {
return fmt.Errorf("redis: expected '*', but got line %q", line)
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' {
return Nil
setCmdsErr(cmds[1:len(cmds)-1], TxFailedErr)
return TxFailedErr
}

var firstCmdErr error

// Parse replies.
// Loop starts from 1 to omit first cmduest (MULTI).
for i := 1; i < cmdsLen; i++ {
cmd := cmds[i]
val, err := cmd.parseReply(cn.rd)
if err != nil {
cmd.setErr(err)
if firstCmdErr == nil {
firstCmdErr = err
}
} else {
cmd.setVal(val)
}
}

return nil
return firstCmdErr
}
5 changes: 4 additions & 1 deletion v2/parser.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,12 @@ const (
stringFloatMap
)

// Represents Redis nil reply.
// Redis nil reply.
var Nil = errors.New("(nil)")

// Redis transaction failed.
var TxFailedErr = errors.New("redis: transaction failed")

var (
errReaderTooSmall = errors.New("redis: reader is too small")
errValNotSet = errors.New("redis: value is not set")
Expand Down
11 changes: 5 additions & 6 deletions v2/pipeline.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ func (c *Pipeline) Discard() error {
return nil
}

// Always returns list of commands and error of the first failed
// Exec always returns list of commands and error of the first failed
// command if any.
func (c *Pipeline) Exec() ([]Cmder, error) {
cmds := c.cmds
Expand All @@ -47,10 +47,11 @@ func (c *Pipeline) Exec() ([]Cmder, error) {

cn, err := c.conn()
if err != nil {
setCmdsErr(cmds, err)
return cmds, err
}

if err := c.execCmds(cmds, cn); err != nil {
if err := c.execCmds(cn, cmds); err != nil {
c.freeConn(cn, err)
return cmds, err
}
Expand All @@ -59,12 +60,10 @@ func (c *Pipeline) Exec() ([]Cmder, error) {
return cmds, nil
}

func (c *Pipeline) execCmds(cmds []Cmder, cn *conn) error {
func (c *Pipeline) execCmds(cn *conn, cmds []Cmder) error {
err := c.writeCmd(cn, cmds...)
if err != nil {
for _, cmd := range cmds {
cmd.setErr(err)
}
setCmdsErr(cmds, err)
return err
}

Expand Down
2 changes: 1 addition & 1 deletion v2/redis.go
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ func (c *baseClient) init(cn *conn, password string, db int64) error {
}

func (c *baseClient) freeConn(cn *conn, err error) {
if err == Nil {
if err == Nil || err == TxFailedErr {
c.putConn(cn)
} else {
c.removeConn(cn)
Expand Down
20 changes: 11 additions & 9 deletions v2/redis_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2643,13 +2643,9 @@ func (t *RedisTest) transactionalIncr(c *C) ([]redis.Cmder, error) {
v, err := strconv.ParseInt(get.Val(), 10, 64)
c.Assert(err, IsNil)

cmds, err := multi.Exec(func() {
return multi.Exec(func() {
multi.Set("key", strconv.FormatInt(v+1, 10))
})
if err == redis.Nil {
return t.transactionalIncr(c)
}
return cmds, err
}

func (t *RedisTest) TestWatchUnwatch(c *C) {
Expand All @@ -2661,10 +2657,16 @@ func (t *RedisTest) TestWatchUnwatch(c *C) {
for i := 0; i < 1000; i++ {
wg.Add(1)
go func() {
cmds, err := t.transactionalIncr(c)
c.Assert(cmds, HasLen, 1)
c.Assert(err, IsNil)
c.Assert(cmds[0].Err(), IsNil)
for {
cmds, err := t.transactionalIncr(c)
if err == redis.TxFailedErr {
continue
}
c.Assert(err, IsNil)
c.Assert(cmds, HasLen, 1)
c.Assert(cmds[0].Err(), IsNil)
break
}
wg.Done()
}()
}
Expand Down

0 comments on commit 39c42d4

Please sign in to comment.