redis/redis.go

package redis // import "gopkg.in/redis.v3"

import (
	"fmt"
	"log"
	"os"
	"sync/atomic"

	"gopkg.in/redis.v3/internal/pool"
)

// Deprecated. Use SetLogger instead.
var Logger *log.Logger

func init() {
	SetLogger(log.New(os.Stderr, "redis: ", log.LstdFlags))
}

func SetLogger(logger *log.Logger) {
	Logger = logger
	pool.Logger = logger
}

type baseClient struct {
	connPool pool.Pooler
	opt      *Options

	onClose func() error // hook called when client is closed
}

func (c *baseClient) String() string {
	return fmt.Sprintf("Redis<%s db:%d>", c.opt.Addr, c.opt.DB)
}

func (c *baseClient) conn() (*pool.Conn, bool, error) {
	cn, isNew, err := c.connPool.Get()
	if err == nil && isNew {
		err = c.initConn(cn)
		if err != nil {
			c.putConn(cn, err, false)
		}
	}
	return cn, isNew, err
}

func (c *baseClient) putConn(cn *pool.Conn, err error, allowTimeout bool) bool {
	if isBadConn(err, allowTimeout) {
		err = c.connPool.Replace(cn, err)
		if err != nil {
			Logger.Printf("pool.Remove failed: %s", err)
		}
		return false
	}

	err = c.connPool.Put(cn)
	if err != nil {
		Logger.Printf("pool.Put failed: %s", err)
	}
	return true
}

func (c *baseClient) initConn(cn *pool.Conn) error {
	if c.opt.Password == "" && c.opt.DB == 0 {
		return nil
	}

	// Temp client for Auth and Select.
	client := newClient(c.opt, pool.NewSingleConnPool(cn))

	if c.opt.Password != "" {
		if err := client.Auth(c.opt.Password).Err(); err != nil {
			return err
		}
	}

	if c.opt.DB > 0 {
		if err := client.Select(c.opt.DB).Err(); err != nil {
			return err
		}
	}

	return nil
}

func (c *baseClient) process(cmd Cmder) {
	for i := 0; i <= c.opt.MaxRetries; i++ {
		if i > 0 {
			cmd.reset()
		}

		cn, _, err := c.conn()
		if err != nil {
			cmd.setErr(err)
			return
		}

		readTimeout := cmd.readTimeout()
		if readTimeout != nil {
			cn.ReadTimeout = *readTimeout
		} else {
			cn.ReadTimeout = c.opt.ReadTimeout
		}
		cn.WriteTimeout = c.opt.WriteTimeout

		if err := writeCmd(cn, cmd); err != nil {
			c.putConn(cn, err, false)
			cmd.setErr(err)
			if shouldRetry(err) {
				continue
			}
			return
		}

		err = cmd.readReply(cn)
		c.putConn(cn, err, readTimeout != nil)
		if shouldRetry(err) {
			continue
		}

		return
	}
}

// Close closes the client, releasing any open resources.
//
// It is rare to Close a Client, as the Client is meant to be
// long-lived and shared between many goroutines.
func (c *baseClient) Close() error {
	var retErr error
	if c.onClose != nil {
		if err := c.onClose(); err != nil && retErr == nil {
			retErr = err
		}
	}
	if err := c.connPool.Close(); err != nil && retErr == nil {
		retErr = err
	}
	return retErr
}

//------------------------------------------------------------------------------

// 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 {
	baseClient
	commandable
}

func newClient(opt *Options, pool pool.Pooler) *Client {
	base := baseClient{opt: opt, connPool: pool}
	return &Client{
		baseClient: base,
		commandable: commandable{
			process: base.process,
		},
	}
}

// NewClient returns a client to the Redis Server specified by Options.
func NewClient(opt *Options) *Client {
	return newClient(opt, newConnPool(opt))
}

// PoolStats returns connection pool stats.
func (c *Client) PoolStats() *PoolStats {
	s := c.connPool.Stats()
	return &PoolStats{
		Requests: atomic.LoadUint32(&s.Requests),
		Hits:     atomic.LoadUint32(&s.Hits),
		Waits:    atomic.LoadUint32(&s.Waits),
		Timeouts: atomic.LoadUint32(&s.Timeouts),

		TotalConns: atomic.LoadUint32(&s.TotalConns),
		FreeConns:  atomic.LoadUint32(&s.FreeConns),
	}
}