forked from mirror/redis
Embed Cmdable into StatefulCmdable
This commit is contained in:
parent
368f0ea0ba
commit
7e8890b644
|
@ -349,7 +349,7 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
|
||||||
opt: opt,
|
opt: opt,
|
||||||
nodes: newClusterNodes(opt),
|
nodes: newClusterNodes(opt),
|
||||||
}
|
}
|
||||||
c.cmdable.process = c.Process
|
c.setProcessor(c.Process)
|
||||||
|
|
||||||
// Add initial nodes.
|
// Add initial nodes.
|
||||||
for _, addr := range opt.Addrs {
|
for _, addr := range opt.Addrs {
|
||||||
|
@ -678,8 +678,7 @@ func (c *ClusterClient) Pipeline() Pipeliner {
|
||||||
pipe := Pipeline{
|
pipe := Pipeline{
|
||||||
exec: c.pipelineExec,
|
exec: c.pipelineExec,
|
||||||
}
|
}
|
||||||
pipe.cmdable.process = pipe.Process
|
pipe.setProcessor(pipe.Process)
|
||||||
pipe.statefulCmdable.process = pipe.Process
|
|
||||||
return &pipe
|
return &pipe
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -801,8 +800,7 @@ func (c *ClusterClient) TxPipeline() Pipeliner {
|
||||||
pipe := Pipeline{
|
pipe := Pipeline{
|
||||||
exec: c.txPipelineExec,
|
exec: c.txPipelineExec,
|
||||||
}
|
}
|
||||||
pipe.cmdable.process = pipe.Process
|
pipe.setProcessor(pipe.Process)
|
||||||
pipe.statefulCmdable.process = pipe.Process
|
|
||||||
return &pipe
|
return &pipe
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
12
commands.go
12
commands.go
|
@ -238,6 +238,7 @@ type Cmdable interface {
|
||||||
}
|
}
|
||||||
|
|
||||||
type StatefulCmdable interface {
|
type StatefulCmdable interface {
|
||||||
|
Cmdable
|
||||||
Auth(password string) *StatusCmd
|
Auth(password string) *StatusCmd
|
||||||
Select(index int) *StatusCmd
|
Select(index int) *StatusCmd
|
||||||
ClientSetName(name string) *BoolCmd
|
ClientSetName(name string) *BoolCmd
|
||||||
|
@ -255,10 +256,20 @@ type cmdable struct {
|
||||||
process func(cmd Cmder) error
|
process func(cmd Cmder) error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (c *cmdable) setProcessor(fn func(Cmder) error) {
|
||||||
|
c.process = fn
|
||||||
|
}
|
||||||
|
|
||||||
type statefulCmdable struct {
|
type statefulCmdable struct {
|
||||||
|
cmdable
|
||||||
process func(cmd Cmder) error
|
process func(cmd Cmder) error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (c *statefulCmdable) setProcessor(fn func(Cmder) error) {
|
||||||
|
c.process = fn
|
||||||
|
c.cmdable.setProcessor(fn)
|
||||||
|
}
|
||||||
|
|
||||||
//------------------------------------------------------------------------------
|
//------------------------------------------------------------------------------
|
||||||
|
|
||||||
func (c *statefulCmdable) Auth(password string) *StatusCmd {
|
func (c *statefulCmdable) Auth(password string) *StatusCmd {
|
||||||
|
@ -280,7 +291,6 @@ func (c *cmdable) Ping() *StatusCmd {
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *cmdable) Wait(numSlaves int, timeout time.Duration) *IntCmd {
|
func (c *cmdable) Wait(numSlaves int, timeout time.Duration) *IntCmd {
|
||||||
|
|
||||||
cmd := NewIntCmd("wait", numSlaves, int(timeout/time.Millisecond))
|
cmd := NewIntCmd("wait", numSlaves, int(timeout/time.Millisecond))
|
||||||
c.process(cmd)
|
c.process(cmd)
|
||||||
return cmd
|
return cmd
|
||||||
|
|
|
@ -10,7 +10,6 @@ import (
|
||||||
type pipelineExecer func([]Cmder) error
|
type pipelineExecer func([]Cmder) error
|
||||||
|
|
||||||
type Pipeliner interface {
|
type Pipeliner interface {
|
||||||
Cmdable
|
|
||||||
StatefulCmdable
|
StatefulCmdable
|
||||||
Process(cmd Cmder) error
|
Process(cmd Cmder) error
|
||||||
Close() error
|
Close() error
|
||||||
|
@ -26,7 +25,6 @@ var _ Pipeliner = (*Pipeline)(nil)
|
||||||
// http://redis.io/topics/pipelining. It's safe for concurrent use
|
// http://redis.io/topics/pipelining. It's safe for concurrent use
|
||||||
// by multiple goroutines.
|
// by multiple goroutines.
|
||||||
type Pipeline struct {
|
type Pipeline struct {
|
||||||
cmdable
|
|
||||||
statefulCmdable
|
statefulCmdable
|
||||||
|
|
||||||
exec pipelineExecer
|
exec pipelineExecer
|
||||||
|
|
10
redis.go
10
redis.go
|
@ -294,7 +294,7 @@ func newClient(opt *Options, pool pool.Pooler) *Client {
|
||||||
connPool: pool,
|
connPool: pool,
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
client.cmdable.process = client.Process
|
client.setProcessor(client.Process)
|
||||||
return &client
|
return &client
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -307,7 +307,7 @@ func NewClient(opt *Options) *Client {
|
||||||
func (c *Client) copy() *Client {
|
func (c *Client) copy() *Client {
|
||||||
c2 := new(Client)
|
c2 := new(Client)
|
||||||
*c2 = *c
|
*c2 = *c
|
||||||
c2.cmdable.process = c2.Process
|
c2.setProcessor(c2.Process)
|
||||||
return c2
|
return c2
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -332,8 +332,7 @@ func (c *Client) Pipeline() Pipeliner {
|
||||||
pipe := Pipeline{
|
pipe := Pipeline{
|
||||||
exec: c.pipelineExecer(c.pipelineProcessCmds),
|
exec: c.pipelineExecer(c.pipelineProcessCmds),
|
||||||
}
|
}
|
||||||
pipe.cmdable.process = pipe.Process
|
pipe.setProcessor(pipe.Process)
|
||||||
pipe.statefulCmdable.process = pipe.Process
|
|
||||||
return &pipe
|
return &pipe
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -346,8 +345,7 @@ func (c *Client) TxPipeline() Pipeliner {
|
||||||
pipe := Pipeline{
|
pipe := Pipeline{
|
||||||
exec: c.pipelineExecer(c.txPipelineProcessCmds),
|
exec: c.pipelineExecer(c.txPipelineProcessCmds),
|
||||||
}
|
}
|
||||||
pipe.cmdable.process = pipe.Process
|
pipe.setProcessor(pipe.Process)
|
||||||
pipe.statefulCmdable.process = pipe.Process
|
|
||||||
return &pipe
|
return &pipe
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
5
ring.go
5
ring.go
|
@ -148,7 +148,7 @@ func NewRing(opt *RingOptions) *Ring {
|
||||||
|
|
||||||
cmdsInfoOnce: new(sync.Once),
|
cmdsInfoOnce: new(sync.Once),
|
||||||
}
|
}
|
||||||
ring.cmdable.process = ring.Process
|
ring.setProcessor(ring.Process)
|
||||||
for name, addr := range opt.Addrs {
|
for name, addr := range opt.Addrs {
|
||||||
clopt := opt.clientOptions()
|
clopt := opt.clientOptions()
|
||||||
clopt.Addr = addr
|
clopt.Addr = addr
|
||||||
|
@ -385,8 +385,7 @@ func (c *Ring) Pipeline() Pipeliner {
|
||||||
pipe := Pipeline{
|
pipe := Pipeline{
|
||||||
exec: c.pipelineExec,
|
exec: c.pipelineExec,
|
||||||
}
|
}
|
||||||
pipe.cmdable.process = pipe.Process
|
pipe.setProcessor(pipe.Process)
|
||||||
pipe.statefulCmdable.process = pipe.Process
|
|
||||||
return &pipe
|
return &pipe
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -82,7 +82,7 @@ func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
client.cmdable.process = client.Process
|
client.setProcessor(client.Process)
|
||||||
|
|
||||||
return &client
|
return &client
|
||||||
}
|
}
|
||||||
|
|
7
tx.go
7
tx.go
|
@ -13,7 +13,6 @@ const TxFailedErr = internal.RedisError("redis: transaction failed")
|
||||||
// by multiple goroutines, because Exec resets list of watched keys.
|
// by multiple goroutines, because Exec resets list of watched keys.
|
||||||
// If you don't need WATCH it is better to use Pipeline.
|
// If you don't need WATCH it is better to use Pipeline.
|
||||||
type Tx struct {
|
type Tx struct {
|
||||||
cmdable
|
|
||||||
statefulCmdable
|
statefulCmdable
|
||||||
baseClient
|
baseClient
|
||||||
}
|
}
|
||||||
|
@ -25,8 +24,7 @@ func (c *Client) newTx() *Tx {
|
||||||
connPool: pool.NewStickyConnPool(c.connPool.(*pool.ConnPool), true),
|
connPool: pool.NewStickyConnPool(c.connPool.(*pool.ConnPool), true),
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
tx.cmdable.process = tx.Process
|
tx.setProcessor(tx.Process)
|
||||||
tx.statefulCmdable.process = tx.Process
|
|
||||||
return &tx
|
return &tx
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -80,8 +78,7 @@ func (c *Tx) Pipeline() Pipeliner {
|
||||||
pipe := Pipeline{
|
pipe := Pipeline{
|
||||||
exec: c.pipelineExecer(c.txPipelineProcessCmds),
|
exec: c.pipelineExecer(c.txPipelineProcessCmds),
|
||||||
}
|
}
|
||||||
pipe.cmdable.process = pipe.Process
|
pipe.setProcessor(pipe.Process)
|
||||||
pipe.statefulCmdable.process = pipe.Process
|
|
||||||
return &pipe
|
return &pipe
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue