pipeline.go

Documentation: github.com/redis/go-redis/v9

     1  package redis
     2  
     3  import (
     4  	"context"
     5  	"errors"
     6  )
     7  
     8  type pipelineExecer func(context.Context, []Cmder) error
     9  
    10  // Pipeliner is a mechanism to realise Redis Pipeline technique.
    11  //
    12  // Pipelining is a technique to extremely speed up processing by packing
    13  // operations to batches, send them at once to Redis and read a replies in a
    14  // single step.
    15  // See https://redis.io/topics/pipelining
    16  //
    17  // Pay attention, that Pipeline is not a transaction, so you can get unexpected
    18  // results in case of big pipelines and small read/write timeouts.
    19  // Redis client has retransmission logic in case of timeouts, pipeline
    20  // can be retransmitted and commands can be executed more then once.
    21  // To avoid this: it is good idea to use reasonable bigger read/write timeouts
    22  // depends of your batch size and/or use TxPipeline.
    23  type Pipeliner interface {
    24  	StatefulCmdable
    25  
    26  	// Len obtains the number of commands in the pipeline that have not yet been executed.
    27  	Len() int
    28  
    29  	// Do is an API for executing any command.
    30  	// If a certain Redis command is not yet supported, you can use Do to execute it.
    31  	Do(ctx context.Context, args ...interface{}) *Cmd
    32  
    33  	// Process queues the cmd for later execution.
    34  	Process(ctx context.Context, cmd Cmder) error
    35  
    36  	// BatchProcess adds multiple commands to be executed into the pipeline buffer.
    37  	BatchProcess(ctx context.Context, cmd ...Cmder) error
    38  
    39  	// Discard discards all commands in the pipeline buffer that have not yet been executed.
    40  	Discard()
    41  
    42  	// Exec sends all the commands buffered in the pipeline to the redis server.
    43  	Exec(ctx context.Context) ([]Cmder, error)
    44  
    45  	// Cmds returns the list of queued commands.
    46  	Cmds() []Cmder
    47  }
    48  
    49  var _ Pipeliner = (*Pipeline)(nil)
    50  
    51  // Pipeline implements pipelining as described in
    52  // http://redis.io/topics/pipelining.
    53  // Please note: it is not safe for concurrent use by multiple goroutines.
    54  type Pipeline struct {
    55  	cmdable
    56  	statefulCmdable
    57  
    58  	exec pipelineExecer
    59  	cmds []Cmder
    60  }
    61  
    62  func (c *Pipeline) init() {
    63  	c.cmdable = c.Process
    64  	c.statefulCmdable = c.Process
    65  }
    66  
    67  // Len returns the number of queued commands.
    68  func (c *Pipeline) Len() int {
    69  	return len(c.cmds)
    70  }
    71  
    72  // Do queues the custom command for later execution.
    73  func (c *Pipeline) Do(ctx context.Context, args ...interface{}) *Cmd {
    74  	cmd := NewCmd(ctx, args...)
    75  	if len(args) == 0 {
    76  		cmd.SetErr(errors.New("redis: please enter the command to be executed"))
    77  		return cmd
    78  	}
    79  	_ = c.Process(ctx, cmd)
    80  	return cmd
    81  }
    82  
    83  // Process queues the cmd for later execution.
    84  func (c *Pipeline) Process(ctx context.Context, cmd Cmder) error {
    85  	return c.BatchProcess(ctx, cmd)
    86  }
    87  
    88  // BatchProcess queues multiple cmds for later execution.
    89  func (c *Pipeline) BatchProcess(ctx context.Context, cmd ...Cmder) error {
    90  	c.cmds = append(c.cmds, cmd...)
    91  	return nil
    92  }
    93  
    94  // Discard resets the pipeline and discards queued commands.
    95  func (c *Pipeline) Discard() {
    96  	c.cmds = c.cmds[:0]
    97  }
    98  
    99  // Exec executes all previously queued commands using one
   100  // client-server roundtrip.
   101  //
   102  // Exec always returns list of commands and error of the first failed
   103  // command if any.
   104  func (c *Pipeline) Exec(ctx context.Context) ([]Cmder, error) {
   105  	if len(c.cmds) == 0 {
   106  		return nil, nil
   107  	}
   108  
   109  	cmds := c.cmds
   110  	c.cmds = nil
   111  
   112  	return cmds, c.exec(ctx, cmds)
   113  }
   114  
   115  func (c *Pipeline) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
   116  	if err := fn(c); err != nil {
   117  		return nil, err
   118  	}
   119  	return c.Exec(ctx)
   120  }
   121  
   122  func (c *Pipeline) Pipeline() Pipeliner {
   123  	return c
   124  }
   125  
   126  func (c *Pipeline) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
   127  	return c.Pipelined(ctx, fn)
   128  }
   129  
   130  func (c *Pipeline) TxPipeline() Pipeliner {
   131  	return c
   132  }
   133  
   134  func (c *Pipeline) Cmds() []Cmder {
   135  	return c.cmds
   136  }
   137
View as plain text