redis/redis.go

195 lines
3.9 KiB
Go
Raw Normal View History

2012-07-25 17:00:50 +04:00
package redis
import (
2012-08-05 16:09:43 +04:00
"crypto/tls"
2012-08-25 17:47:27 +04:00
"log"
2012-08-05 16:09:43 +04:00
"net"
2012-08-25 17:47:27 +04:00
"os"
2012-07-25 17:00:50 +04:00
"sync"
2012-10-26 19:21:14 +04:00
"time"
2012-08-09 14:12:41 +04:00
)
2012-07-25 17:00:50 +04:00
2012-08-25 17:47:27 +04:00
// Package logger.
var Logger = log.New(os.Stdout, "redis: ", log.Ldate|log.Ltime)
type OpenConnFunc func() (net.Conn, error)
type CloseConnFunc func(net.Conn) error
2012-08-05 16:09:43 +04:00
type InitConnFunc func(*Client) error
2012-07-25 17:00:50 +04:00
2012-08-05 16:09:43 +04:00
func TCPConnector(addr string) OpenConnFunc {
return func() (net.Conn, error) {
2012-10-26 19:21:14 +04:00
return net.DialTimeout("tcp", addr, 3*time.Second)
2012-08-05 16:09:43 +04:00
}
2012-07-26 22:43:21 +04:00
}
2012-08-05 16:09:43 +04:00
func TLSConnector(addr string, tlsConfig *tls.Config) OpenConnFunc {
return func() (net.Conn, error) {
2012-10-26 19:21:14 +04:00
conn, err := net.DialTimeout("tcp", addr, 3*time.Second)
if err != nil {
return nil, err
}
2012-11-11 18:37:50 +04:00
return tls.Client(conn, tlsConfig), nil
2012-08-05 16:09:43 +04:00
}
2012-07-25 17:00:50 +04:00
}
func UnixConnector(addr string) OpenConnFunc {
return func() (net.Conn, error) {
return net.DialTimeout("unix", addr, 3*time.Second)
}
}
2012-08-05 16:09:43 +04:00
func AuthSelectFunc(password string, db int64) InitConnFunc {
if password == "" && db < 0 {
return nil
}
2012-08-05 16:09:43 +04:00
return func(client *Client) error {
if password != "" {
2012-08-06 16:09:48 +04:00
auth := client.Auth(password)
if auth.Err() != nil {
return auth.Err()
2012-08-05 16:09:43 +04:00
}
}
2012-07-29 13:42:00 +04:00
if db >= 0 {
2012-08-06 16:09:48 +04:00
sel := client.Select(db)
if sel.Err() != nil {
return sel.Err()
}
2012-08-05 16:09:43 +04:00
}
return nil
2012-07-25 17:00:50 +04:00
}
}
2012-08-09 14:12:41 +04:00
//------------------------------------------------------------------------------
2012-07-26 19:16:17 +04:00
type BaseClient struct {
ConnPool ConnPool
2012-08-05 16:09:43 +04:00
InitConn InitConnFunc
reqs []Req
2012-08-25 23:44:53 +04:00
reqsMtx sync.Mutex
2012-07-26 19:16:17 +04:00
}
2012-08-14 19:20:22 +04:00
func (c *BaseClient) WriteReq(conn *Conn, reqs ...Req) error {
2012-08-17 22:36:48 +04:00
buf := make([]byte, 0, 1000)
2012-08-14 19:20:22 +04:00
for _, req := range reqs {
2012-08-17 22:36:48 +04:00
buf = appendReq(buf, req.Args())
2012-08-14 19:20:22 +04:00
}
2012-08-17 22:36:48 +04:00
_, err := conn.RW.Write(buf)
return err
2012-08-09 18:06:26 +04:00
}
func (c *BaseClient) conn() (*Conn, error) {
conn, isNew, err := c.ConnPool.Get()
if err != nil {
return nil, err
}
if isNew && c.InitConn != nil {
client := &Client{
BaseClient: &BaseClient{
ConnPool: NewSingleConnPoolConn(c.ConnPool, conn, true),
},
}
err = c.InitConn(client)
if err != nil {
if err := c.ConnPool.Remove(conn); err != nil {
2012-08-25 17:47:27 +04:00
Logger.Printf("ConnPool.Remove error: %v", err)
}
return nil, err
}
}
return conn, nil
}
func (c *BaseClient) Process(req Req) {
2012-08-05 16:09:43 +04:00
if c.reqs == nil {
c.Run(req)
} else {
c.Queue(req)
}
2012-07-25 17:00:50 +04:00
}
func (c *BaseClient) Run(req Req) {
conn, err := c.conn()
2012-07-26 22:43:21 +04:00
if err != nil {
req.SetErr(err)
return
}
2012-08-14 19:20:22 +04:00
err = c.WriteReq(conn, req)
2012-07-26 22:43:21 +04:00
if err != nil {
if err := c.ConnPool.Remove(conn); err != nil {
2012-08-25 17:47:27 +04:00
Logger.Printf("ConnPool.Remove error: %v", err)
}
2012-07-26 22:43:21 +04:00
req.SetErr(err)
return
}
2012-08-05 16:09:43 +04:00
val, err := req.ParseReply(conn.Rd)
2012-07-26 19:16:17 +04:00
if err != nil {
2012-08-26 13:18:42 +04:00
if _, ok := err.(*parserError); ok {
if err := c.ConnPool.Remove(conn); err != nil {
2012-08-25 17:47:27 +04:00
Logger.Printf("ConnPool.Remove error: %v", err)
}
2012-08-26 13:18:42 +04:00
} else {
if err := c.ConnPool.Add(conn); err != nil {
Logger.Printf("ConnPool.Add error: %v", err)
}
}
2012-07-26 19:16:17 +04:00
req.SetErr(err)
return
}
2012-08-05 16:09:43 +04:00
if err := c.ConnPool.Add(conn); err != nil {
2012-08-25 17:47:27 +04:00
Logger.Printf("ConnPool.Add error: %v", err)
}
2012-07-26 22:43:21 +04:00
req.SetVal(val)
2012-07-26 19:16:17 +04:00
}
2012-08-25 23:44:53 +04:00
// Queues request to be executed later.
func (c *BaseClient) Queue(req Req) {
2012-08-25 23:44:53 +04:00
c.reqsMtx.Lock()
c.reqs = append(c.reqs, req)
2012-08-25 23:44:53 +04:00
c.reqsMtx.Unlock()
}
2012-07-26 19:16:17 +04:00
func (c *BaseClient) Close() error {
return c.ConnPool.Close()
}
//------------------------------------------------------------------------------
type Client struct {
*BaseClient
2012-08-20 14:42:33 +04:00
TryEvalShaMinLen int
2012-08-05 16:09:43 +04:00
}
func NewClient(openConn OpenConnFunc, closeConn CloseConnFunc, initConn InitConnFunc) *Client {
return &Client{
BaseClient: &BaseClient{
ConnPool: NewMultiConnPool(openConn, closeConn, 10),
InitConn: initConn,
},
2012-08-20 14:42:33 +04:00
TryEvalShaMinLen: 80,
2012-07-29 13:42:00 +04:00
}
}
2012-08-05 16:09:43 +04:00
func NewTCPClient(addr string, password string, db int64) *Client {
return NewClient(TCPConnector(addr), nil, AuthSelectFunc(password, db))
}
2012-07-29 13:42:00 +04:00
func NewTLSClient(addr string, tlsConfig *tls.Config, password string, db int64) *Client {
return NewClient(
TLSConnector(addr, tlsConfig),
nil,
AuthSelectFunc(password, db),
)
2012-07-29 13:42:00 +04:00
}
func NewUnixClient(addr string, password string, db int64) *Client {
return NewClient(UnixConnector(addr), nil, AuthSelectFunc(password, db))
}