Added config, sentinel

Author: keimoon

config.go

@@ -0,0 +1,16 @@
+package gore
+
+// Config keeps some default configurations. Time is measured in second
+var Config = &struct {
+	ConnectTimeout  int
+	RequestTimeout  int
+	ReconnectTime   int
+	PoolInitialSize int
+	PoolMaximumSize int
+}{
+	ConnectTimeout:  5,
+	RequestTimeout:  10,
+	ReconnectTime:   2,
+	PoolInitialSize: 5,
+	PoolMaximumSize: 10,
+}

conn.go

@@ -21,29 +21,30 @@ type Conn struct {
 	mutex          sync.Mutex
 	rb             *bufio.Reader
 	wb             *bufio.Writer
+	sentinel       bool
 	RequestTimeout time.Duration
 }
 
 // Dial opens a TCP connection with a redis server.
 func Dial(address string) (*Conn, error) {
 	conn := &Conn{
-		RequestTimeout: 10 * time.Second,
+		RequestTimeout: time.Duration(Config.RequestTimeout) * time.Second,
 	}
 	conn.mutex.Lock()
-        defer conn.mutex.Unlock()
+	defer conn.mutex.Unlock()
 	err := conn.connect(address, 0)
 	return conn, err
 }
 
 // DialTimeout opens a TCP connection with a redis server with a connection timeout
 func DialTimeout(address string, timeout time.Duration) (*Conn, error) {
-        conn := &Conn{
-		RequestTimeout: 10 * time.Second,
-        }
+	conn := &Conn{
+		RequestTimeout: time.Duration(Config.RequestTimeout) * time.Second,
+	}
 	conn.mutex.Lock()
-        defer conn.mutex.Unlock()
-        err := conn.connect(address, timeout)
-        return conn, err
+	defer conn.mutex.Unlock()
+	err := conn.connect(address, timeout)
+	return conn, err
 }
 
 // Close closes the connection
@@ -92,21 +93,19 @@ func (c *Conn) connect(address string, timeout time.Duration) error {
 }
 
 func (c *Conn) fail() {
-	c.reconnect()
-}
-
-func (c *Conn) reconnect() {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
-	if c.state == connStateReconnecting {
-		return
+	if !c.sentinel {
+		c.mutex.Lock()
+		defer c.mutex.Unlock()
+		if c.state == connStateReconnecting {
+			return
+		}
+		c.tcpConn.Close()
+		c.state = connStateReconnecting
+		for {
+			if err := c.connect(c.address, 0); err == nil {
+				break
+			}
+			time.Sleep(time.Duration(Config.ReconnectTime) * time.Second)
+		}
 	}
-	c.tcpConn.Close()
-	c.state = connStateReconnecting
-	for {
-                if err := c.connect(c.address, 0); err == nil {
-			break
-                }
-                time.Sleep(2 * time.Second)
-        }
 }

doc.go

@@ -83,6 +83,11 @@ The following slice element types are supported:
   - FixInt and VarInt
   - *gore.Pair for converting map data from HGETALL or ZRANGE WITHSCORES
 
+Reply returns from HGETALL or SENTINEL master can be converted into a map
+using Map:
+
+  m, err:= rep.Map()
+
 Pipeline
 
 Gore supports pipelining using gore.Pipeline:

pool.go

@@ -24,6 +24,7 @@ type Pool struct {
 	cond                 *sync.Cond
 	address              string
 	closed               bool
+	sentinel             bool
 }
 
 // Dial initializes connection from the pool to redis server.
@@ -34,10 +35,10 @@ func (p *Pool) Dial(address string) error {
 		p.RequestTimeout = 10 * time.Second
 	}
 	if p.InitialConn <= 0 {
-		p.InitialConn = 5
+		p.InitialConn = Config.PoolInitialSize
 	}
 	if p.MaximumConn <= 0 {
-		p.MaximumConn = 10
+		p.MaximumConn = Config.PoolMaximumSize
 	}
 	if p.MaximumConn < p.InitialConn {
 		p.MaximumConn = p.InitialConn
@@ -63,6 +64,7 @@ func (p *Pool) Close() {
 	}
 	p.l.Init()
 	p.currentNumberOfConn = p.l.Len()
+	p.unusableNumberOfConn = 0
 	p.cond.Broadcast()
 }
 
@@ -137,6 +139,7 @@ func (p *Pool) connect(timeout time.Duration) (err error) {
 		if err != nil {
 			return err
 		}
+		conn.sentinel = p.sentinel
 		p.l.PushBack(conn)
 	}
 	return nil
@@ -154,10 +157,24 @@ func (p *Pool) pushBack(conn *Conn) {
 			p.cond.Signal()
 			p.mutex.Unlock()
 			break
+		} else if p.sentinel {
+			// Give up this conn
+			conn.Close()
+			break
 		} else if !markedUnusable {
 			markedUnusable = true
 			p.unusableNumberOfConn++
 		}
 		time.Sleep(2 * time.Second)
 	}
 }
+
+func (p *Pool) sentinelGonnaGiveYouUp() {
+	for p.connect(time.Duration(Config.ConnectTimeout)*time.Second) != nil {
+	}
+	p.closed = false
+}
+
+func (p *Pool) sentinelGonnaLetYouDown() {
+	p.Close()
+}

pubsub.go

@@ -35,6 +35,8 @@ type Subscriptions struct {
 	lock           sync.Mutex
 	ready          bool
 	readyChannel   chan bool
+	// Sentinel will set this to true to handle read error from sentinel server.
+	throwError bool
 }
 
 // NewSubscriptions returns new Subscriptions
@@ -46,6 +48,7 @@ func NewSubscriptions(conn *Conn) *Subscriptions {
 		messageChannel: make(chan *Message, 100),
 		ready:          true,
 		readyChannel:   make(chan bool, 1),
+		throwError:     false,
 	}
 	go s.receive()
 	return s
@@ -114,7 +117,12 @@ func (s *Subscriptions) receive() {
 		for {
 			rep, err := readReply(s.conn)
 			if err != nil {
+				if s.throwError {
+					s.messageChannel <- nil
+					return
+				}
 				s.ready = false
+				go s.resubscribe()
 				break
 			}
 			if !rep.IsArray() {
@@ -155,10 +163,7 @@ func (s *Subscriptions) do(command string, channel ...string) error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 	err := NewCommand(command, s.makeArgs(channel...)...).Send(s.conn)
-	if err != nil {
-		s.ready = false
-		go s.resubscribe()
-	} else {
+	if err == nil {
 		switch {
 		case command[0] == 'S':
 			for _, ch := range channel {
@@ -211,8 +216,9 @@ func (s *Subscriptions) resubscribe() {
 		}
 		err = NewCommand("PSUBSCRIBE", s.makeArgs(pchannels...)...).Send(s.conn)
 		if err != nil {
-                        continue
-                }
+			continue
+		}
+		break
 	}
 	s.ready = true
 	s.readyChannel <- true

reply.go

@@ -254,6 +254,34 @@ func (r *Reply) Slice(s interface{}) error {
 	return nil
 }
 
+// Map converts the reply into a map[string]string.
+// It will return error unless the reply is an array reply from HGETALL,
+// or SENTINEL master
+func (r *Reply) Map() (map[string]string, error) {
+	if r.IsNil() {
+		return nil, ErrNil
+	}
+	if !r.IsArray() {
+		return nil, ErrType
+	}
+	if len(r.arrayValue)%2 != 0 {
+		return nil, ErrType
+	}
+	m := make(map[string]string)
+	for i := 0; i < len(r.arrayValue)/2; i++ {
+		first, err := r.arrayValue[2*i].String()
+		if err != nil {
+			continue
+		}
+		second, err := r.arrayValue[2*i+1].String()
+		if err != nil && err != ErrNil {
+			continue
+		}
+		m[first] = second
+	}
+	return m, nil
+}
+
 // Error returns error message
 func (r *Reply) Error() (string, error) {
 	if r.Type() != ReplyError {
@@ -301,10 +329,10 @@ func (r *Reply) IsError() bool {
 func Receive(conn *Conn) (r *Reply, err error) {
 	conn.Lock()
 	defer func() {
-                conn.Unlock()
-                if err != nil {
-                        conn.fail()
-                }
+		conn.Unlock()
+		if err != nil {
+			conn.fail()
+		}
 	}()
 	if conn.RequestTimeout != 0 {
 		conn.tcpConn.SetReadDeadline(time.Now().Add(conn.RequestTimeout))