remove socket timeouts

Author: CMogilko

connection.go

@@ -378,11 +378,6 @@ func (c *Connection) sendQuery(q Query) error {
 	binary.LittleEndian.PutUint64(b, uint64(q.Token))
 	binary.LittleEndian.PutUint32(b[8:], uint32(len(b)-respHeaderLen))
 
-	// Set timeout
-	if c.opts.WriteTimeout != 0 {
-		c.Conn.SetWriteDeadline(time.Now().Add(c.opts.WriteTimeout))
-	}
-
 	// Send the JSON encoding of the query itself.
 	if err = c.writeData(b); err != nil {
 		c.setBad()
@@ -402,10 +397,8 @@ func (c *Connection) nextToken() int64 {
 // readResponse attempts to read a Response from the server, if no response
 // could be read then an error is returned.
 func (c *Connection) readResponse() (*Response, error) {
-	// Set timeout
-	if c.opts.ReadTimeout != 0 {
-		c.Conn.SetReadDeadline(time.Now().Add(c.opts.ReadTimeout))
-	}
+	// due to this is pooled connection, it always reads from socket even if idle
+	// timeouts should be only on query-level with context
 
 	// Read response header (token+length)
 	headerBuf := [respHeaderLen]byte{}

connection_helper.go

@@ -17,10 +17,14 @@ func (c *Connection) read(buf []byte) (total int, err error) {
 }
 
 func (c *Connection) contextFromConnectionOpts() context.Context {
-	sum := c.opts.ReadTimeout + c.opts.WriteTimeout
-	if c.opts.ReadTimeout == 0 || c.opts.WriteTimeout == 0 {
+	// back compatibility
+	min := c.opts.ReadTimeout
+	if c.opts.WriteTimeout < min {
+		min = c.opts.WriteTimeout
+	}
+	if min == 0 {
 		return context.Background()
 	}
-	ctx, _ := context.WithTimeout(context.Background(), sum)
+	ctx, _ := context.WithTimeout(context.Background(), min)
 	return ctx
 }

connection_test.go

@@ -25,9 +25,9 @@ func (s *ConnectionSuite) TestConnection_Query_Ok(c *test.C) {
 	header := respHeader(token, respData)
 
 	conn := &connMock{}
-	conn.On("Write", writeData).Return(len(writeData), nil)
-	conn.On("Read", respHeaderLen).Return(header, respHeaderLen, nil)
-	conn.On("Read", len(respData)).Return(respData, len(respData), nil)
+	conn.On("Write", writeData).Return(len(writeData), nil, nil)
+	conn.On("Read", respHeaderLen).Return(header, respHeaderLen, nil, nil)
+	conn.On("Read", len(respData)).Return(respData, len(respData), nil, nil)
 	conn.On("Close").Return(nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{})
@@ -60,9 +60,9 @@ func (s *ConnectionSuite) TestConnection_Query_DefaultDBOk(c *test.C) {
 	header := respHeader(token, respData)
 
 	conn := &connMock{}
-	conn.On("Write", writeData).Return(len(writeData), nil)
-	conn.On("Read", respHeaderLen).Return(header, respHeaderLen, nil)
-	conn.On("Read", len(respData)).Return(respData, len(respData), nil)
+	conn.On("Write", writeData).Return(len(writeData), nil, nil)
+	conn.On("Read", respHeaderLen).Return(header, respHeaderLen, nil, nil)
+	conn.On("Read", len(respData)).Return(respData, len(respData), nil, nil)
 	conn.On("Close").Return(nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{Database: "db"})
@@ -106,7 +106,7 @@ func (s *ConnectionSuite) TestConnection_Query_SendFail(c *test.C) {
 	writeData := serializeQuery(token, q)
 
 	conn := &connMock{}
-	conn.On("Write", writeData).Return(0, io.EOF)
+	conn.On("Write", writeData).Return(0, io.EOF, nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{})
 	response, cursor, err := connection.Query(ctx, q)
@@ -126,9 +126,9 @@ func (s *ConnectionSuite) TestConnection_Query_NoReplyOk(c *test.C) {
 	header := respHeader(token, respData)
 
 	conn := &connMock{}
-	conn.On("Write", writeData).Return(len(writeData), nil)
-	conn.On("Read", respHeaderLen).Return(header, respHeaderLen, nil)
-	conn.On("Read", len(respData)).Return(respData, len(respData), nil)
+	conn.On("Write", writeData).Return(len(writeData), nil, nil)
+	conn.On("Read", respHeaderLen).Return(header, respHeaderLen, nil, nil)
+	conn.On("Read", len(respData)).Return(respData, len(respData), nil, nil)
 	conn.On("Close").Return(nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{})
@@ -151,9 +151,8 @@ func (s *ConnectionSuite) TestConnection_Query_TimeoutWrite(c *test.C) {
 	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)
+	conn.On("Write", writeData).Return(len(writeData), nil, nil)
+	conn.On("Write", stopData).Return(len(stopData), nil, nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{ReadTimeout: time.Millisecond, WriteTimeout: time.Millisecond})
 	connection.readRequestsChan = make(chan tokenAndPromise, 0)
@@ -174,9 +173,8 @@ func (s *ConnectionSuite) TestConnection_Query_TimeoutRead(c *test.C) {
 	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)
+	conn.On("Write", writeData).Return(len(writeData), nil, 10*time.Millisecond)
+	conn.On("Write", stopData).Return(len(stopData), nil, nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{ReadTimeout: time.Millisecond, WriteTimeout: time.Millisecond})
 	response, cursor, err := connection.Query(ctx, q)
@@ -196,7 +194,7 @@ func (s *ConnectionSuite) TestConnection_Query_SendFailTracing(c *test.C) {
 	writeData := serializeQuery(token, q)
 
 	conn := &connMock{}
-	conn.On("Write", writeData).Return(0, io.EOF)
+	conn.On("Write", writeData).Return(0, io.EOF, nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{UseOpentracing: true})
 	response, cursor, err := connection.Query(ctx, q)
@@ -306,8 +304,7 @@ func (s *ConnectionSuite) TestConnection_readResponse_TimeoutHeader(c *test.C) {
 	timeout := time.Second
 
 	conn := &connMock{}
-	conn.On("SetReadDeadline").Return(nil)
-	conn.On("Read", respHeaderLen).Return(nil, 0, io.EOF)
+	conn.On("Read", respHeaderLen).Return(nil, 0, io.EOF, nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{ReadTimeout: timeout})
 
@@ -325,8 +322,8 @@ func (s *ConnectionSuite) TestConnection_readResponse_BodySocketErr(c *test.C) {
 	header := respHeader(token, respData)
 
 	conn := &connMock{}
-	conn.On("Read", respHeaderLen).Return(header, len(header), nil)
-	conn.On("Read", len(respData)).Return(nil, 0, io.EOF)
+	conn.On("Read", respHeaderLen).Return(header, len(header), nil, nil)
+	conn.On("Read", len(respData)).Return(nil, 0, io.EOF, nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{})
 
@@ -344,8 +341,8 @@ func (s *ConnectionSuite) TestConnection_readResponse_BodyUnmarshalErr(c *test.C
 	header := respHeader(token, respData)
 
 	conn := &connMock{}
-	conn.On("Read", respHeaderLen).Return(header, len(header), nil)
-	conn.On("Read", len(respData)).Return(make([]byte, len(respData)), len(respData), nil)
+	conn.On("Read", respHeaderLen).Return(header, len(header), nil, nil)
+	conn.On("Read", len(respData)).Return(make([]byte, len(respData)), len(respData), nil, nil)
 
 	connection := newConnection(conn, "addr", &ConnectOpts{})
 

go.mod

@@ -1,21 +1,28 @@
 module gopkg.in/rethinkdb/rethinkdb-go.v6
 
 require (
-	github.com/cenkalti/backoff v2.0.0+incompatible
+	github.com/bitly/go-hostpool v0.1.0 // indirect
+	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
+	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/golang/protobuf v1.2.0
+	github.com/golang/protobuf v1.3.4
 	github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed
 	github.com/kr/pretty v0.1.0 // indirect
-	github.com/opentracing/opentracing-go v1.0.2
-	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
+	github.com/onsi/ginkgo v1.12.0 // indirect
+	github.com/onsi/gomega v1.9.0 // indirect
+	github.com/opentracing/opentracing-go v1.1.0
 	github.com/sirupsen/logrus v1.0.6
-	github.com/stretchr/objx v0.1.1 // indirect
-	github.com/stretchr/testify v1.2.2
-	golang.org/x/crypto v0.0.0-20180820150726-614d502a4dac
-	golang.org/x/net v0.0.0-20180826012351-8a410e7b638d
-	golang.org/x/sys v0.0.0-20180828065106-d99a578cf41b // indirect
-	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127
-	gopkg.in/fatih/pool.v2 v2.0.0
+	github.com/stretchr/objx v0.2.0 // indirect
+	github.com/stretchr/testify v1.5.1
+	golang.org/x/crypto v0.0.0-20200302210943-78000ba7a073
+	golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3
+	golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e // indirect
+	gopkg.in/airbrake/gobrake.v2 v2.0.9 // indirect
+	gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f
+	gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2 // indirect
+	gopkg.in/yaml.v2 v2.2.8 // indirect
 )
 
-go 1.13
+go 1.14

mock_test.go

@@ -371,6 +371,54 @@ func (s *MockSuite) TestMockRethinkStructsRunWrite(c *test.C) {
 	mock.AssertExpectations(c)
 }
 
+func (s *MockSuite) TestMockMapSliceResultOk(c *test.C) {
+	type Some struct {
+		Id string
+	}
+
+	result := []map[string]interface{}{
+		{"Id": "test1"},
+		{"Id": "test2"},
+	}
+
+	mock := NewMock()
+	q := DB("test").Table("test").GetAll()
+	mock.On(q).Return(result, nil)
+	res, err := q.Run(mock)
+	c.Assert(err, test.IsNil)
+
+	var casted []*Some
+	err = res.All(&casted)
+	c.Assert(err, test.IsNil)
+
+	c.Assert(casted[0].Id, test.Equals, "test1")
+	c.Assert(casted[1].Id, test.Equals, "test2")
+}
+
+func (s *MockSuite) TestMockPointerSliceResultOk(c *test.C) {
+	type Some struct {
+		Id string
+	}
+
+	result := []*Some{
+		{Id: "test1"},
+		{Id: "test2"},
+	}
+
+	mock := NewMock()
+	q := DB("test").Table("test").GetAll()
+	mock.On(q).Return(result, nil)
+	res, err := q.Run(mock)
+	c.Assert(err, test.IsNil)
+
+	var casted []*Some
+	err = res.All(&casted)
+	c.Assert(err, test.IsNil)
+
+	c.Assert(casted[0].Id, test.Equals, "test1")
+	c.Assert(casted[1].Id, test.Equals, "test2")
+}
+
 type simpleTestingT struct {
 	failed bool
 }

mocks_test.go

@@ -16,11 +16,19 @@ func (m *connMock) Read(b []byte) (n int, err error) {
 	if ok {
 		copy(b, rbuf)
 	}
+	timeout := args.Get(3)
+	if timeout != nil {
+		time.Sleep(timeout.(time.Duration))
+	}
 	return args.Int(1), args.Error(2)
 }
 
 func (m *connMock) Write(b []byte) (n int, err error) {
 	args := m.Called(b)
+	timeout := args.Get(2)
+	if timeout != nil {
+		time.Sleep(timeout.(time.Duration))
+	}
 	return args.Int(0), args.Error(1)
 }
 

pool.go

@@ -6,7 +6,6 @@ import (
 	"sync/atomic"
 
 	"golang.org/x/net/context"
-	"gopkg.in/fatih/pool.v2"
 )
 
 var (
@@ -59,7 +58,7 @@ func NewPool(host Host, opts *ConnectOpts) (*Pool, error) {
 // Ping verifies a connection to the database is still alive,
 // establishing a connection if necessary.
 func (p *Pool) Ping() error {
-	_, _, err := p.conn()
+	_, err := p.conn()
 	return err
 }
 
@@ -84,12 +83,12 @@ func (p *Pool) Close() error {
 	return nil
 }
 
-func (p *Pool) conn() (*Connection, *pool.PoolConn, error) {
+func (p *Pool) conn() (*Connection, error) {
 	p.mu.RLock()
 
 	if p.closed {
 		p.mu.RUnlock()
-		return nil, nil, errPoolClosed
+		return nil, errPoolClosed
 	}
 	p.mu.RUnlock()
 
@@ -99,7 +98,7 @@ func (p *Pool) conn() (*Connection, *pool.PoolConn, error) {
 	}
 	pos = pos % int32(len(p.conns))
 
-	return p.conns[pos], nil, nil
+	return p.conns[pos], nil
 }
 
 // SetInitialPoolCap sets the initial capacity of the connection pool.
@@ -128,7 +127,7 @@ func (p *Pool) SetMaxOpenConns(n int) {
 
 // Exec executes a query without waiting for any response.
 func (p *Pool) Exec(ctx context.Context, q Query) error {
-	c, _, err := p.conn()
+	c, err := p.conn()
 	if err != nil {
 		return err
 	}
@@ -139,7 +138,7 @@ func (p *Pool) Exec(ctx context.Context, q Query) error {
 
 // Query executes a query and waits for the response
 func (p *Pool) Query(ctx context.Context, q Query) (*Cursor, error) {
-	c, _, err := p.conn()
+	c, err := p.conn()
 	if err != nil {
 		return nil, err
 	}
@@ -152,7 +151,7 @@ func (p *Pool) Query(ctx context.Context, q Query) (*Cursor, error) {
 func (p *Pool) Server() (ServerResponse, error) {
 	var response ServerResponse
 
-	c, _, err := p.conn()
+	c, err := p.conn()
 	if err != nil {
 		return response, err
 	}

rethinkdb.go

@@ -10,6 +10,8 @@ import (
 )
 
 var (
+	// Log is logger for debug purpuses.
+	// deprecated
 	Log *logrus.Logger
 )
 

session.go

@@ -46,9 +46,11 @@ type ConnectOpts struct {
 	Timeout time.Duration `rethinkdb:"timeout,omitempty" json:"timeout,omitempty"`
 	// WriteTimeout is the amount of time the driver will wait when sending the
 	// query to the server
+	// Deprecated: use RunOpts.Context instead
 	WriteTimeout time.Duration `rethinkdb:"write_timeout,omitempty" json:"write_timeout,omitempty"`
 	// ReadTimeout is the amount of time the driver will wait for a response from
 	// the server when executing queries.
+	// Deprecated: use RunOpts.Context instead
 	ReadTimeout time.Duration `rethinkdb:"read_timeout,omitempty" json:"read_timeout,omitempty"`
 	// KeepAlivePeriod is the keep alive period used by the connection, by default
 	// this is 30s. It is not possible to disable keep alive messages