Merge branch 'develop' into next

Author: gabor-boros

.travis.yml

@@ -6,6 +6,7 @@ go:
   - 1.9.x
   - 1.10.x
   - 1.11.x
+  - 1.12.x
   - master
 
 cache: apt

CHANGELOG.md

@@ -2,6 +2,10 @@
 All notable changes to this project will be documented in this file.
 This project adheres to [Semantic Versioning](http://semver.org/).
 
+## v5.0.1 - 2018-10-18
+
+- Fix noreply queries memory leak due unnecessary for responses
+
 ## v5.0.0 - 2018-09-12
 
 - Moved to rethinkdb organization

README.md

@@ -12,7 +12,7 @@ Current version: v5.1.0 (RethinkDB v2.4)
 
 Please note that this version of the driver only supports versions of RethinkDB using the v0.4 protocol (any versions of the driver older than RethinkDB 2.0 will not work).
 
-If you need any help you can find me on the [RethinkDB slack](http://slack.rethinkdb.com/) in the #gorethink channel.
+If you need any help you can find me on the [RethinkDB slack](https://rethinkdb.slack.com/) in the #gorethink channel.
 
 ## Installation
 
@@ -456,6 +456,43 @@ func TestSomething(t *testing.T) {
 }
 ```
 
+If you want the cursor to block on some of the response values, you can pass in
+a value of type `chan interface{}` and the cursor will block until a value is
+available to read on the channel.  Or you can pass in a function with signature
+`func() interface{}`: the cursor will call the function (which may block).  Here
+is the example above adapted to use a channel.
+
+```go
+func TestSomething(t *testing.T) {
+	mock := r.NewMock()
+	ch := make(chan []interface{})
+	mock.On(r.Table("people")).Return(ch, nil)
+	go func() {
+		ch <- []interface{}{
+			map[string]interface{}{"id": 1, "name": "John Smith"},
+			map[string]interface{}{"id": 2, "name": "Jane Smith"},
+		}
+		ch <- []interface{}{map[string]interface{}{"id": 3, "name": "Jack Smith"}}
+		close(ch)
+	}()
+	cursor, err := r.Table("people").Run(mock)
+	if err != nil {
+		t.Errorf("err is: %v", err)
+	}
+
+	var rows []interface{}
+	err = cursor.All(&rows)
+	if err != nil {
+		t.Errorf("err is: %v", err)
+	}
+
+	// Test result of rows
+
+	mock.AssertExpectations(t)
+}
+
+```
+
 The mocking implementation is based on amazing https://github.com/stretchr/testify library, thanks to @stretchr for their awesome work!
 
 ## Benchmarks
@@ -464,17 +501,17 @@ Everyone wants their project's benchmarks to be speedy. And while we know that R
 
 Thanks to @jaredfolkins for the contribution.
 
-| Type    |  Value   |
-| --- | --- |
-| **Model Name** | MacBook Pro |
-| **Model Identifier** | MacBookPro11,3 |
-| **Processor Name** | Intel Core i7 |
-| **Processor Speed** | 2.3 GHz |
-| **Number of Processors** | 1 |
-| **Total Number of Cores** | 4 |
-| **L2 Cache (per Core)** | 256 KB |
-| **L3 Cache** | 6 MB |
-| **Memory** | 16 GB |
+| Type                      | Value          |
+| ------------------------- | -------------- |
+| **Model Name**            | MacBook Pro    |
+| **Model Identifier**      | MacBookPro11,3 |
+| **Processor Name**        | Intel Core i7  |
+| **Processor Speed**       | 2.3 GHz        |
+| **Number of Processors**  | 1              |
+| **Total Number of Cores** | 4              |
+| **L2 Cache (per Core)**   | 256 KB         |
+| **L3 Cache**              | 6 MB           |
+| **Memory**                | 16 GB          |
 
 ```bash
 BenchmarkBatch200RandomWrites                20                              557227775                     ns/op

connection.go

@@ -209,12 +209,7 @@ func (c *Connection) Query(ctx context.Context, q Query) (*Response, *Cursor, er
 	}
 
 	if noreply, ok := q.Opts["noreply"]; ok && noreply.(bool) {
-		select {
-		case c.readRequestsChan <- tokenAndPromise{ctx: ctx, query: &q, span: fetchingSpan}:
-			return nil, nil, nil
-		case <-ctx.Done():
-			return c.stopQuery(&q)
-		}
+		return nil, nil, nil
 	}
 
 	promise := make(chan responseAndCursor, 1)
@@ -469,7 +464,7 @@ func (c *Connection) processResponse(ctx context.Context, q Query, response *Res
 	case p.Response_WAIT_COMPLETE:
 		return c.processWaitResponse(response)
 	default:
-		return nil, nil, RQLDriverError{rqlError("Unexpected response type: %v")}
+		return nil, nil, RQLDriverError{rqlError(fmt.Sprintf("Unexpected response type: %v", response.Type.String()))}
 	}
 }
 

connection_test.go

@@ -143,30 +143,6 @@ func (s *ConnectionSuite) TestConnection_Query_NoReplyOk(c *test.C) {
 	conn.AssertExpectations(c)
 }
 
-func (s *ConnectionSuite) TestConnection_Query_NoReplyTimeoutWrite(c *test.C) {
-	ctx, cancel := context.WithCancel(context.Background())
-	token := int64(1)
-	q := testQuery(DB("db").Table("table").Get("id"))
-	q.Opts["noreply"] = true
-	writeData := serializeQuery(token, q)
-	stopData := serializeQuery(token, newStopQuery(token))
-
-	conn := &connMock{}
-	conn.On("Write", writeData).Return(len(writeData), nil)
-	conn.On("Write", stopData).Return(len(stopData), nil)
-	conn.On("SetWriteDeadline").Return(nil)
-
-	connection := newConnection(conn, "addr", &ConnectOpts{ReadTimeout: time.Millisecond, WriteTimeout: time.Millisecond})
-	connection.readRequestsChan = make(chan tokenAndPromise, 0)
-	cancel()
-	response, cursor, err := connection.Query(ctx, q)
-
-	c.Assert(response, test.IsNil)
-	c.Assert(cursor, test.IsNil)
-	c.Assert(err, test.Equals, ErrQueryTimeout)
-	conn.AssertExpectations(c)
-}
-
 func (s *ConnectionSuite) TestConnection_Query_TimeoutWrite(c *test.C) {
 	ctx, cancel := context.WithCancel(context.Background())
 	token := int64(1)

cursor.go

@@ -227,7 +227,6 @@ func (c *Cursor) nextLocked(dest interface{}, progressCursor bool) (bool, error)
 			if progressCursor {
 				c.buffer = c.buffer[1:]
 			}
-
 			err := encoding.Decode(dest, data)
 			if err != nil {
 				return false, err
@@ -494,7 +493,6 @@ func (c *Cursor) Listen(channel interface{}) {
 			if !c.Next(elemp.Interface()) {
 				break
 			}
-
 			channelv.Send(elemp.Elem())
 		}
 

cursor_test.go

@@ -0,0 +1,34 @@
+package rethinkdb
+
+import (
+	test "gopkg.in/check.v1"
+	"gopkg.in/rethinkdb/rethinkdb-go.v5/internal/integration/tests"
+)
+
+type CursorSuite struct{}
+
+var _ = test.Suite(&CursorSuite{})
+
+func (s *CursorSuite) TestCursor_One_Ok(c *test.C) {
+	data := map[string]interface{}{
+		"A": 1,
+		"B": true,
+	}
+
+	mock := NewMock()
+	ch := make(chan []interface{})
+	mock.On(DB("test").Table("test")).Return(ch, nil)
+	go func() {
+		ch <- []interface{}{data}
+		close(ch)
+	}()
+	res, err := DB("test").Table("test").Run(mock)
+	c.Assert(err, test.IsNil)
+
+	var response interface{}
+	err = res.One(&response)
+
+	c.Assert(err, test.IsNil)
+	c.Assert(response, tests.JsonEquals, data)
+	mock.AssertExpectations(c)
+}

encoding/decoder_test.go

@@ -433,7 +433,7 @@ func TestDecodeUnmarshalerPointer(t *testing.T) {
 		t.Errorf("got error %v, expected nil", err)
 	}
 	if !jsonEqual(out, want) {
-		t.Errorf("got %q, want %q", out, want)
+		t.Errorf("got %+v, want %+v", out, want)
 	}
 }
 
@@ -587,7 +587,7 @@ func TestDecodeCustomTypeEncodingPointer(t *testing.T) {
 		t.Errorf("got error %v, expected nil", err)
 	}
 	if !jsonEqual(out, want) {
-		t.Errorf("got %q, want %q", out, want)
+		t.Errorf("got %+v, want %+v", out, want)
 	}
 }