redis/redis.go

763 lines
16 KiB
Go
Raw Normal View History

2017-02-18 17:42:34 +03:00
package redis
2012-07-25 17:00:50 +04:00
import (
"context"
2015-05-15 15:21:28 +03:00
"fmt"
2014-07-31 17:01:54 +04:00
"log"
"time"
2020-03-11 17:29:16 +03:00
"github.com/go-redis/redis/v8/internal"
"github.com/go-redis/redis/v8/internal/pool"
"github.com/go-redis/redis/v8/internal/proto"
2012-08-09 14:12:41 +04:00
)
2012-07-25 17:00:50 +04:00
2019-08-09 16:23:56 +03:00
// Nil reply returned by Redis when key does not exist.
2018-02-22 15:14:30 +03:00
const Nil = proto.Nil
2016-07-02 15:52:10 +03:00
func SetLogger(logger *log.Logger) {
2016-04-09 14:52:01 +03:00
internal.Logger = logger
}
//------------------------------------------------------------------------------
type Hook interface {
BeforeProcess(ctx context.Context, cmd Cmder) (context.Context, error)
AfterProcess(ctx context.Context, cmd Cmder) error
BeforeProcessPipeline(ctx context.Context, cmds []Cmder) (context.Context, error)
AfterProcessPipeline(ctx context.Context, cmds []Cmder) error
}
type hooks struct {
hooks []Hook
}
func (hs *hooks) lock() {
2019-08-24 12:22:52 +03:00
hs.hooks = hs.hooks[:len(hs.hooks):len(hs.hooks)]
}
func (hs hooks) clone() hooks {
2019-12-29 13:06:43 +03:00
clone := hs
clone.lock()
2019-12-29 13:06:43 +03:00
return clone
}
2019-05-31 17:03:20 +03:00
func (hs *hooks) AddHook(hook Hook) {
hs.hooks = append(hs.hooks, hook)
}
2019-06-04 13:30:47 +03:00
func (hs hooks) process(
ctx context.Context, cmd Cmder, fn func(context.Context, Cmder) error,
) error {
ctx, err := hs.beforeProcess(ctx, cmd)
if err != nil {
2020-02-03 12:53:47 +03:00
cmd.SetErr(err)
return err
}
2019-06-04 13:30:47 +03:00
cmdErr := fn(ctx, cmd)
if err := hs.afterProcess(ctx, cmd); err != nil {
2020-02-03 12:53:47 +03:00
cmd.SetErr(err)
return err
}
return cmdErr
}
func (hs hooks) beforeProcess(ctx context.Context, cmd Cmder) (context.Context, error) {
for _, h := range hs.hooks {
var err error
ctx, err = h.BeforeProcess(ctx, cmd)
if err != nil {
return nil, err
}
}
return ctx, nil
}
2019-08-24 11:55:13 +03:00
func (hs hooks) afterProcess(ctx context.Context, cmd Cmder) error {
var firstErr error
for _, h := range hs.hooks {
err := h.AfterProcess(ctx, cmd)
if err != nil && firstErr == nil {
firstErr = err
}
}
2019-08-24 11:55:13 +03:00
return firstErr
}
2019-06-04 13:30:47 +03:00
func (hs hooks) processPipeline(
ctx context.Context, cmds []Cmder, fn func(context.Context, []Cmder) error,
) error {
ctx, err := hs.beforeProcessPipeline(ctx, cmds)
if err != nil {
setCmdsErr(cmds, err)
return err
}
2019-06-04 13:30:47 +03:00
cmdsErr := fn(ctx, cmds)
if err := hs.afterProcessPipeline(ctx, cmds); err != nil {
setCmdsErr(cmds, err)
return err
}
return cmdsErr
}
func (hs hooks) beforeProcessPipeline(ctx context.Context, cmds []Cmder) (context.Context, error) {
for _, h := range hs.hooks {
var err error
ctx, err = h.BeforeProcessPipeline(ctx, cmds)
if err != nil {
return nil, err
}
}
return ctx, nil
}
2019-08-24 11:55:13 +03:00
func (hs hooks) afterProcessPipeline(ctx context.Context, cmds []Cmder) error {
var firstErr error
for _, h := range hs.hooks {
err := h.AfterProcessPipeline(ctx, cmds)
if err != nil && firstErr == nil {
firstErr = err
}
}
2019-08-24 11:55:13 +03:00
return firstErr
}
2020-02-14 16:37:35 +03:00
func (hs hooks) processTxPipeline(
ctx context.Context, cmds []Cmder, fn func(context.Context, []Cmder) error,
) error {
2020-03-11 17:26:42 +03:00
cmds = wrapMultiExec(ctx, cmds)
2020-02-14 16:37:35 +03:00
return hs.processPipeline(ctx, cmds, fn)
}
//------------------------------------------------------------------------------
type baseClient struct {
opt *Options
2018-03-07 14:50:14 +03:00
connPool pool.Pooler
onClose func() error // hook called when client is closed
}
2020-02-02 15:59:27 +03:00
func newBaseClient(opt *Options, connPool pool.Pooler) *baseClient {
return &baseClient{
opt: opt,
connPool: connPool,
}
}
func (c *baseClient) clone() *baseClient {
clone := *c
return &clone
}
func (c *baseClient) withTimeout(timeout time.Duration) *baseClient {
opt := c.opt.clone()
opt.ReadTimeout = timeout
opt.WriteTimeout = timeout
clone := c.clone()
clone.opt = opt
return clone
}
2015-05-15 15:21:28 +03:00
func (c *baseClient) String() string {
2016-10-05 23:20:05 +03:00
return fmt.Sprintf("Redis<%s db:%d>", c.getAddr(), c.opt.DB)
2015-05-15 15:21:28 +03:00
}
2019-06-14 16:00:03 +03:00
func (c *baseClient) newConn(ctx context.Context) (*pool.Conn, error) {
cn, err := c.connPool.NewConn(ctx)
2017-07-09 10:07:20 +03:00
if err != nil {
return nil, err
}
2019-07-25 13:28:15 +03:00
err = c.initConn(ctx, cn)
2019-03-25 14:02:31 +03:00
if err != nil {
_ = c.connPool.CloseConn(cn)
return nil, err
2017-07-09 10:07:20 +03:00
}
return cn, nil
}
2019-06-04 14:05:29 +03:00
func (c *baseClient) getConn(ctx context.Context) (*pool.Conn, error) {
2020-02-02 12:09:27 +03:00
if c.opt.Limiter != nil {
err := c.opt.Limiter.Allow()
2018-10-14 10:53:48 +03:00
if err != nil {
return nil, err
}
}
2019-06-04 14:05:29 +03:00
cn, err := c._getConn(ctx)
2018-10-14 10:53:48 +03:00
if err != nil {
2020-02-02 12:09:27 +03:00
if c.opt.Limiter != nil {
c.opt.Limiter.ReportResult(err)
2018-10-14 10:53:48 +03:00
}
return nil, err
}
2020-03-11 17:26:42 +03:00
2018-10-14 10:53:48 +03:00
return cn, nil
}
2019-06-04 14:05:29 +03:00
func (c *baseClient) _getConn(ctx context.Context) (*pool.Conn, error) {
cn, err := c.connPool.Get(ctx)
2016-03-15 15:04:35 +03:00
if err != nil {
2018-05-28 17:27:24 +03:00
return nil, err
2016-03-15 15:04:35 +03:00
}
2020-03-11 17:26:42 +03:00
if cn.Inited {
return cn, nil
}
err = internal.WithSpan(ctx, "init_conn", func(ctx context.Context) error {
return c.initConn(ctx, cn)
})
2019-03-25 14:02:31 +03:00
if err != nil {
2019-08-08 10:36:13 +03:00
c.connPool.Remove(cn, err)
if err := internal.Unwrap(err); err != nil {
return nil, err
}
2019-03-25 14:02:31 +03:00
return nil, err
2016-03-12 13:41:02 +03:00
}
2017-04-18 13:12:38 +03:00
2018-05-28 17:27:24 +03:00
return cn, nil
2012-07-26 19:16:17 +04:00
}
2019-07-25 13:28:15 +03:00
func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
2019-03-25 14:02:31 +03:00
if cn.Inited {
return nil
}
cn.Inited = true
2016-03-15 15:04:35 +03:00
2017-05-25 14:16:39 +03:00
if c.opt.Password == "" &&
c.opt.DB == 0 &&
!c.opt.readOnly &&
2017-05-25 14:16:39 +03:00
c.opt.OnConnect == nil {
2016-03-12 13:41:02 +03:00
return nil
}
connPool := pool.NewSingleConnPool(nil)
connPool.SetConn(cn)
conn := newConn(ctx, c.opt, connPool)
2020-03-11 17:26:42 +03:00
_, err := conn.Pipelined(ctx, func(pipe Pipeliner) error {
if c.opt.Password != "" {
2020-03-11 17:26:42 +03:00
pipe.Auth(ctx, c.opt.Password)
2016-03-12 13:41:02 +03:00
}
if c.opt.DB > 0 {
2020-03-11 17:26:42 +03:00
pipe.Select(ctx, c.opt.DB)
2016-03-12 13:41:02 +03:00
}
if c.opt.readOnly {
2020-03-11 17:26:42 +03:00
pipe.ReadOnly(ctx)
}
return nil
})
2017-05-25 14:16:39 +03:00
if err != nil {
return err
}
if c.opt.OnConnect != nil {
return c.opt.OnConnect(conn)
}
return nil
2016-03-12 13:41:02 +03:00
}
2019-08-07 16:57:48 +03:00
func (c *baseClient) releaseConn(cn *pool.Conn, err error) {
2020-02-02 12:09:27 +03:00
if c.opt.Limiter != nil {
c.opt.Limiter.ReportResult(err)
2019-08-07 16:57:48 +03:00
}
if isBadConn(err, false) {
2019-08-08 10:36:13 +03:00
c.connPool.Remove(cn, err)
2019-08-07 16:57:48 +03:00
} else {
c.connPool.Put(cn)
}
}
func (c *baseClient) withConn(
ctx context.Context, fn func(context.Context, *pool.Conn) error,
) error {
2020-03-11 17:26:42 +03:00
return internal.WithSpan(ctx, "with_conn", func(ctx context.Context) error {
cn, err := c.getConn(ctx)
if err != nil {
return err
}
defer func() {
c.releaseConn(cn, err)
}()
2020-03-11 17:26:42 +03:00
err = fn(ctx, cn)
return err
})
}
2019-06-04 13:30:47 +03:00
func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
err := c._process(ctx, cmd)
if err != nil {
2020-02-03 12:53:47 +03:00
cmd.SetErr(err)
return err
}
return nil
}
func (c *baseClient) _process(ctx context.Context, cmd Cmder) error {
var lastErr error
2017-07-09 13:10:07 +03:00
for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ {
2020-03-11 17:26:42 +03:00
attempt := attempt
var retry bool
err := internal.WithSpan(ctx, "process", func(ctx context.Context) error {
if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
return err
}
2019-07-30 12:13:00 +03:00
}
2017-05-25 08:08:44 +03:00
2020-03-11 17:26:42 +03:00
retryTimeout := true
err := c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmd(wr, cmd)
})
if err != nil {
return err
}
err = cn.WithReader(ctx, c.cmdTimeout(cmd), cmd.readReply)
if err != nil {
retryTimeout = cmd.readTimeout() == nil
return err
}
return nil
})
2020-03-11 17:26:42 +03:00
if err == nil {
return nil
}
2020-03-11 17:26:42 +03:00
retry = isRetryableError(err, retryTimeout)
return err
})
2020-03-11 17:26:42 +03:00
if err == nil || !retry {
return err
}
2020-03-11 17:26:42 +03:00
lastErr = err
2012-07-26 19:16:17 +04:00
}
return lastErr
2014-05-11 11:42:40 +04:00
}
2017-07-09 13:10:07 +03:00
func (c *baseClient) retryBackoff(attempt int) time.Duration {
return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff)
}
func (c *baseClient) cmdTimeout(cmd Cmder) time.Duration {
if timeout := cmd.readTimeout(); timeout != nil {
2018-09-13 09:14:52 +03:00
t := *timeout
if t == 0 {
return 0
}
return t + 10*time.Second
}
2018-01-24 21:38:47 +03:00
return c.opt.ReadTimeout
}
2014-05-11 11:42:40 +04:00
// Close closes the client, releasing any open resources.
2015-09-12 09:36:03 +03:00
//
// It is rare to Close a Client, as the Client is meant to be
// long-lived and shared between many goroutines.
2014-05-11 11:42:40 +04:00
func (c *baseClient) Close() error {
var firstErr error
if c.onClose != nil {
if err := c.onClose(); err != nil {
firstErr = err
}
}
if err := c.connPool.Close(); err != nil && firstErr == nil {
firstErr = err
}
return firstErr
2012-07-26 19:16:17 +04:00
}
2016-10-05 23:20:05 +03:00
func (c *baseClient) getAddr() string {
return c.opt.Addr
}
2019-06-04 13:30:47 +03:00
func (c *baseClient) processPipeline(ctx context.Context, cmds []Cmder) error {
2019-06-04 14:05:29 +03:00
return c.generalProcessPipeline(ctx, cmds, c.pipelineProcessCmds)
2018-01-20 13:26:33 +03:00
}
2019-06-04 13:30:47 +03:00
func (c *baseClient) processTxPipeline(ctx context.Context, cmds []Cmder) error {
2019-06-04 14:05:29 +03:00
return c.generalProcessPipeline(ctx, cmds, c.txPipelineProcessCmds)
2018-01-20 13:26:33 +03:00
}
type pipelineProcessor func(context.Context, *pool.Conn, []Cmder) (bool, error)
2016-12-13 18:28:39 +03:00
2019-06-04 14:05:29 +03:00
func (c *baseClient) generalProcessPipeline(
ctx context.Context, cmds []Cmder, p pipelineProcessor,
) error {
err := c._generalProcessPipeline(ctx, cmds, p)
if err != nil {
setCmdsErr(cmds, err)
return err
}
return cmdsFirstErr(cmds)
}
func (c *baseClient) _generalProcessPipeline(
ctx context.Context, cmds []Cmder, p pipelineProcessor,
) error {
var lastErr error
2018-01-20 13:26:33 +03:00
for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ {
if attempt > 0 {
2019-07-30 12:13:00 +03:00
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
return err
}
2018-01-20 13:26:33 +03:00
}
2017-08-31 15:22:47 +03:00
var canRetry bool
lastErr = c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
var err error
canRetry, err = p(ctx, cn, cmds)
2018-01-20 13:26:33 +03:00
return err
})
if lastErr == nil || !canRetry || !isRetryableError(lastErr, true) {
return lastErr
2016-12-13 18:28:39 +03:00
}
}
return lastErr
2016-12-13 18:28:39 +03:00
}
func (c *baseClient) pipelineProcessCmds(
ctx context.Context, cn *pool.Conn, cmds []Cmder,
) (bool, error) {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmds(wr, cmds)
2018-08-15 11:53:15 +03:00
})
if err != nil {
2016-12-13 18:28:39 +03:00
return true, err
}
err = cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
2018-08-15 11:53:15 +03:00
return pipelineReadCmds(rd, cmds)
})
return true, err
2016-12-13 18:28:39 +03:00
}
2018-08-17 13:56:37 +03:00
func pipelineReadCmds(rd *proto.Reader, cmds []Cmder) error {
2017-08-31 15:22:47 +03:00
for _, cmd := range cmds {
2018-08-15 09:38:58 +03:00
err := cmd.readReply(rd)
if err != nil && !isRedisError(err) {
2017-08-31 15:22:47 +03:00
return err
2016-12-13 18:28:39 +03:00
}
}
2017-08-31 15:22:47 +03:00
return nil
2016-12-13 18:28:39 +03:00
}
func (c *baseClient) txPipelineProcessCmds(
ctx context.Context, cn *pool.Conn, cmds []Cmder,
) (bool, error) {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
2020-02-14 16:37:35 +03:00
return writeCmds(wr, cmds)
2018-08-15 11:53:15 +03:00
})
2018-08-15 09:38:58 +03:00
if err != nil {
2016-12-13 18:28:39 +03:00
return true, err
}
err = cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
2020-02-14 16:37:35 +03:00
statusCmd := cmds[0].(*StatusCmd)
// Trim multi and exec.
cmds = cmds[1 : len(cmds)-1]
err := txPipelineReadQueued(rd, statusCmd, cmds)
2018-08-15 11:53:15 +03:00
if err != nil {
return err
}
2020-02-14 16:37:35 +03:00
2018-08-15 11:53:15 +03:00
return pipelineReadCmds(rd, cmds)
})
return false, err
2016-12-13 18:28:39 +03:00
}
2020-03-11 17:26:42 +03:00
func wrapMultiExec(ctx context.Context, cmds []Cmder) []Cmder {
2020-02-14 16:37:35 +03:00
if len(cmds) == 0 {
panic("not reached")
}
2020-02-14 16:37:35 +03:00
cmds = append(cmds, make([]Cmder, 2)...)
copy(cmds[1:], cmds[:len(cmds)-2])
2020-03-11 17:26:42 +03:00
cmds[0] = NewStatusCmd(ctx, "multi")
cmds[len(cmds)-1] = NewSliceCmd(ctx, "exec")
2020-02-14 16:37:35 +03:00
return cmds
2016-12-13 18:28:39 +03:00
}
2020-02-14 16:37:35 +03:00
func txPipelineReadQueued(rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder) error {
2016-12-13 18:28:39 +03:00
// Parse queued replies.
2020-02-14 16:37:35 +03:00
if err := statusCmd.readReply(rd); err != nil {
2017-08-31 15:22:47 +03:00
return err
2016-12-13 18:28:39 +03:00
}
2018-10-11 13:53:40 +03:00
for range cmds {
2020-02-14 16:37:35 +03:00
if err := statusCmd.readReply(rd); err != nil && !isRedisError(err) {
2017-08-31 15:22:47 +03:00
return err
2016-12-13 18:28:39 +03:00
}
}
// Parse number of replies.
2018-08-15 09:38:58 +03:00
line, err := rd.ReadLine()
2016-12-13 18:28:39 +03:00
if err != nil {
if err == Nil {
err = TxFailedErr
}
return err
}
switch line[0] {
case proto.ErrorReply:
return proto.ParseErrorReply(line)
case proto.ArrayReply:
// ok
default:
err := fmt.Errorf("redis: expected '*', but got line %q", line)
return err
}
return nil
}
2014-05-11 11:42:40 +04:00
//------------------------------------------------------------------------------
2015-09-12 09:36:03 +03:00
// Client is a Redis client representing a pool of zero or more
// underlying connections. It's safe for concurrent use by multiple
// goroutines.
type Client struct {
2020-02-02 15:59:27 +03:00
*baseClient
2019-08-24 12:22:52 +03:00
cmdable
hooks
2018-12-13 11:31:02 +03:00
ctx context.Context
2015-01-24 15:12:48 +03:00
}
// NewClient returns a client to the Redis Server specified by Options.
func NewClient(opt *Options) *Client {
opt.init()
2018-01-20 13:26:33 +03:00
c := Client{
2020-02-02 15:59:27 +03:00
baseClient: newBaseClient(opt, newConnPool(opt)),
ctx: context.Background(),
2015-01-24 15:12:48 +03:00
}
2019-08-24 12:22:52 +03:00
c.cmdable = c.Process
2018-01-20 13:26:33 +03:00
return &c
2012-08-05 16:09:43 +04:00
}
2020-02-02 15:59:27 +03:00
func (c *Client) clone() *Client {
clone := *c
clone.cmdable = clone.Process
clone.hooks.lock()
2020-02-02 15:59:27 +03:00
return &clone
}
func (c *Client) WithTimeout(timeout time.Duration) *Client {
clone := c.clone()
clone.baseClient = c.baseClient.withTimeout(timeout)
return clone
}
func (c *Client) Context() context.Context {
2019-07-04 11:18:06 +03:00
return c.ctx
}
func (c *Client) WithContext(ctx context.Context) *Client {
if ctx == nil {
panic("nil context")
}
2020-02-02 15:59:27 +03:00
clone := c.clone()
2019-05-31 17:03:20 +03:00
clone.ctx = ctx
2020-02-02 15:59:27 +03:00
return clone
}
2020-03-11 17:26:42 +03:00
func (c *Client) Conn(ctx context.Context) *Conn {
return newConn(ctx, c.opt, pool.NewSingleConnPool(c.connPool))
}
2019-06-04 13:30:47 +03:00
// Do creates a Cmd from the args and processes the cmd.
2020-03-11 17:26:42 +03:00
func (c *Client) Do(ctx context.Context, args ...interface{}) *Cmd {
cmd := NewCmd(ctx, args...)
_ = c.Process(ctx, cmd)
2019-06-04 13:30:47 +03:00
return cmd
}
2020-03-11 17:26:42 +03:00
func (c *Client) Process(ctx context.Context, cmd Cmder) error {
2019-06-04 13:30:47 +03:00
return c.hooks.process(ctx, cmd, c.baseClient.process)
}
2019-06-04 13:30:47 +03:00
func (c *Client) processPipeline(ctx context.Context, cmds []Cmder) error {
return c.hooks.processPipeline(ctx, cmds, c.baseClient.processPipeline)
}
func (c *Client) processTxPipeline(ctx context.Context, cmds []Cmder) error {
2020-02-14 16:37:35 +03:00
return c.hooks.processTxPipeline(ctx, cmds, c.baseClient.processTxPipeline)
2017-01-11 06:32:10 +03:00
}
2017-05-25 14:16:39 +03:00
// Options returns read-only Options that were used to create the client.
func (c *Client) Options() *Options {
return c.opt
}
2017-09-11 10:12:00 +03:00
type PoolStats pool.Stats
// PoolStats returns connection pool stats.
2016-01-19 19:36:40 +03:00
func (c *Client) PoolStats() *PoolStats {
2017-09-11 10:12:00 +03:00
stats := c.connPool.Stats()
return (*PoolStats)(stats)
2016-01-19 19:36:40 +03:00
}
2020-03-11 17:26:42 +03:00
func (c *Client) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipeline().Pipelined(ctx, fn)
2016-12-13 18:28:39 +03:00
}
2017-05-02 18:00:53 +03:00
func (c *Client) Pipeline() Pipeliner {
pipe := Pipeline{
2019-07-25 13:28:15 +03:00
ctx: c.ctx,
2018-01-20 13:26:33 +03:00
exec: c.processPipeline,
}
2019-05-31 17:03:20 +03:00
pipe.init()
return &pipe
}
2020-03-11 17:26:42 +03:00
func (c *Client) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
return c.TxPipeline().Pipelined(ctx, fn)
}
2016-12-16 15:19:53 +03:00
// TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
func (c *Client) TxPipeline() Pipeliner {
2016-12-13 18:28:39 +03:00
pipe := Pipeline{
2019-07-25 13:28:15 +03:00
ctx: c.ctx,
2018-01-20 13:26:33 +03:00
exec: c.processTxPipeline,
}
2019-05-31 17:03:20 +03:00
pipe.init()
2016-12-13 18:28:39 +03:00
return &pipe
}
func (c *Client) pubSub() *PubSub {
2018-07-23 15:55:13 +03:00
pubsub := &PubSub{
2017-07-09 10:07:20 +03:00
opt: c.opt,
2020-03-11 17:26:42 +03:00
newConn: func(ctx context.Context, channels []string) (*pool.Conn, error) {
return c.newConn(ctx)
},
2017-07-09 10:07:20 +03:00
closeConn: c.connPool.CloseConn,
}
2018-07-23 15:55:13 +03:00
pubsub.init()
return pubsub
}
// Subscribe subscribes the client to the specified channels.
2017-06-17 12:43:19 +03:00
// Channels can be omitted to create empty subscription.
// Note that this method does not wait on a response from Redis, so the
// subscription may not be active immediately. To force the connection to wait,
// you may call the Receive() method on the returned *PubSub like so:
//
// sub := client.Subscribe(queryResp)
// iface, err := sub.Receive()
// if err != nil {
// // handle error
// }
2018-12-13 11:31:02 +03:00
//
// // Should be *Subscription, but others are possible if other actions have been
// // taken on sub since it was created.
// switch iface.(type) {
// case *Subscription:
// // subscribe succeeded
// case *Message:
// // received first message
// case *Pong:
// // pong received
// default:
// // handle error
// }
//
// ch := sub.Channel()
2020-03-11 17:26:42 +03:00
func (c *Client) Subscribe(ctx context.Context, channels ...string) *PubSub {
pubsub := c.pubSub()
if len(channels) > 0 {
2020-03-11 17:26:42 +03:00
_ = pubsub.Subscribe(ctx, channels...)
}
2017-04-11 16:53:55 +03:00
return pubsub
}
// PSubscribe subscribes the client to the given patterns.
2017-06-17 12:43:19 +03:00
// Patterns can be omitted to create empty subscription.
2020-03-11 17:26:42 +03:00
func (c *Client) PSubscribe(ctx context.Context, channels ...string) *PubSub {
pubsub := c.pubSub()
if len(channels) > 0 {
2020-03-11 17:26:42 +03:00
_ = pubsub.PSubscribe(ctx, channels...)
}
2017-04-11 16:53:55 +03:00
return pubsub
}
2017-05-25 14:16:39 +03:00
//------------------------------------------------------------------------------
2019-07-25 13:28:15 +03:00
type conn struct {
2017-05-25 14:16:39 +03:00
baseClient
2019-05-31 17:03:20 +03:00
cmdable
2017-05-25 14:16:39 +03:00
statefulCmdable
}
2019-07-25 13:28:15 +03:00
// Conn is like Client, but its pool contains single connection.
type Conn struct {
*conn
ctx context.Context
}
func newConn(ctx context.Context, opt *Options, connPool pool.Pooler) *Conn {
c := Conn{
2019-07-25 13:28:15 +03:00
conn: &conn{
baseClient: baseClient{
opt: opt,
connPool: connPool,
2019-07-25 13:28:15 +03:00
},
},
2019-07-25 13:28:15 +03:00
ctx: ctx,
}
2019-05-31 17:03:20 +03:00
c.cmdable = c.Process
c.statefulCmdable = c.Process
return &c
}
2020-03-11 17:26:42 +03:00
func (c *Conn) Process(ctx context.Context, cmd Cmder) error {
2019-06-04 13:30:47 +03:00
return c.baseClient.process(ctx, cmd)
}
2020-03-11 17:26:42 +03:00
func (c *Conn) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipeline().Pipelined(ctx, fn)
2017-05-25 14:16:39 +03:00
}
func (c *Conn) Pipeline() Pipeliner {
pipe := Pipeline{
2019-07-25 13:28:15 +03:00
ctx: c.ctx,
2018-01-20 13:26:33 +03:00
exec: c.processPipeline,
2017-05-25 14:16:39 +03:00
}
2019-05-31 17:03:20 +03:00
pipe.init()
2017-05-25 14:16:39 +03:00
return &pipe
}
2020-03-11 17:26:42 +03:00
func (c *Conn) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
return c.TxPipeline().Pipelined(ctx, fn)
2017-05-25 14:16:39 +03:00
}
// TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
func (c *Conn) TxPipeline() Pipeliner {
pipe := Pipeline{
2019-07-25 13:28:15 +03:00
ctx: c.ctx,
2018-01-20 13:26:33 +03:00
exec: c.processTxPipeline,
2017-05-25 14:16:39 +03:00
}
2019-05-31 17:03:20 +03:00
pipe.init()
2017-05-25 14:16:39 +03:00
return &pipe
}