v2: Remove timed out connection from the pool.

This commit is contained in:
Vladimir Mihailenco 2013-09-17 10:35:52 +03:00
parent dc90e359e9
commit 73b92efa94
3 changed files with 18 additions and 25 deletions

View File

@ -29,16 +29,6 @@ var (
//------------------------------------------------------------------------------
type parserError struct {
err error
}
func (e *parserError) Error() string {
return e.err.Error()
}
//------------------------------------------------------------------------------
func appendReq(buf []byte, args []string) []byte {
buf = append(buf, '*')
buf = strconv.AppendUint(buf, uint64(len(args)), 10)
@ -163,7 +153,7 @@ func parseStringFloatMapReply(rd reader) (interface{}, error) {
func _parseReply(rd reader, typ replyType) (interface{}, error) {
line, err := readLine(rd)
if err != nil {
return 0, &parserError{err}
return 0, err
}
switch line[0] {
@ -174,7 +164,7 @@ func _parseReply(rd reader, typ replyType) (interface{}, error) {
case ':':
v, err := strconv.ParseInt(string(line[1:]), 10, 64)
if err != nil {
return 0, &parserError{err}
return 0, err
}
return v, nil
case '$':
@ -184,13 +174,13 @@ func _parseReply(rd reader, typ replyType) (interface{}, error) {
replyLenInt32, err := strconv.ParseInt(string(line[1:]), 10, 32)
if err != nil {
return "", &parserError{err}
return "", err
}
replyLen := int(replyLenInt32) + 2
line, err = readN(rd, replyLen)
if err != nil {
return "", &parserError{err}
return "", err
}
return string(line[:len(line)-2]), nil
case '*':
@ -200,7 +190,7 @@ func _parseReply(rd reader, typ replyType) (interface{}, error) {
repliesNum, err := strconv.ParseInt(string(line[1:]), 10, 64)
if err != nil {
return nil, &parserError{err}
return nil, err
}
switch typ {
@ -278,7 +268,7 @@ func _parseReply(rd reader, typ replyType) (interface{}, error) {
}
value, err := strconv.ParseFloat(valueS, 64)
if err != nil {
return nil, &parserError{err}
return nil, err
}
m[key] = value
@ -299,6 +289,6 @@ func _parseReply(rd reader, typ replyType) (interface{}, error) {
return vals, nil
}
default:
return nil, &parserError{fmt.Errorf("redis: can't parse %q", line)}
return nil, fmt.Errorf("redis: can't parse %q", line)
}
}

View File

@ -4,6 +4,8 @@ type PipelineClient struct {
*Client
}
// TODO: rename to Pipeline
// TODO: return just *PipelineClient
func (c *Client) PipelineClient() (*PipelineClient, error) {
return &PipelineClient{
Client: &Client{
@ -39,6 +41,8 @@ func (c *PipelineClient) DiscardQueued() {
c.reqsMtx.Unlock()
}
// TODO: rename to Run or ...
// TODO: should return error if one of the commands failed
func (c *PipelineClient) RunQueued() ([]Req, error) {
c.reqsMtx.Lock()
reqs := c.reqs
@ -49,18 +53,17 @@ func (c *PipelineClient) RunQueued() ([]Req, error) {
return []Req{}, nil
}
conn, err := c.conn()
cn, err := c.conn()
if err != nil {
return nil, err
}
err = c.runReqs(reqs, conn)
if err != nil {
c.removeConn(conn)
if err := c.runReqs(reqs, cn); err != nil {
c.removeConn(cn)
return nil, err
}
c.putConn(conn)
c.putConn(cn)
return reqs, nil
}

View File

@ -119,10 +119,10 @@ func (c *baseClient) run(req Req) {
val, err := req.ParseReply(cn.Rd)
if err != nil {
if _, ok := err.(*parserError); ok {
c.removeConn(cn)
} else {
if err == Nil {
c.putConn(cn)
} else {
c.removeConn(cn)
}
req.SetErr(err)
return