changes offset

Author: CMogilko

CHANGELOG.md

@@ -7,6 +7,8 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 - Reworked and tested new connection pools with multiple queries per connection
 - Socket Read- and WriteTimeout replaced with context timeout
 - Mock assert fix
+- Connection pool fixed initial size
+- Changes added offsets
 
 ## v6.0.0 - 2019-12-22
 

pool.go

@@ -12,16 +12,21 @@ var (
 	errPoolClosed = errors.New("rethinkdb: pool is closed")
 )
 
+const (
+	poolIsNotClosed int32 = 0
+	poolIsClosed    int32 = 1
+)
+
 // A Pool is used to store a pool of connections to a single RethinkDB server
 type Pool struct {
 	host Host
 	opts *ConnectOpts
 
 	conns   []*Connection
 	pointer int32
+	closed  int32
 
-	mu     sync.RWMutex // protects following fields
-	closed bool
+	mu sync.Mutex // protects lazy creating connections
 }
 
 // NewPool creates a new connection pool for the given host
@@ -40,7 +45,7 @@ func NewPool(host Host, opts *ConnectOpts) (*Pool, error) {
 
 	conns := make([]*Connection, maxOpen)
 	var err error
-	for i := range conns {
+	for i := 0; i < opts.InitialCap; i++ {
 		conns[i], err = NewConnection(host.String(), opts)
 		if err != nil {
 			return nil, err
@@ -52,6 +57,7 @@ func NewPool(host Host, opts *ConnectOpts) (*Pool, error) {
 		pointer: -1,
 		host:    host,
 		opts:    opts,
+		closed:  poolIsNotClosed,
 	}, nil
 }
 
@@ -67,11 +73,17 @@ func (p *Pool) Ping() error {
 // It is rare to Close a Pool, as the Pool handle is meant to be
 // long-lived and shared between many goroutines.
 func (p *Pool) Close() error {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	if p.closed {
+	if atomic.LoadInt32(&p.closed) == poolIsClosed {
+		return nil
+	}
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if p.closed == poolIsClosed {
 		return nil
 	}
+	p.closed = poolIsClosed
 
 	for _, c := range p.conns {
 		err := c.Close()
@@ -84,20 +96,29 @@ func (p *Pool) Close() error {
 }
 
 func (p *Pool) conn() (*Connection, error) {
-	p.mu.RLock()
-
-	if p.closed {
-		p.mu.RUnlock()
+	if atomic.LoadInt32(&p.closed) == poolIsClosed {
 		return nil, errPoolClosed
 	}
-	p.mu.RUnlock()
 
 	pos := atomic.AddInt32(&p.pointer, 1)
 	if pos == int32(len(p.conns)) {
 		atomic.StoreInt32(&p.pointer, 0)
 	}
 	pos = pos % int32(len(p.conns))
 
+	if p.conns[pos] == nil {
+		p.mu.Lock()
+		defer p.mu.Unlock()
+
+		if p.conns[pos] == nil {
+			var err error
+			p.conns[pos], err = NewConnection(p.host.String(), p.opts)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+
 	return p.conns[pos], nil
 }
 

query.go

@@ -289,11 +289,13 @@ type WriteResponse struct {
 // ChangeResponse is a helper type used when dealing with changefeeds. The type
 // contains both the value before the query and the new value.
 type ChangeResponse struct {
-	NewValue interface{} `rethinkdb:"new_val,omitempty"`
-	OldValue interface{} `rethinkdb:"old_val,omitempty"`
-	State    string      `rethinkdb:"state,omitempty"`
-	Error    string      `rethinkdb:"error,omitempty"`
-	Type     string      `rethinkdb:"type,omitempty"`
+	NewValue  interface{} `rethinkdb:"new_val,omitempty"`
+	OldValue  interface{} `rethinkdb:"old_val,omitempty"`
+	State     string      `rethinkdb:"state,omitempty"`
+	Error     string      `rethinkdb:"error,omitempty"`
+	Type      string      `rethinkdb:"type,omitempty"`
+	OldOffset int         `rethinkdb:"old_offset,omitempty"`
+	NewOffset int         `rethinkdb:"new_offset,omitempty"`
 }
 
 // RunOpts contains the optional arguments for the Run function.

session.go

@@ -78,10 +78,8 @@ type ConnectOpts struct {
 	// the first query is executed.
 	InitialCap int `rethinkdb:"initial_cap,omitempty" json:"initial_cap,omitempty"`
 	// MaxOpen is used by the internal connection pool and is used to configure
-	// the maximum number of connections held in the pool. If all available
-	// connections are being used then the driver will open new connections as
-	// needed however they will not be returned to the pool. By default the
-	// maximum number of connections is 2
+	// the maximum number of connections held in the pool. By default the
+	// maximum number of connections is 1
 	MaxOpen int `rethinkdb:"max_open,omitempty" json:"max_open,omitempty"`
 
 	// Below options are for cluster discovery, please note there is a high