redis/redis.go

289 lines
6.1 KiB
Go
Raw Normal View History

2016-11-26 11:33:06 +03:00
package redis // import "gopkg.in/redis.v5"
2012-07-25 17:00:50 +04:00
import (
2015-05-15 15:21:28 +03:00
"fmt"
2014-07-31 17:01:54 +04:00
"log"
"time"
2016-10-09 13:49:28 +03:00
"gopkg.in/redis.v5/internal"
"gopkg.in/redis.v5/internal/pool"
2012-08-09 14:12:41 +04:00
)
2012-07-25 17:00:50 +04:00
2016-07-02 15:52:10 +03:00
// Redis nil reply, .e.g. when key does not exist.
const Nil = internal.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
}
2014-05-11 11:42:40 +04:00
type baseClient struct {
connPool pool.Pooler
opt *Options
process func(Cmder) error
onClose func() error // hook called when client is closed
2012-07-26 22:43:21 +04:00
}
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
}
2016-09-29 15:07:04 +03:00
func (c *baseClient) conn() (*pool.Conn, bool, error) {
cn, isNew, err := c.connPool.Get()
2016-03-15 15:04:35 +03:00
if err != nil {
2016-09-29 15:07:04 +03:00
return nil, false, err
2016-03-15 15:04:35 +03:00
}
if !cn.Inited {
if err := c.initConn(cn); err != nil {
2016-03-17 19:00:47 +03:00
_ = c.connPool.Remove(cn, err)
2016-09-29 15:07:04 +03:00
return nil, false, err
2016-03-12 13:41:02 +03:00
}
}
2016-09-29 15:07:04 +03:00
return cn, isNew, nil
2012-07-26 19:16:17 +04:00
}
func (c *baseClient) putConn(cn *pool.Conn, err error, allowTimeout bool) bool {
if internal.IsBadConn(err, allowTimeout) {
2016-03-17 19:00:47 +03:00
_ = c.connPool.Remove(cn, err)
return false
}
_ = c.connPool.Put(cn)
return true
}
2016-03-12 13:41:02 +03:00
func (c *baseClient) initConn(cn *pool.Conn) error {
2016-03-15 15:04:35 +03:00
cn.Inited = true
if c.opt.Password == "" && c.opt.DB == 0 && !c.opt.ReadOnly {
2016-03-12 13:41:02 +03:00
return nil
}
// Temp client for Auth and Select.
client := newClient(c.opt, pool.NewSingleConnPool(cn))
_, err := client.Pipelined(func(pipe *Pipeline) error {
if c.opt.Password != "" {
pipe.Auth(c.opt.Password)
2016-03-12 13:41:02 +03:00
}
if c.opt.DB > 0 {
pipe.Select(c.opt.DB)
2016-03-12 13:41:02 +03:00
}
if c.opt.ReadOnly {
pipe.ReadOnly()
}
return nil
})
return err
2016-03-12 13:41:02 +03:00
}
2016-06-17 15:09:38 +03:00
func (c *baseClient) Process(cmd Cmder) error {
if c.process != nil {
return c.process(cmd)
}
return c.defaultProcess(cmd)
}
// WrapProcess replaces the process func. It takes a function createWrapper
// which is supplied by the user. createWrapper takes the old process func as
// an input and returns the new wrapper process func. createWrapper should
// use call the old process func within the new process func.
func (c *baseClient) WrapProcess(fn func(oldProcess func(cmd Cmder) error) func(cmd Cmder) error) {
c.process = fn(c.defaultProcess)
}
func (c *baseClient) defaultProcess(cmd Cmder) error {
for i := 0; i <= c.opt.MaxRetries; i++ {
if i > 0 {
cmd.reset()
}
2012-07-26 22:43:21 +04:00
2016-09-29 15:07:04 +03:00
cn, _, err := c.conn()
if err != nil {
cmd.setErr(err)
2016-06-17 15:09:38 +03:00
return err
}
2012-07-26 22:43:21 +04:00
cn.SetWriteTimeout(c.opt.WriteTimeout)
if err := writeCmd(cn, cmd); err != nil {
2016-03-08 18:18:52 +03:00
c.putConn(cn, err, false)
cmd.setErr(err)
if err != nil && internal.IsRetryableError(err) {
continue
}
2016-06-17 15:09:38 +03:00
return err
}
cn.SetReadTimeout(c.cmdTimeout(cmd))
2015-10-07 17:09:20 +03:00
err = cmd.readReply(cn)
c.putConn(cn, err, false)
if err != nil && internal.IsRetryableError(err) {
continue
}
2016-06-17 15:09:38 +03:00
return err
2012-07-26 19:16:17 +04:00
}
2016-06-17 15:09:38 +03:00
return cmd.Err()
2014-05-11 11:42:40 +04:00
}
func (c *baseClient) cmdTimeout(cmd Cmder) time.Duration {
if timeout := cmd.readTimeout(); timeout != nil {
return *timeout
} else {
return c.opt.ReadTimeout
}
}
func (c *baseClient) closed() bool {
return c.connPool.Closed()
}
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 == 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
}
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 {
2016-03-09 12:49:05 +03:00
baseClient
cmdable
2015-01-24 15:12:48 +03:00
}
func newClient(opt *Options, pool pool.Pooler) *Client {
2016-03-09 12:49:05 +03:00
base := baseClient{opt: opt, connPool: pool}
2016-04-13 11:52:47 +03:00
client := &Client{
baseClient: base,
cmdable: cmdable{base.Process},
2015-01-24 15:12:48 +03:00
}
2016-04-13 11:52:47 +03:00
return client
2012-08-05 16:09:43 +04:00
}
2015-09-12 09:36:03 +03:00
// NewClient returns a client to the Redis Server specified by Options.
func NewClient(opt *Options) *Client {
2016-06-05 14:10:30 +03:00
opt.init()
return newClient(opt, newConnPool(opt))
}
2016-01-19 19:36:40 +03:00
// PoolStats returns connection pool stats.
2016-01-19 19:36:40 +03:00
func (c *Client) PoolStats() *PoolStats {
s := c.connPool.Stats()
return &PoolStats{
2016-03-19 17:10:34 +03:00
Requests: s.Requests,
Hits: s.Hits,
Timeouts: s.Timeouts,
2016-03-19 17:10:34 +03:00
TotalConns: s.TotalConns,
FreeConns: s.FreeConns,
}
2016-01-19 19:36:40 +03:00
}
func (c *Client) Pipeline() *Pipeline {
pipe := Pipeline{
exec: c.pipelineExec,
}
pipe.cmdable.process = pipe.Process
pipe.statefulCmdable.process = pipe.Process
return &pipe
}
func (c *Client) Pipelined(fn func(*Pipeline) error) ([]Cmder, error) {
return c.Pipeline().pipelined(fn)
}
func (c *Client) pipelineExec(cmds []Cmder) error {
var firstErr error
for i := 0; i <= c.opt.MaxRetries; i++ {
if i > 0 {
resetCmds(cmds)
}
2016-09-29 15:07:04 +03:00
cn, _, err := c.conn()
if err != nil {
setCmdsErr(cmds, err)
return err
}
retry, err := c.execCmds(cn, cmds)
c.putConn(cn, err, false)
if err == nil {
return nil
}
if firstErr == nil {
firstErr = err
}
if !retry {
break
}
}
return firstErr
}
func (c *Client) execCmds(cn *pool.Conn, cmds []Cmder) (retry bool, firstErr error) {
cn.SetWriteTimeout(c.opt.WriteTimeout)
if err := writeCmd(cn, cmds...); err != nil {
setCmdsErr(cmds, err)
return true, err
}
// Set read timeout for all commands.
cn.SetReadTimeout(c.opt.ReadTimeout)
for i, cmd := range cmds {
err := cmd.readReply(cn)
if err == nil {
continue
}
if i == 0 && internal.IsNetworkError(err) {
return true, err
}
if firstErr == nil {
firstErr = err
}
}
return false, firstErr
}
func (c *Client) pubSub() *PubSub {
return &PubSub{
base: baseClient{
opt: c.opt,
connPool: pool.NewStickyConnPool(c.connPool.(*pool.ConnPool), false),
},
}
}
// Subscribe subscribes the client to the specified channels.
func (c *Client) Subscribe(channels ...string) (*PubSub, error) {
pubsub := c.pubSub()
return pubsub, pubsub.Subscribe(channels...)
}
// PSubscribe subscribes the client to the given patterns.
func (c *Client) PSubscribe(channels ...string) (*PubSub, error) {
pubsub := c.pubSub()
return pubsub, pubsub.PSubscribe(channels...)
}