1
0
mirror of https://github.com/redis/go-redis.git synced 2025-09-08 19:52:07 +03:00

Pipeliner expose queued commands (#3496)

* Pipeliner expose queued commands

Signed-off-by: Xiaolong Chen <fukua95@gmail.com>

* add tests and update some comments

Signed-off-by: Xiaolong Chen <fukua95@gmail.com>

---------

Signed-off-by: Xiaolong Chen <fukua95@gmail.com>
This commit is contained in:
cxljs
2025-09-01 22:44:26 +08:00
committed by GitHub
parent 6b9cbe8c54
commit fafec3f3ce
3 changed files with 37 additions and 5 deletions

View File

@@ -253,6 +253,7 @@ var (
_ Cmdable = (*Tx)(nil)
_ Cmdable = (*Ring)(nil)
_ Cmdable = (*ClusterClient)(nil)
_ Cmdable = (*Pipeline)(nil)
)
type cmdable func(ctx context.Context, cmd Cmder) error

View File

@@ -7,7 +7,7 @@ import (
type pipelineExecer func(context.Context, []Cmder) error
// Pipeliner is an mechanism to realise Redis Pipeline technique.
// Pipeliner is a mechanism to realise Redis Pipeline technique.
//
// Pipelining is a technique to extremely speed up processing by packing
// operations to batches, send them at once to Redis and read a replies in a
@@ -23,21 +23,24 @@ type pipelineExecer func(context.Context, []Cmder) error
type Pipeliner interface {
StatefulCmdable
// Len is to obtain the number of commands in the pipeline that have not yet been executed.
// Len obtains the number of commands in the pipeline that have not yet been executed.
Len() int
// Do is an API for executing any command.
// If a certain Redis command is not yet supported, you can use Do to execute it.
Do(ctx context.Context, args ...interface{}) *Cmd
// Process is to put the commands to be executed into the pipeline buffer.
// Process puts the commands to be executed into the pipeline buffer.
Process(ctx context.Context, cmd Cmder) error
// Discard is to discard all commands in the cache that have not yet been executed.
// Discard discards all commands in the pipeline buffer that have not yet been executed.
Discard()
// Exec is to send all the commands buffered in the pipeline to the redis-server.
// Exec sends all the commands buffered in the pipeline to the redis server.
Exec(ctx context.Context) ([]Cmder, error)
// Cmds returns the list of queued commands.
Cmds() []Cmder
}
var _ Pipeliner = (*Pipeline)(nil)
@@ -119,3 +122,7 @@ func (c *Pipeline) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([
func (c *Pipeline) TxPipeline() Pipeliner {
return c
}
func (c *Pipeline) Cmds() []Cmder {
return c.cmds
}

View File

@@ -36,6 +36,30 @@ var _ = Describe("pipelining", func() {
Expect(get.Val()).To(Equal(""))
})
It("exports queued commands", func() {
p := client.Pipeline()
cmds := p.Cmds()
Expect(cmds).To(BeEmpty())
p.Set(ctx, "foo", "bar", 0)
p.Get(ctx, "foo")
cmds = p.Cmds()
Expect(cmds).To(HaveLen(p.Len()))
Expect(cmds[0].Name()).To(Equal("set"))
Expect(cmds[1].Name()).To(Equal("get"))
cmds, err := p.Exec(ctx)
Expect(err).NotTo(HaveOccurred())
Expect(cmds).To(HaveLen(2))
Expect(cmds[0].Name()).To(Equal("set"))
Expect(cmds[0].(*redis.StatusCmd).Val()).To(Equal("OK"))
Expect(cmds[1].Name()).To(Equal("get"))
Expect(cmds[1].(*redis.StringCmd).Val()).To(Equal("bar"))
cmds = p.Cmds()
Expect(cmds).To(BeEmpty())
})
assertPipeline := func() {
It("returns no errors when there are no commands", func() {
_, err := pipe.Exec(ctx)