Merge pull request #460 from arnodel/master

Add ability to block cursors when testing with mocks

Author: CMogilko

README.md

@@ -12,7 +12,7 @@ Current version: v5.0.1 (RethinkDB v2.3)
 
 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,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

query_control.go

@@ -223,8 +223,8 @@ func Error(args ...interface{}) Term {
 	return constructRootTerm("Error", p.Term_ERROR, args, map[string]interface{}{})
 }
 
-// Args is a special term usd to splice an array of arguments into another term.
-// This is useful when you want to call a varadic term such as GetAll with a set
+// Args is a special term used to splice an array of arguments into another term.
+// This is useful when you want to call a variadic term such as GetAll with a set
 // of arguments provided at runtime.
 func Args(args ...interface{}) Term {
 	return constructRootTerm("Args", p.Term_ARGS, args, map[string]interface{}{})