gitea/vendor/gopkg.in/redis.v2/redis.go

232 lines
4.1 KiB
Go
Raw Normal View History

2016-11-04 06:16:01 +08:00
package redis
import (
"log"
"net"
"time"
)
type baseClient struct {
connPool pool
opt *options
cmds []Cmder
}
func (c *baseClient) writeCmd(cn *conn, cmds ...Cmder) error {
buf := cn.buf[:0]
for _, cmd := range cmds {
buf = appendArgs(buf, cmd.args())
}
_, err := cn.Write(buf)
return err
}
func (c *baseClient) conn() (*conn, error) {
cn, isNew, err := c.connPool.Get()
if err != nil {
return nil, err
}
if isNew {
if err := c.initConn(cn); err != nil {
c.removeConn(cn)
return nil, err
}
}
return cn, nil
}
func (c *baseClient) initConn(cn *conn) error {
if c.opt.Password == "" && c.opt.DB == 0 {
return nil
}
pool := newSingleConnPool(c.connPool, false)
pool.SetConn(cn)
// Client is not closed because we want to reuse underlying connection.
client := &Client{
baseClient: &baseClient{
opt: c.opt,
connPool: pool,
},
}
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) freeConn(cn *conn, ei error) error {
if cn.rd.Buffered() > 0 {
return c.connPool.Remove(cn)
}
if _, ok := ei.(redisError); ok {
return c.connPool.Put(cn)
}
return c.connPool.Remove(cn)
}
func (c *baseClient) removeConn(cn *conn) {
if err := c.connPool.Remove(cn); err != nil {
log.Printf("pool.Remove failed: %s", err)
}
}
func (c *baseClient) putConn(cn *conn) {
if err := c.connPool.Put(cn); err != nil {
log.Printf("pool.Put failed: %s", err)
}
}
func (c *baseClient) Process(cmd Cmder) {
if c.cmds == nil {
c.run(cmd)
} else {
c.cmds = append(c.cmds, cmd)
}
}
func (c *baseClient) run(cmd Cmder) {
cn, err := c.conn()
if err != nil {
cmd.setErr(err)
return
}
if timeout := cmd.writeTimeout(); timeout != nil {
cn.writeTimeout = *timeout
} else {
cn.writeTimeout = c.opt.WriteTimeout
}
if timeout := cmd.readTimeout(); timeout != nil {
cn.readTimeout = *timeout
} else {
cn.readTimeout = c.opt.ReadTimeout
}
if err := c.writeCmd(cn, cmd); err != nil {
c.freeConn(cn, err)
cmd.setErr(err)
return
}
if err := cmd.parseReply(cn.rd); err != nil {
c.freeConn(cn, err)
return
}
c.putConn(cn)
}
// Close closes the client, releasing any open resources.
func (c *baseClient) Close() error {
return c.connPool.Close()
}
//------------------------------------------------------------------------------
type options struct {
Password string
DB int64
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
PoolSize int
IdleTimeout time.Duration
}
type Options struct {
Network string
Addr string
// Dialer creates new network connection and has priority over
// Network and Addr options.
Dialer func() (net.Conn, error)
Password string
DB int64
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
PoolSize int
IdleTimeout time.Duration
}
func (opt *Options) getPoolSize() int {
if opt.PoolSize == 0 {
return 10
}
return opt.PoolSize
}
func (opt *Options) getDialTimeout() time.Duration {
if opt.DialTimeout == 0 {
return 5 * time.Second
}
return opt.DialTimeout
}
func (opt *Options) options() *options {
return &options{
DB: opt.DB,
Password: opt.Password,
DialTimeout: opt.getDialTimeout(),
ReadTimeout: opt.ReadTimeout,
WriteTimeout: opt.WriteTimeout,
PoolSize: opt.getPoolSize(),
IdleTimeout: opt.IdleTimeout,
}
}
type Client struct {
*baseClient
}
func NewClient(clOpt *Options) *Client {
opt := clOpt.options()
dialer := clOpt.Dialer
if dialer == nil {
dialer = func() (net.Conn, error) {
return net.DialTimeout(clOpt.Network, clOpt.Addr, opt.DialTimeout)
}
}
return &Client{
baseClient: &baseClient{
opt: opt,
connPool: newConnPool(newConnFunc(dialer), opt),
},
}
}
// Deprecated. Use NewClient instead.
func NewTCPClient(opt *Options) *Client {
opt.Network = "tcp"
return NewClient(opt)
}
// Deprecated. Use NewClient instead.
func NewUnixClient(opt *Options) *Client {
opt.Network = "unix"
return NewClient(opt)
}