re
/
redis
forked from mirror/redis
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Implement TxPipeline for Ring

This commit is contained in:
Vladimir Mihailenco 2019-07-19 12:00:11 +03:00
parent fb642fe859
commit 9d9e126af0
2 changed files with 32 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cluster.go
View File

@ -1061,7 +1061,7 @@ func (c *ClusterClient) _processPipeline(ctx context.Context, cmds []Cmder) erro
cn, err := node.Client.getConn(ctx)
if err != nil {
if err == pool.ErrClosed {
c.mapCmdsByNode(cmds, failedCmds)
_ = c.mapCmdsByNode(cmds, failedCmds)
} else {
setCmdsErr(cmds, err)
}
@ -1265,7 +1265,7 @@ func (c *ClusterClient) _processTxPipeline(ctx context.Context, cmds []Cmder) er
cn, err := node.Client.getConn(ctx)
if err != nil {
if err == pool.ErrClosed {
c.mapCmdsByNode(cmds, failedCmds)
_ = c.mapCmdsByNode(cmds, failedCmds)
} else {
setCmdsErr(cmds, err)
}

41
ring.go
View File

@ -586,10 +586,32 @@ func (c *Ring) Pipeline() Pipeliner {
}
func (c *Ring) processPipeline(ctx context.Context, cmds []Cmder) error {
return c.hooks.processPipeline(ctx, cmds, c._processPipeline)
return c.hooks.processPipeline(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
return c.generalProcessPipeline(ctx, cmds, false)
})
}
func (c *Ring) _processPipeline(ctx context.Context, cmds []Cmder) error {
func (c *Ring) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
return c.TxPipeline().Pipelined(fn)
}
func (c *Ring) TxPipeline() Pipeliner {
pipe := Pipeline{
exec: c.processTxPipeline,
}
pipe.init()
return &pipe
}
func (c *Ring) processTxPipeline(ctx context.Context, cmds []Cmder) error {
return c.hooks.processPipeline(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
return c.generalProcessPipeline(ctx, cmds, true)
})
}
func (c *Ring) generalProcessPipeline(
ctx context.Context, cmds []Cmder, tx bool,
) error {
cmdsMap := make(map[string][]Cmder)
for _, cmd := range cmds {
cmdInfo := c.cmdInfo(cmd.Name())
@ -626,7 +648,12 @@ func (c *Ring) _processPipeline(ctx context.Context, cmds []Cmder) error {
return
}
canRetry, err := shard.Client.pipelineProcessCmds(ctx, cn, cmds)
var canRetry bool
if tx {
canRetry, err = shard.Client.txPipelineProcessCmds(ctx, cn, cmds)
} else {
canRetry, err = shard.Client.pipelineProcessCmds(ctx, cn, cmds)
}
shard.Client.releaseConnStrict(cn, err)
if canRetry && internal.IsRetryableError(err, true) {
@ -650,14 +677,6 @@ func (c *Ring) _processPipeline(ctx context.Context, cmds []Cmder) error {
return cmdsFirstErr(cmds)
}
func (c *Ring) TxPipeline() Pipeliner {
panic("not implemented")
}
func (c *Ring) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
panic("not implemented")
}
// Close closes the ring client, releasing any open resources.
//
// It is rare to Close a Ring, as the Ring is meant to be long-lived