当前仓库属于关闭状态,部分功能使用受限,详情请查阅 仓库状态说明
1 Star 1 Fork 0

仓库镜像/go-redis-redis
关闭

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
pipeline.go 2.17 KB
一键复制 编辑 原始数据 按行查看 历史
Vladimir Mihailenco 提交于 2015-10-07 17:09 . Refactor reply parser.
package redis
// Pipeline implements pipelining as described in
// http://redis.io/topics/pipelining. It's NOT safe for concurrent use
// by multiple goroutines.
type Pipeline struct {
commandable
client *baseClient
cmds []Cmder
closed bool
}
func (c *Client) Pipeline() *Pipeline {
pipe := &Pipeline{
client: c.baseClient,
cmds: make([]Cmder, 0, 10),
}
pipe.commandable.process = pipe.process
return pipe
}
func (c *Client) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
pipe := c.Pipeline()
if err := fn(pipe); err != nil {
return nil, err
}
cmds, err := pipe.Exec()
pipe.Close()
return cmds, err
}
func (pipe *Pipeline) process(cmd Cmder) {
pipe.cmds = append(pipe.cmds, cmd)
}
// Close closes the pipeline, releasing any open resources.
func (pipe *Pipeline) Close() error {
pipe.Discard()
pipe.closed = true
return nil
}
// Discard resets the pipeline and discards queued commands.
func (pipe *Pipeline) Discard() error {
if pipe.closed {
return errClosed
}
pipe.cmds = pipe.cmds[:0]
return nil
}
// Exec always returns list of commands and error of the first failed
// command if any.
func (pipe *Pipeline) Exec() (cmds []Cmder, retErr error) {
if pipe.closed {
return nil, errClosed
}
if len(pipe.cmds) == 0 {
return pipe.cmds, nil
}
cmds = pipe.cmds
pipe.cmds = make([]Cmder, 0, 10)
failedCmds := cmds
for i := 0; i <= pipe.client.opt.MaxRetries; i++ {
cn, err := pipe.client.conn()
if err != nil {
setCmdsErr(failedCmds, err)
return cmds, err
}
if i > 0 {
resetCmds(failedCmds)
}
failedCmds, err = execCmds(cn, failedCmds)
pipe.client.putConn(cn, err)
if err != nil && retErr == nil {
retErr = err
}
if len(failedCmds) == 0 {
break
}
}
return cmds, retErr
}
func execCmds(cn *conn, cmds []Cmder) ([]Cmder, error) {
if err := cn.writeCmds(cmds...); err != nil {
setCmdsErr(cmds, err)
return cmds, err
}
var firstCmdErr error
var failedCmds []Cmder
for _, cmd := range cmds {
err := cmd.readReply(cn)
if err == nil {
continue
}
if firstCmdErr == nil {
firstCmdErr = err
}
if shouldRetry(err) {
failedCmds = append(failedCmds, cmd)
}
}
return failedCmds, firstCmdErr
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
Go
1
https://gitee.com/github-image/go-redis-redis.git
git@gitee.com:github-image/go-redis-redis.git
github-image
go-redis-redis
go-redis-redis
v3.2.12

搜索帮助