Skip to content

Commit 891125e

Browse files
CMogilkoCMogilko
committed
remove mocking from cursor
1 parent 117ecc7 commit 891125e

File tree

4 files changed

+124
-61
lines changed

4 files changed

+124
-61
lines changed

‎README.md‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -465,11 +465,15 @@ is the example above adapted to use a channel.
465465
```go
466466
func TestSomething(t *testing.T) {
467467
mock := r.NewMock()
468-
ch := make(chan interface{})
469-
mock.On(r.Table("people")).Return([]interface{}{ch, ch}, nil)
468+
ch := make(chan []interface{})
469+
mock.On(r.Table("people")).Return(ch, nil)
470470
go func() {
471-
ch <- map[string]interface{}{"id": 1, "name": "John Smith"}
472-
ch <- map[string]interface{}{"id": 2, "name": "Jane Smith"}
471+
ch <- []interface{}{
472+
map[string]interface{}{"id": 1, "name": "John Smith"},
473+
map[string]interface{}{"id": 2, "name": "Jane Smith"},
474+
}
475+
ch <- []interface{}{map[string]interface{}{"id": 3, "name": "Jack Smith"}}
476+
close(ch)
473477
}()
474478
cursor, err := r.Table("people").Run(mock)
475479
if err != nil {

‎cursor.go‎

Lines changed: 0 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -208,13 +208,6 @@ func (c *Cursor) Next(dest interface{}) bool {
208208
return hasMore
209209
}
210210

211-
// This can be used for testing. Typically the function would block to simulate
212-
// network lag and test concurrency.
213-
type delayedData struct {
214-
f func() interface{} // This function produces the data
215-
data interface{} // This is initially nil, then equal to f()
216-
}
217-
218211
func (c *Cursor) nextLocked(dest interface{}, progressCursor bool) (bool, error) {
219212
for {
220213
if err := c.seekCursor(true); err != nil {
@@ -234,17 +227,6 @@ func (c *Cursor) nextLocked(dest interface{}, progressCursor bool) (bool, error)
234227
if progressCursor {
235228
c.buffer = c.buffer[1:]
236229
}
237-
if dd, ok := data.(delayedData); ok {
238-
if dd.data == nil {
239-
240-
// Here we remove the lock as in c.fetchMore() because the
241-
// function is likely to block.
242-
c.mu.Unlock()
243-
dd.data = dd.f()
244-
c.mu.Lock()
245-
}
246-
data = dd.data
247-
}
248230
err := encoding.Decode(dest, data)
249231
if err != nil {
250232
return false, err

‎mock.go‎

Lines changed: 103 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,11 @@
11
package rethinkdb
22

33
import (
4+
"encoding/binary"
45
"encoding/json"
56
"fmt"
7+
"gopkg.in/check.v1"
8+
"net"
69
"reflect"
710
"sync"
811
"time"
@@ -338,43 +341,41 @@ func (m *Mock) Query(ctx context.Context, q Query) (*Cursor, error) {
338341
return nil, query.Error
339342
}
340343

344+
var conn *Connection = nil
345+
responseVal := reflect.ValueOf(query.Response)
346+
if responseVal.Kind() == reflect.Chan || responseVal.Kind() == reflect.Func {
347+
conn = newConnection(newMockConn(query.Response), "mock", &ConnectOpts{})
348+
349+
query.Query.Type = p.Query_CONTINUE
350+
query.Query.Token = conn.nextToken()
351+
352+
conn.runConnection()
353+
}
354+
355+
if ctx == nil {
356+
ctx = context.Background()
357+
}
358+
341359
// Build cursor and return
342-
c := newCursor(ctx, nil, "", query.Query.Token, query.Query.Term, query.Query.Opts)
360+
c := newCursor(ctx, conn, "", query.Query.Token, query.Query.Term, query.Query.Opts)
343361
c.finished = true
344362
c.fetching = false
345363
c.isAtom = true
346364

347-
responseVal := reflect.ValueOf(query.Response)
348365
if responseVal.Kind() == reflect.Slice || responseVal.Kind() == reflect.Array {
349366
for i := 0; i < responseVal.Len(); i++ {
350-
c.buffer = append(c.buffer, getMockValue(responseVal.Index(i).Interface()))
367+
c.buffer = append(c.buffer, responseVal.Index(i).Interface())
351368
}
369+
} else if conn != nil {
370+
conn.cursors[query.Query.Token] = c
371+
c.finished = false
352372
} else {
353-
c.buffer = append(c.buffer, getMockValue(query.Response))
373+
c.buffer = append(c.buffer, query.Response)
354374
}
355375

356376
return c, nil
357377
}
358378

359-
// getMockValue turns some responses to delayedData: values of "chan
360-
// interface{}" type will turn to delayed data that produce data when there is
361-
// an element available on the channel. Values of "func() interface{}" type
362-
// will produce data by calling the function.
363-
func getMockValue(val interface{}) interface{} {
364-
switch v := val.(type) {
365-
case chan interface{}:
366-
return delayedData{
367-
f: func() interface{} { return <-v },
368-
}
369-
case func() interface{}:
370-
return delayedData{
371-
f: v,
372-
}
373-
default:
374-
return val
375-
}
376-
}
377-
378379
func (m *Mock) Exec(ctx context.Context, q Query) error {
379380
_, err := m.Query(ctx, q)
380381

@@ -422,3 +423,82 @@ func (m *Mock) queries() []MockQuery {
422423
defer m.mu.Unlock()
423424
return append([]MockQuery{}, m.Queries...)
424425
}
426+
427+
type mockConn struct {
428+
c *check.C
429+
mu sync.Mutex
430+
value []byte
431+
tokens chan int64
432+
valueGetter func() []interface{}
433+
}
434+
435+
func newMockConn(responseGetter interface{}) *mockConn {
436+
c := &mockConn{tokens: make(chan int64, 1)}
437+
switch g := responseGetter.(type) {
438+
case chan []interface{}:
439+
c.valueGetter = func() []interface{} { return <-g }
440+
case func() []interface{}:
441+
c.valueGetter = g
442+
default:
443+
panic(fmt.Sprintf("unsupported value generator type: %T", responseGetter))
444+
}
445+
return c
446+
}
447+
448+
func (c *mockConn) Read(b []byte) (n int, err error) {
449+
c.mu.Lock()
450+
defer c.mu.Unlock()
451+
452+
if c.value == nil {
453+
values := c.valueGetter()
454+
455+
jresps := make([]json.RawMessage, len(values))
456+
for i := range values {
457+
jresps[i], err = json.Marshal(values[i])
458+
if err != nil {
459+
panic(fmt.Sprintf("failed to encode response: %v", err))
460+
}
461+
}
462+
463+
token := <-c.tokens
464+
resp := Response{
465+
Token: token,
466+
Responses: jresps,
467+
Type: p.Response_SUCCESS_PARTIAL,
468+
}
469+
if values == nil {
470+
resp.Type = p.Response_SUCCESS_SEQUENCE
471+
}
472+
473+
c.value, err = json.Marshal(resp)
474+
if err != nil {
475+
panic(fmt.Sprintf("failed to encode response: %v", err))
476+
}
477+
478+
if len(b) != respHeaderLen {
479+
panic("wrong header len")
480+
}
481+
binary.LittleEndian.PutUint64(b[:8], uint64(token))
482+
binary.LittleEndian.PutUint32(b[8:], uint32(len(c.value)))
483+
return len(b), nil
484+
} else {
485+
copy(b, c.value)
486+
c.value = nil
487+
return len(b), nil
488+
}
489+
}
490+
491+
func (c *mockConn) Write(b []byte) (n int, err error) {
492+
if len(b) < 8 {
493+
panic("bad socket write")
494+
}
495+
token := int64(binary.LittleEndian.Uint64(b[:8]))
496+
c.tokens <- token
497+
return len(b), nil
498+
}
499+
func (c *mockConn) Close() error { panic("not implemented") }
500+
func (c *mockConn) LocalAddr() net.Addr { panic("not implemented") }
501+
func (c *mockConn) RemoteAddr() net.Addr { panic("not implemented") }
502+
func (c *mockConn) SetDeadline(t time.Time) error { panic("not implemented") }
503+
func (c *mockConn) SetReadDeadline(t time.Time) error { panic("not implemented") }
504+
func (c *mockConn) SetWriteDeadline(t time.Time) error { panic("not implemented") }

‎mock_test.go‎

Lines changed: 13 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -57,8 +57,6 @@ func (s *MockSuite) TestMockRunSuccessSingleResult(c *test.C) {
5757
c.Assert(err, test.IsNil)
5858
c.Assert(response, tests.JsonEquals, map[string]interface{}{"id": "mocked"})
5959
mock.AssertExpectations(c)
60-
61-
res.Close()
6260
}
6361

6462
func (s *MockSuite) TestMockRunSuccessMultipleResults(c *test.C) {
@@ -76,17 +74,17 @@ func (s *MockSuite) TestMockRunSuccessMultipleResults(c *test.C) {
7674
c.Assert(err, test.IsNil)
7775
c.Assert(response, tests.JsonEquals, []interface{}{map[string]interface{}{"id": "mocked"}})
7876
mock.AssertExpectations(c)
79-
80-
res.Close()
8177
}
8278

8379
func (s *MockSuite) TestMockRunSuccessChannel(c *test.C) {
8480
mock := NewMock()
85-
ch := make(chan interface{})
86-
mock.On(DB("test").Table("test")).Return([]interface{}{ch, ch}, nil)
81+
ch := make(chan []interface{})
82+
mock.On(DB("test").Table("test")).Return(ch, nil)
8783
go func() {
88-
ch <- 1
89-
ch <- 2
84+
ch <- []interface{}{1, 2}
85+
ch <- []interface{}{3}
86+
ch <- []interface{}{4}
87+
close(ch)
9088
}()
9189
res, err := DB("test").Table("test").Run(mock)
9290
c.Assert(err, test.IsNil)
@@ -95,20 +93,21 @@ func (s *MockSuite) TestMockRunSuccessChannel(c *test.C) {
9593
err = res.All(&response)
9694

9795
c.Assert(err, test.IsNil)
98-
c.Assert(response, tests.JsonEquals, []interface{}{1, 2})
96+
c.Assert(response, tests.JsonEquals, []interface{}{1, 2, 3, 4})
9997
mock.AssertExpectations(c)
100-
101-
res.Close()
10298
}
10399

104100
func (s *MockSuite) TestMockRunSuccessFunction(c *test.C) {
105101
mock := NewMock()
106102
n := 0
107-
f := func() interface{} {
103+
f := func() []interface{} {
108104
n++
109-
return n
105+
if n == 4 {
106+
return nil
107+
}
108+
return []interface{}{n}
110109
}
111-
mock.On(DB("test").Table("test")).Return([]interface{}{f, f, 3}, nil)
110+
mock.On(DB("test").Table("test")).Return(f, nil)
112111
res, err := DB("test").Table("test").Run(mock)
113112
c.Assert(err, test.IsNil)
114113

@@ -118,8 +117,6 @@ func (s *MockSuite) TestMockRunSuccessFunction(c *test.C) {
118117
c.Assert(err, test.IsNil)
119118
c.Assert(response, tests.JsonEquals, []interface{}{1, 2, 3})
120119
mock.AssertExpectations(c)
121-
122-
res.Close()
123120
}
124121

125122
func (s *MockSuite) TestMockRunSuccessMultipleResults_type(c *test.C) {

0 commit comments

Comments
 (0)