Merge branch 'develop' of https://github.com/rethinkdb/rethinkdb-go into develop

Author: CMogilko

README.md

@@ -456,6 +456,39 @@ 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([]interface{}{ch, ch}, nil)
+	go func() {
+		ch <- map[string]interface{}{"id": 1, "name": "John Smith"}
+		ch <- map[string]interface{}{"id": 2, "name": "Jane Smith"}
+	}()
+	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 +497,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

cursor.go

@@ -208,6 +208,13 @@ func (c *Cursor) Next(dest interface{}) bool {
 	return hasMore
 }
 
+// This can be used for testing.  Typically the function would block to simulate
+// network lag and test concurrency.
+type delayedData struct {
+	f    func() interface{} // This function produces the data
+	data interface{}        // This is initially nil, then equal to f()
+}
+
 func (c *Cursor) nextLocked(dest interface{}, progressCursor bool) (bool, error) {
 	for {
 		if err := c.seekCursor(true); err != nil {
@@ -227,7 +234,17 @@ func (c *Cursor) nextLocked(dest interface{}, progressCursor bool) (bool, error)
 			if progressCursor {
 				c.buffer = c.buffer[1:]
 			}
-
+			if dd, ok := data.(delayedData); ok {
+				if dd.data == nil {
+
+					// Here we remove the lock as in c.fetchMore() because the
+					// function is likely to block.
+					c.mu.Unlock()
+					dd.data = dd.f()
+					c.mu.Lock()
+				}
+				data = dd.data
+			}
 			err := encoding.Decode(dest, data)
 			if err != nil {
 				return false, err
@@ -494,7 +511,6 @@ func (c *Cursor) Listen(channel interface{}) {
 			if !c.Next(elemp.Interface()) {
 				break
 			}
-
 			channelv.Send(elemp.Elem())
 		}
 

mock.go

@@ -103,6 +103,16 @@ func (mq *MockQuery) unlock() {
 // Return specifies the return arguments for the expectation.
 //
 //    mock.On(r.Table("test")).Return(nil, errors.New("failed"))
+//
+// values of "chan interface{}" type will turn to delayed data that produce data
+// when there is an element available on the channel.  Values of "func()
+// interface{}" type will produce data by calling the function. E.g.
+//
+//    f := func() interface{} { return 2 }
+//    ch := make(chan interface{})
+//    mock.On(r.Table("test")).Return([]interface{}{ch, f, 3})
+//
+//    Running the query above will block until a value is pushed onto ch.
 func (mq *MockQuery) Return(response interface{}, err error) *MockQuery {
 	mq.lock()
 	defer mq.unlock()
@@ -337,15 +347,34 @@ func (m *Mock) Query(ctx context.Context, q Query) (*Cursor, error) {
 	responseVal := reflect.ValueOf(query.Response)
 	if responseVal.Kind() == reflect.Slice || responseVal.Kind() == reflect.Array {
 		for i := 0; i < responseVal.Len(); i++ {
-			c.buffer = append(c.buffer, responseVal.Index(i).Interface())
+			c.buffer = append(c.buffer, getMockValue(responseVal.Index(i).Interface()))
 		}
 	} else {
-		c.buffer = append(c.buffer, query.Response)
+		c.buffer = append(c.buffer, getMockValue(query.Response))
 	}
 
 	return c, nil
 }
 
+// getMockValue turns some responses to delayedData:  values of "chan
+// interface{}" type will turn to delayed data that produce data when there is
+// an element available on the channel.  Values of "func() interface{}" type
+// will produce data by calling the function.
+func getMockValue(val interface{}) interface{} {
+	switch v := val.(type) {
+	case chan interface{}:
+		return delayedData{
+			f: func() interface{} { return <-v },
+		}
+	case func() interface{}:
+		return delayedData{
+			f: v,
+		}
+	default:
+		return val
+	}
+}
+
 func (m *Mock) Exec(ctx context.Context, q Query) error {
 	_, err := m.Query(ctx, q)
 

mock_test.go

@@ -3,9 +3,10 @@ package rethinkdb
 import (
 	"fmt"
 
+	"testing"
+
 	test "gopkg.in/check.v1"
 	"gopkg.in/rethinkdb/rethinkdb-go.v5/internal/integration/tests"
-	"testing"
 )
 
 // Hook up gocheck into the gotest runner.
@@ -79,6 +80,48 @@ func (s *MockSuite) TestMockRunSuccessMultipleResults(c *test.C) {
 	res.Close()
 }
 
+func (s *MockSuite) TestMockRunSuccessChannel(c *test.C) {
+	mock := NewMock()
+	ch := make(chan interface{})
+	mock.On(DB("test").Table("test")).Return([]interface{}{ch, ch}, nil)
+	go func() {
+		ch <- 1
+		ch <- 2
+	}()
+	res, err := DB("test").Table("test").Run(mock)
+	c.Assert(err, test.IsNil)
+
+	var response []interface{}
+	err = res.All(&response)
+
+	c.Assert(err, test.IsNil)
+	c.Assert(response, tests.JsonEquals, []interface{}{1, 2})
+	mock.AssertExpectations(c)
+
+	res.Close()
+}
+
+func (s *MockSuite) TestMockRunSuccessFunction(c *test.C) {
+	mock := NewMock()
+	n := 0
+	f := func() interface{} {
+		n++
+		return n
+	}
+	mock.On(DB("test").Table("test")).Return([]interface{}{f, f, 3}, nil)
+	res, err := DB("test").Table("test").Run(mock)
+	c.Assert(err, test.IsNil)
+
+	var response []interface{}
+	err = res.All(&response)
+
+	c.Assert(err, test.IsNil)
+	c.Assert(response, tests.JsonEquals, []interface{}{1, 2, 3})
+	mock.AssertExpectations(c)
+
+	res.Close()
+}
+
 func (s *MockSuite) TestMockRunSuccessMultipleResults_type(c *test.C) {
 	type document struct {
 		Id string