refactor: replace logrus and cleanup dead code (#508)

Author: majst01

CHANGELOG.md

@@ -110,7 +110,7 @@ Thanks to everybody who got involved with this project over the last ~4 years an
 ### Changed
 
  - Moved project to `gorethink` organisation
- - Fixed behaviour when unmarshaling nil slices
+ - Fixed behaviour when unmarshalling nil slices
 
 ### Fixed
 
@@ -356,7 +356,7 @@ r.Connect(
  - Added more documentation and examples for `GetAll`.
 
 ### Fixed
-- Fixed `RunWrite` not defering its call to `Cursor.Close()`. This could cause issues if an error occurred when decoding the result.
+- Fixed `RunWrite` not deferring its call to `Cursor.Close()`. This could cause issues if an error occurred when decoding the result.
 - Fixed panic when calling `Error()` on a RethinkDB-go `rqlError`.
 
 ## v1.3.0 - 2016-01-11
@@ -557,7 +557,7 @@ For more details checkout the [README](https://github.com/gorethink/gorethink/bl
 ## v0.6.1 - 2015-02-13
 
 - Reduce GC by using buffers when reading and writing
-- Fixed encoding `time.Time` ignoring millseconds
+- Fixed encoding `time.Time` ignoring milliseconds
 - Fixed pointers in structs that implement the `Marshaler`/`Unmarshaler` interfaces being ignored
 
 ## v0.6.0 - 2015-01-01

cluster.go

@@ -3,16 +3,16 @@ package rethinkdb
 import (
 	"errors"
 	"fmt"
+	"log/slog"
 	"sort"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	"context"
+
+	"github.com/cenkalti/backoff/v4"
 	"github.com/hailocab/go-hostpool"
-	"github.com/sirupsen/logrus"
-	"golang.org/x/net/context"
-	"gopkg.in/cenkalti/backoff.v2"
 )
 
 var errClusterClosed = errors.New("rethinkdb: cluster is closed")
@@ -31,6 +31,7 @@ const (
 // This should hopefully soon be replaced by a backoff system.
 type Cluster struct {
 	opts *ConnectOpts
+	log  *slog.Logger
 
 	mu     sync.RWMutex
 	seeds  []Host // Initial host nodes specified by user.
@@ -45,12 +46,18 @@ type Cluster struct {
 
 // NewCluster creates a new cluster by connecting to the given hosts.
 func NewCluster(hosts []Host, opts *ConnectOpts) (*Cluster, error) {
+	log := slog.Default()
+	if opts != nil && opts.Log != nil {
+		log = opts.Log
+	}
+
 	c := &Cluster{
 		hp:          newHostPool(opts),
 		seeds:       hosts,
 		opts:        opts,
 		closed:      clusterWorking,
 		connFactory: NewConnection,
+		log:         log,
 	}
 
 	err := c.run()
@@ -217,7 +224,7 @@ func (c *Cluster) discover() {
 
 			return c.listenForNodeChanges()
 		}, b, func(err error, wait time.Duration) {
-			Log.Debugf("Error discovering hosts %s, waiting: %s", err, wait)
+			c.log.Debug("Error discovering hosts", "waiting", wait, "error", err)
 		})
 	}
 }
@@ -237,7 +244,7 @@ func (c *Cluster) listenForNodeChanges() error {
 		c.opts,
 	)
 	if err != nil {
-		return fmt.Errorf("Error building query: %s", err)
+		return fmt.Errorf("Error building query: %w", err)
 	}
 
 	cursor, err := node.Query(context.Background(), q) // no need for timeout due to Changes()
@@ -253,9 +260,6 @@ func (c *Cluster) listenForNodeChanges() error {
 		OldVal *nodeStatus `rethinkdb:"old_val"`
 	}
 	for cursor.Next(&result) {
-		addr := fmt.Sprintf("%s:%d", result.NewVal.Network.Hostname, result.NewVal.Network.ReqlPort)
-		addr = strings.ToLower(addr)
-
 		if result.NewVal != nil && result.OldVal == nil {
 			// added new node
 			if !c.nodeExists(result.NewVal.ID) {
@@ -268,11 +272,7 @@ func (c *Cluster) listenForNodeChanges() error {
 					node, err := c.connectNodeWithStatus(result.NewVal)
 					if err == nil {
 						c.addNode(node)
-
-						Log.WithFields(logrus.Fields{
-							"id":   node.ID,
-							"host": node.Host.String(),
-						}).Debug("Connected to node")
+						c.log.Debug("Connected to node", "id", node.ID, "host", node.Host.String())
 					}
 					return err
 				}, b)
@@ -306,14 +306,14 @@ func (c *Cluster) connectCluster() error {
 		conn, err := c.connFactory(host.String(), c.opts)
 		if err != nil {
 			attemptErr = err
-			Log.Warnf("Error creating connection: %s", err.Error())
+			c.log.Warn("Error creating connection", "error", err)
 			continue
 		}
 
 		svrRsp, err := conn.Server()
 		if err != nil {
 			attemptErr = err
-			Log.Warnf("Error fetching server ID: %s", err)
+			c.log.Warn("Error fetching server ID", "error", err)
 			_ = conn.Close()
 
 			continue
@@ -323,19 +323,15 @@ func (c *Cluster) connectCluster() error {
 		node, err := c.connectNode(svrRsp.ID, []Host{host})
 		if err != nil {
 			attemptErr = err
-			Log.Warnf("Error connecting to node: %s", err)
+			c.log.Warn("Error connecting to node", "error", err)
 			continue
 		}
 
 		if _, ok := nodeSet[node.ID]; !ok {
-			Log.WithFields(logrus.Fields{
-				"id":   node.ID,
-				"host": node.Host.String(),
-			}).Debug("Connected to node")
-
+			c.log.Debug("Connected to node", "id", node.ID, "host", node.Host.String())
 			nodeSet[node.ID] = node
 		} else {
-			// dublicate node
+			// duplicate node
 			_ = node.Close()
 		}
 	}

cluster_test.go

@@ -5,13 +5,15 @@ import (
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
+	"io"
+	"log/slog"
+	"net"
+	"time"
+
 	"github.com/stretchr/testify/mock"
 	test "gopkg.in/check.v1"
 	"gopkg.in/rethinkdb/rethinkdb-go.v6/encoding"
 	p "gopkg.in/rethinkdb/rethinkdb-go.v6/ql2"
-	"io"
-	"net"
-	"time"
 )
 
 type ClusterSuite struct{}
@@ -40,6 +42,7 @@ func (s *ClusterSuite) TestCluster_NewSingle_NoDiscover_Ok(c *test.C) {
 		opts:        opts,
 		closed:      clusterWorking,
 		connFactory: mockedConnectionFactory(dialMock),
+		log:         slog.Default(),
 	}
 
 	err := cluster.run()
@@ -84,6 +87,7 @@ func (s *ClusterSuite) TestCluster_NewMultiple_NoDiscover_Ok(c *test.C) {
 		opts:        opts,
 		closed:      clusterWorking,
 		connFactory: mockedConnectionFactory(dialMock),
+		log:         slog.Default(),
 	}
 
 	err := cluster.run()
@@ -115,6 +119,7 @@ func (s *ClusterSuite) TestCluster_NewSingle_NoDiscover_DialFail(c *test.C) {
 		opts:        opts,
 		closed:      clusterWorking,
 		connFactory: mockedConnectionFactory(dialMock),
+		log:         slog.Default(),
 	}
 
 	err := cluster.run()
@@ -146,6 +151,7 @@ func (s *ClusterSuite) TestCluster_NewMultiple_NoDiscover_DialHalfFail(c *test.C
 		opts:        opts,
 		closed:      clusterWorking,
 		connFactory: mockedConnectionFactory(dialMock),
+		log:         slog.Default(),
 	}
 
 	err := cluster.run()
@@ -175,6 +181,7 @@ func (s *ClusterSuite) TestCluster_NewMultiple_NoDiscover_DialFail(c *test.C) {
 		opts:        opts,
 		closed:      clusterWorking,
 		connFactory: mockedConnectionFactory(dialMock),
+		log:         slog.Default(),
 	}
 
 	err := cluster.run()
@@ -200,6 +207,7 @@ func (s *ClusterSuite) TestCluster_NewSingle_NoDiscover_ServerFail(c *test.C) {
 		opts:        opts,
 		closed:      clusterWorking,
 		connFactory: mockedConnectionFactory(dialMock),
+		log:         slog.Default(),
 	}
 
 	err := cluster.run()
@@ -233,6 +241,7 @@ func (s *ClusterSuite) TestCluster_NewSingle_NoDiscover_PingFail(c *test.C) {
 		opts:        opts,
 		closed:      clusterWorking,
 		connFactory: mockedConnectionFactory(dialMock),
+		log:         slog.Default(),
 	}
 
 	err := cluster.run()
@@ -275,6 +284,7 @@ func (s *ClusterSuite) TestCluster_NewSingle_Discover_Ok(c *test.C) {
 		closed:           clusterWorking,
 		connFactory:      mockedConnectionFactory(dialMock),
 		discoverInterval: 10 * time.Second,
+		log:              slog.Default(),
 	}
 
 	err := cluster.run()
@@ -333,6 +343,7 @@ func (s *ClusterSuite) TestCluster_NewMultiple_Discover_Ok(c *test.C) {
 		closed:           clusterWorking,
 		connFactory:      mockedConnectionFactory(dialMock),
 		discoverInterval: 10 * time.Second,
+		log:              slog.Default(),
 	}
 
 	err := cluster.run()
@@ -487,7 +498,7 @@ func expectServerStatus(conn *connMock, token int64, nodeIDs []string, hosts []H
 	})
 
 	rawQ2 := makeContinueQueryRaw(token)
-	// maybe - connection may be closed until cursor fetchs next batch
+	// maybe - connection may be closed until cursor fetches next batch
 	conn.On("Write", rawQ2).Return(0, nil, nil).Maybe().Run(func(args mock.Arguments) {
 		<-readRChan
 	})

connection.go

@@ -10,12 +10,13 @@ import (
 	"time"
 
 	"bytes"
+	"context"
+	"sync"
+
 	"github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/ext"
 	"github.com/opentracing/opentracing-go/log"
-	"golang.org/x/net/context"
 	p "gopkg.in/rethinkdb/rethinkdb-go.v6/ql2"
-	"sync"
 )
 
 const (
@@ -163,7 +164,7 @@ func (c *Connection) Query(ctx context.Context, q Query) (*Response, *Cursor, er
 		c.setBad()
 		return nil, nil, ErrConnectionClosed
 	}
-	if ctx == nil {
+	if ctx == nil || ctx == context.TODO() {
 		ctx = c.contextFromConnectionOpts()
 	}
 
@@ -280,7 +281,7 @@ func (c *Connection) processResponses() {
 		var ok bool
 
 		select {
-		case respPair, openned := <-c.responseChan:
+		case respPair, opened := <-c.responseChan:
 			if respPair.err != nil {
 				// Transport socket error, can't continue to work
 				// Don't know return to who (no token) - return to all
@@ -289,7 +290,7 @@ func (c *Connection) processResponses() {
 				_ = c.Close() // next `if` will be called indirect cascade by closing chans
 				continue
 			}
-			if !openned { // responseChan is connClosed (stopReadChan is closed too)
+			if !opened { // responseChan is connClosed (stopReadChan is closed too)
 				close(c.stopProcessingChan)
 				broadcastError(readRequests, ErrConnectionClosed)
 				c.cursors = nil

connection_handshake.go

@@ -80,7 +80,6 @@ func (c *connectionHandshakeV0_4) writeHandshakeReq() error {
 
 	// Send the protocol type as a 4-byte little-endian-encoded integer
 	binary.LittleEndian.PutUint32(data[pos:], uint32(p.VersionDummy_JSON))
-	pos += 4
 
 	return c.conn.writeData(data)
 }
@@ -90,7 +89,7 @@ func (c *connectionHandshakeV0_4) readHandshakeSuccess() error {
 	line, err := reader.ReadBytes('\x00')
 	if err != nil {
 		if err == io.EOF {
-			return fmt.Errorf("Unexpected EOF: %s", string(line))
+			return fmt.Errorf("unexpected EOF: %s", string(line))
 		}
 		return err
 	}

connection_helper.go

@@ -1,7 +1,7 @@
 package rethinkdb
 
 import (
-	"golang.org/x/net/context"
+	"context"
 	"io"
 )
 

connection_test.go

@@ -1,17 +1,18 @@
 package rethinkdb
 
 import (
+	"context"
 	"encoding/binary"
 	"encoding/json"
+	"io"
+	"sync"
+	"time"
+
 	"github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/mocktracer"
 	"github.com/stretchr/testify/mock"
-	"golang.org/x/net/context"
 	test "gopkg.in/check.v1"
 	p "gopkg.in/rethinkdb/rethinkdb-go.v6/ql2"
-	"io"
-	"sync"
-	"time"
 )
 
 func runConnection(c *Connection) <-chan struct{} {
@@ -109,15 +110,15 @@ func (s *ConnectionSuite) TestConnection_Query_DefaultDBOk(c *test.C) {
 }
 
 func (s *ConnectionSuite) TestConnection_Query_Nil(c *test.C) {
-	response, cursor, err := (*Connection)(nil).Query(nil, Query{})
+	response, cursor, err := (*Connection)(nil).Query(context.TODO(), Query{})
 	c.Assert(err, test.Equals, ErrConnectionClosed)
 	c.Assert(response, test.IsNil)
 	c.Assert(cursor, test.IsNil)
 }
 
 func (s *ConnectionSuite) TestConnection_Query_NilConn(c *test.C) {
 	connection := newConnection(nil, "addr", &ConnectOpts{Database: "db"})
-	response, cursor, err := connection.Query(nil, Query{})
+	response, cursor, err := connection.Query(context.TODO(), Query{})
 	c.Assert(err, test.Equals, ErrConnectionClosed)
 	c.Assert(response, test.IsNil)
 	c.Assert(cursor, test.IsNil)
@@ -157,7 +158,7 @@ func (s *ConnectionSuite) TestConnection_Query_NoReplyOk(c *test.C) {
 
 	connection := newConnection(conn, "addr", &ConnectOpts{})
 	done := runConnection(connection)
-	response, cursor, err := connection.Query(nil, q)
+	response, cursor, err := connection.Query(context.TODO(), q)
 	connection.Close()
 	<-done