Merge branch 'release/v0.3.1'

Author: dancannon

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## v0.3.1 - 14 June 2014
+
+- Fixed "Token ## not in stream cache" error (#103)
+- Changed Exec to no longer use NoReply. It now waits for the server to respond.
+
 ## v0.3 (RethinkDB v1.13) - 26 June 2014
 
 - Replaced `ResultRows`/`ResultRow` with `Cursor`, `Cursor` has the `Next`, `All` and `One` methods which stores the relevant value in the value pointed at by result. For more information check the examples.

README.md

@@ -4,7 +4,7 @@ GoRethink - RethinkDB Driver for Go [![wercker status](https://app.wercker.com/s
 [Go](http://golang.org/) driver for [RethinkDB](http://www.rethinkdb.com/) made by [Daniel Cannon](http://github.com/dancannon) and based off of Christopher Hesse's [RethinkGo](https://github.com/christopherhesse/rethinkgo) driver.
 
 
-Current version: v0.3 (RethinkDB v1.13) 
+Current version: v0.3.1 (RethinkDB v1.13) 
 
 **Version 0.3 introduced some API changes, for more information check the [change log](CHANGELOG.md)**
 

connection.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net"
+	"sync"
 	"time"
 
 	"code.google.com/p/goprotobuf/proto"
@@ -23,7 +24,9 @@ type Connection struct {
 	// embed the net.Conn type, so that we can effectively define new methods on
 	// it (interfaces do not allow that)
 	net.Conn
+	s *Session
 
+	sync.Mutex
 	closed bool
 }
 
@@ -73,7 +76,10 @@ func Dial(s *Session) (*Connection, error) {
 		return nil, RqlDriverError{fmt.Sprintf("Server dropped connection with message: \"%s\"", response)}
 	}
 
-	return &Connection{conn, false}, nil
+	return &Connection{
+		s:    s,
+		Conn: conn,
+	}, nil
 }
 
 func TestOnBorrow(c *Connection, t time.Time) error {
@@ -112,9 +118,9 @@ func (c *Connection) ReadResponse(s *Session, token int64) (*p.Response, error)
 
 		if response.GetToken() == token {
 			return response, nil
-		} else if _, ok := s.cache[response.GetToken()]; ok {
+		} else if cursor, ok := s.checkCache(token); ok {
 			// Handle batch response
-			s.handleBatchResponse(response)
+			s.handleBatchResponse(cursor, response)
 		} else {
 			return nil, RqlDriverError{"Unexpected response received"}
 		}
@@ -153,6 +159,7 @@ func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]i
 
 	// Return immediately if the noreply option was set
 	if noreply, ok := opts["noreply"]; ok && noreply.(bool) {
+		c.Close()
 		return nil, nil
 	} else if async {
 		return nil, nil
@@ -185,17 +192,14 @@ func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]i
 	case p.Response_SUCCESS_PARTIAL, p.Response_SUCCESS_SEQUENCE, p.Response_SUCCESS_FEED:
 		cursor := &Cursor{
 			session: s,
+			conn:    c,
 			query:   q,
 			term:    t,
 			opts:    opts,
 			profile: profile,
-			timeout: -1,
 		}
-		cursor.gotReply.L = &cursor.mu
 
-		s.Lock()
-		s.cache[*q.Token] = cursor
-		s.Unlock()
+		s.setCache(*q.Token, cursor)
 
 		cursor.extend(response)
 
@@ -230,15 +234,14 @@ func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]i
 
 		cursor := &Cursor{
 			session:  s,
+			conn:     c,
 			query:    q,
 			term:     t,
 			opts:     opts,
 			profile:  profile,
 			buffer:   value,
 			finished: true,
-			timeout:  -1,
 		}
-		cursor.gotReply.L = &cursor.mu
 
 		return cursor, nil
 	case p.Response_WAIT_COMPLETE:
@@ -249,7 +252,19 @@ func (c *Connection) SendQuery(s *Session, q *p.Query, t Term, opts map[string]i
 }
 
 func (c *Connection) Close() error {
+	err := c.s.noreplyWaitQuery()
+	if err != nil {
+		return err
+	}
+
+	return c.CloseNoWait()
+}
+
+func (c *Connection) CloseNoWait() error {
+	c.Lock()
 	c.closed = true
+	c.Unlock()
+
 	return c.Conn.Close()
 }
 

doc.go

@@ -1,6 +1,6 @@
 // Go driver for RethinkDB
 //
-// Current version: v0.3 (RethinkDB v1.13)
+// Current version: v0.3.1 (RethinkDB v1.13)
 // For more in depth information on how to use RethinkDB check out the API docs
 // at http://rethinkdb.com/api
 package gorethink

query.go

@@ -202,18 +202,16 @@ func (t Term) RunWrite(s *Session, optArgs ...RunOpts) (WriteResponse, error) {
 	return response, err
 }
 
-// Exec runs the query but does not return the result (It also automatically sets
-// the noreply option).
+// Exec runs the query but does not return the result.
 func (t Term) Exec(s *Session, optArgs ...RunOpts) error {
-	// Ensure that noreply is set to true
-	if len(optArgs) >= 1 {
-		optArgs[0].NoReply = true
-	} else {
-		optArgs = append(optArgs, RunOpts{
-			NoReply: true,
-		})
+	res, err := t.Run(s, optArgs...)
+	if err != nil {
+		return err
+	}
+	err = res.Close()
+	if err != nil {
+		return err
 	}
 
-	_, err := t.Run(s, optArgs...)
-	return err
+	return nil
 }

query_control_test.go

@@ -1,6 +1,7 @@
 package gorethink
 
 import (
+	"testing"
 	"time"
 
 	test "launchpad.net/gocheck"
@@ -152,6 +153,10 @@ func (s *RethinkSuite) TestControlJs(c *test.C) {
 }
 
 func (s *RethinkSuite) TestControlHttp(c *test.C) {
+	if testing.Short() {
+		c.Skip("-short set")
+	}
+
 	var response map[string]interface{}
 	query := Http("httpbin.org/get?data=1")
 	res, err := query.Run(sess)
@@ -179,7 +184,7 @@ func (s *RethinkSuite) TestControlJson(c *test.C) {
 
 func (s *RethinkSuite) TestControlError(c *test.C) {
 	query := Error("An error occurred")
-	_, err := query.Run(sess)
+	err := query.Exec(sess)
 	c.Assert(err, test.NotNil)
 
 	c.Assert(err, test.NotNil)

query_select_test.go

@@ -1,6 +1,10 @@
 package gorethink
 
-import test "launchpad.net/gocheck"
+import (
+	"fmt"
+
+	test "launchpad.net/gocheck"
+)
 
 func (s *RethinkSuite) TestSelectGet(c *test.C) {
 	// Ensure table + database exist
@@ -300,3 +304,58 @@ func (s *RethinkSuite) TestSelectMany(c *test.C) {
 	c.Assert(res.Err(), test.IsNil)
 	c.Assert(n, test.Equals, 100)
 }
+
+func (s *RethinkSuite) TestSelectManyConcurrent(c *test.C) {
+	// Ensure table + database exist
+	DbCreate("test").RunWrite(sess)
+	Db("test").TableCreate("TestMany").RunWrite(sess)
+	Db("test").Table("TestMany").Delete().RunWrite(sess)
+
+	// Insert rows
+	for i := 0; i < 1; i++ {
+		data := []interface{}{}
+
+		for j := 0; j < 100; j++ {
+			data = append(data, map[string]interface{}{
+				"i": i,
+				"j": j,
+			})
+		}
+
+		Db("test").Table("TestMany").Insert(data).Run(sess)
+	}
+
+	// Test queries concurrently
+	attempts := 1
+	waitChannel := make(chan error, attempts)
+
+	for i := 0; i < attempts; i++ {
+		go func(i int, c chan error) {
+			res, err := Db("test").Table("TestMany").Run(sess, RunOpts{
+				BatchConf: map[string]interface{}{"max_els": 5, "max_size": 20},
+			})
+			if err != nil {
+				c <- err
+			}
+
+			var response []map[string]interface{}
+			err = res.All(&response)
+			if err != nil {
+				c <- err
+			}
+
+			if len(response) != 100 {
+				c <- fmt.Errorf("expected response length 100, received %d", len(response))
+			}
+
+			c <- nil
+		}(i, waitChannel)
+	}
+
+	for i := 0; i < attempts; i++ {
+		ret := <-waitChannel
+		if ret != nil {
+			c.Fatal("non-nil error returned (%s)", ret)
+		}
+	}
+}