Merge branch 'feature/simple-pool-again' into develop

Author: dancannon

CHANGELOG.md

@@ -13,9 +13,13 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 ### Changed
  - Timezones from `time.Time` are now stored in the database, before all times were stored as UTC. To convert a go `time.Time` back to UTC you can call  `t.In(time.UTC)`.
  - Improved host selection to use `hailocab/go-hostpool` to select nodes based on recent responses and timings.
+ - Changed connection pool to use `fatih/pool` instead of a custom connection pool, this has caused some internal API changes and the behaviour of `MaxIdle` and `MaxOpen` has slightly changed. This change was made mostly to make driver maintenance easier.
+     + `MaxIdle` now configures the initial size of the pool, the name of this field will likely change in the future.
+     + Not setting `MaxOpen` no longer creates an unbounded connection pool per host but instead creates a pool with a maximum capacity of 2 per host.
 
 ### Deprecated
  - Deprecated the option `NodeRefreshInterval` in `ConnectOpts`
+ - Deprecated `SetMaxIdleConns` and `SetMaxOpenConns`, these options should now only be set when creating the session.
 
 ## v1.1.4
 ### Added

cluster.go

@@ -138,7 +138,7 @@ func (c *Cluster) discover() {
 
 			return c.listenForNodeChanges()
 		}, b, func(err error, wait time.Duration) {
-			Log.Debugf("Error discovering hosts %s, waiting %s", err, wait)
+			Log.Debugf("Error discovering hosts %s, waiting: %s", err, wait)
 		})
 	}
 }
@@ -158,7 +158,7 @@ func (c *Cluster) listenForNodeChanges() error {
 		c.opts,
 	)
 	if err != nil {
-		return fmt.Errorf("Error building query %s", err)
+		return fmt.Errorf("Error building query: %s", err)
 	}
 
 	cursor, err := node.Query(q)
@@ -217,7 +217,7 @@ func (c *Cluster) connectNodes(hosts []Host) {
 	for _, host := range hosts {
 		conn, err := NewConnection(host.String(), c.opts)
 		if err != nil {
-			Log.Warnf("Error creating connection %s", err.Error())
+			Log.Warnf("Error creating connection: %s", err.Error())
 			continue
 		}
 		defer conn.Close()
@@ -228,13 +228,13 @@ func (c *Cluster) connectNodes(hosts []Host) {
 			c.opts,
 		)
 		if err != nil {
-			Log.Warnf("Error building query %s", err)
+			Log.Warnf("Error building query: %s", err)
 			continue
 		}
 
 		_, cursor, err := conn.Query(q)
 		if err != nil {
-			Log.Warnf("Error fetching cluster status %s", err)
+			Log.Warnf("Error fetching cluster status: %s", err)
 			continue
 		}
 
@@ -260,6 +260,8 @@ func (c *Cluster) connectNodes(hosts []Host) {
 						}).Debug("Connected to node")
 						nodeSet[node.ID] = node
 					}
+				} else {
+					Log.Warnf("Error connecting to node: %s", err)
 				}
 			}
 		} else {
@@ -272,6 +274,8 @@ func (c *Cluster) connectNodes(hosts []Host) {
 					}).Debug("Connected to node")
 					nodeSet[node.ID] = node
 				}
+			} else {
+				Log.Warnf("Error connecting to node: %s", err)
 			}
 		}
 	}

connection.go

@@ -30,9 +30,10 @@ type Response struct {
 // Connection is a connection to a rethinkdb database. Connection is not thread
 // safe and should only be accessed be a single goroutine
 type Connection struct {
+	net.Conn
+
 	address string
 	opts    *ConnectOpts
-	conn    net.Conn
 
 	_       [4]byte
 	mu      sync.Mutex
@@ -52,19 +53,19 @@ func NewConnection(address string, opts *ConnectOpts) (*Connection, error) {
 	// Connect to Server
 	nd := net.Dialer{Timeout: c.opts.Timeout}
 	if c.opts.TLSConfig == nil {
-		c.conn, err = nd.Dial("tcp", address)
+		c.Conn, err = nd.Dial("tcp", address)
 	} else {
-		c.conn, err = tls.DialWithDialer(&nd, "tcp", address, c.opts.TLSConfig)
+		c.Conn, err = tls.DialWithDialer(&nd, "tcp", address, c.opts.TLSConfig)
 	}
 	if err != nil {
 		return nil, err
 	}
 	// Enable TCP Keepalives on TCP connections
-	if tc, ok := c.conn.(*net.TCPConn); ok {
+	if tc, ok := c.Conn.(*net.TCPConn); ok {
 		if err := tc.SetKeepAlive(true); err != nil {
 			// Don't send COM_QUIT before handshake.
-			c.conn.Close()
-			c.conn = nil
+			c.Conn.Close()
+			c.Conn = nil
 			return nil, err
 		}
 	}
@@ -88,9 +89,9 @@ func (c *Connection) Close() error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
-	if c.conn != nil {
-		c.conn.Close()
-		c.conn = nil
+	if c.Conn != nil {
+		c.Conn.Close()
+		c.Conn = nil
 	}
 
 	c.cursors = nil
@@ -108,7 +109,7 @@ func (c *Connection) Query(q Query) (*Response, *Cursor, error) {
 		c.mu.Unlock()
 		return nil, nil, ErrConnectionClosed
 	}
-	if c.conn == nil {
+	if c.Conn == nil {
 		c.bad = true
 		c.mu.Unlock()
 		return nil, nil, ErrConnectionClosed
@@ -191,9 +192,9 @@ func (c *Connection) sendQuery(q Query) error {
 
 	// Set timeout
 	if c.opts.WriteTimeout == 0 {
-		c.conn.SetWriteDeadline(time.Time{})
+		c.Conn.SetWriteDeadline(time.Time{})
 	} else {
-		c.conn.SetWriteDeadline(time.Now().Add(c.opts.WriteTimeout))
+		c.Conn.SetWriteDeadline(time.Now().Add(c.opts.WriteTimeout))
 	}
 
 	// Send the JSON encoding of the query itself.
@@ -217,9 +218,9 @@ func (c *Connection) nextToken() int64 {
 func (c *Connection) readResponse() (*Response, error) {
 	// Set timeout
 	if c.opts.ReadTimeout == 0 {
-		c.conn.SetReadDeadline(time.Time{})
+		c.Conn.SetReadDeadline(time.Time{})
 	} else {
-		c.conn.SetReadDeadline(time.Now().Add(c.opts.ReadTimeout))
+		c.Conn.SetReadDeadline(time.Now().Add(c.opts.ReadTimeout))
 	}
 
 	// Read response header (token+length)
@@ -350,6 +351,13 @@ func (c *Connection) processWaitResponse(q Query, response *Response) (*Response
 	return response, nil, nil
 }
 
+func (c *Connection) isBad() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	return c.bad
+}
+
 var responseCache = make(chan *Response, 16)
 
 func newCachedResponse() *Response {

connection_helper.go

@@ -12,7 +12,7 @@ import (
 
 // Write 'data' to conn
 func (c *Connection) writeData(data []byte) error {
-	_, err := c.conn.Write(data[:])
+	_, err := c.Conn.Write(data[:])
 	if err != nil {
 		return RQLConnectionError{err.Error()}
 	}
@@ -46,7 +46,7 @@ func (c *Connection) writeHandshakeReq() error {
 }
 
 func (c *Connection) readHandshakeSuccess() error {
-	reader := bufio.NewReader(c.conn)
+	reader := bufio.NewReader(c.Conn)
 	line, err := reader.ReadBytes('\x00')
 	if err != nil {
 		if err == io.EOF {
@@ -68,7 +68,7 @@ func (c *Connection) readHandshakeSuccess() error {
 func (c *Connection) read(buf []byte, length int) (total int, err error) {
 	var n int
 	for total < length {
-		if n, err = c.conn.Read(buf[total:length]); err != nil {
+		if n, err = c.Conn.Read(buf[total:length]); err != nil {
 			break
 		}
 		total += n

cursor.go

@@ -47,7 +47,7 @@ func newCursor(conn *Connection, cursorType string, token int64, term *Term, opt
 //     err = cursor.Err() // get any error encountered during iteration
 //     ...
 type Cursor struct {
-	releaseConn func(error)
+	releaseConn func() error
 
 	conn       *Connection
 	token      int64
@@ -111,7 +111,7 @@ func (c *Cursor) Close() error {
 	if conn == nil {
 		return nil
 	}
-	if conn.conn == nil {
+	if conn.Conn == nil {
 		return nil
 	}
 
@@ -129,7 +129,9 @@ func (c *Cursor) Close() error {
 	}
 
 	if c.releaseConn != nil {
-		c.releaseConn(err)
+		if err := c.releaseConn(); err != nil {
+			return err
+		}
 	}
 
 	c.closed = true

cursor_test.go

@@ -355,6 +355,7 @@ func (s *RethinkSuite) TestCursorChangesClose(c *test.C) {
 	// res, err := DB("test").Table("Table3").Changes().Run(session)
 	res, err := DB("test").Table("Table3").Changes().Run(session)
 	c.Assert(err, test.IsNil)
+	c.Assert(res, test.NotNil)
 
 	// Ensure that the cursor can be closed
 	err = res.Close()