fix tests socket leakage

Author: CMogilko

.gitignore

@@ -27,3 +27,4 @@ _testmain.go
 .wercker
 
 cover.html
+.idea/

.travis.yml

@@ -4,6 +4,8 @@ go:
   - 1.7.x
   - 1.8.x
   - 1.9.x
+  - 1.10.x
+  - 1.11.x
   - tip
 
 cache: apt
@@ -18,8 +20,8 @@ before_script:
   - sudo apt-get update
   - sudo apt-get install rethinkdb
   - rethinkdb > /dev/null 2>&1 &
-  - rethinkdb --port-offset 1 --directory rethinkdb_data1 --join localhost:29016 > /dev/null 2>&1 &
-  - rethinkdb --port-offset 2 --directory rethinkdb_data2 --join localhost:29016 > /dev/null 2>&1 &
-  - rethinkdb --port-offset 3 --directory rethinkdb_data3 --join localhost:29016 > /dev/null 2>&1 &
+  - rethinkdb --port-offset 1 --directory rethinkdb_data1 --join localhost:29015 > /dev/null 2>&1 &
+  - rethinkdb --port-offset 2 --directory rethinkdb_data2 --join localhost:29015 > /dev/null 2>&1 &
+  - rethinkdb --port-offset 3 --directory rethinkdb_data3 --join localhost:29015 > /dev/null 2>&1 &
 
 script: go test -tags='cluster' -short -race -v ./...

internal/integration/reql_tests/common.go

@@ -57,13 +57,13 @@ func runAndAssert(suite suite.Suite, expected, v interface{}, session *r.Session
 
 func fetchAndAssert(suite suite.Suite, expected, result interface{}, count int) {
 	switch v := expected.(type) {
-	case Expected:
+	case compare.Expected:
 		v.Fetch = true
 		v.FetchCount = count
 
 		expected = v
 	default:
-		expected = Expected(compare.Expected{
+		expected = compare.Expected(compare.Expected{
 			Val:        v,
 			Fetch:      true,
 			FetchCount: count,
@@ -105,17 +105,15 @@ func assertExpected(suite suite.Suite, expected interface{}, obtainedCursor *r.C
 	switch expected := expected.(type) {
 	case Err:
 		expected.assert(suite, obtainedCursor, obtainedErr)
-	case Expected:
-		expected.assert(suite, obtainedCursor, obtainedErr)
+	case compare.Expected:
+		assert(suite, expected, obtainedCursor, obtainedErr)
 	default:
-		Expected(compare.Expected{Val: expected}).assert(suite, obtainedCursor, obtainedErr)
+		assert(suite, compare.Expected{Val: expected}, obtainedCursor, obtainedErr)
 	}
 }
 
-type Expected compare.Expected
-
-func (expected Expected) assert(suite suite.Suite, obtainedCursor *r.Cursor, obtainedErr error) {
-	if suite.NoError(obtainedErr, "Query returned unexpected error") {
+func assert(suite suite.Suite, expected compare.Expected, obtainedCursor *r.Cursor, obtainedErr error) {
+	if !suite.NoError(obtainedErr, "Query returned unexpected error") {
 		return
 	}
 

internal/utils/conn_counter.go

@@ -0,0 +1,91 @@
+package utils
+
+import (
+	"fmt"
+	"net"
+	"os"
+	"regexp"
+	"runtime"
+	"runtime/debug"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+var connsCount int64
+var printer sync.Once
+
+var mu sync.Mutex
+var createdStacks = make(map[string]int)
+var reg = regexp.MustCompile(`0[xX][0-9a-fA-F]+`)
+
+type connCounting struct {
+	net.Conn
+
+	closed bool
+}
+
+// Socket leak debug net.Conn wrapper
+func NewCountingConn(conn net.Conn) net.Conn{
+	c := &connCounting{
+		Conn: conn,
+		closed: false,
+	}
+	runtime.SetFinalizer(c, func(cc *connCounting){
+		if !cc.closed {
+			atomic.AddInt64(&connsCount, -1)
+			cc.closed = true
+		}
+	})
+
+	atomic.AddInt64(&connsCount, 1)
+	printer.Do(func(){
+		go func() {
+			t := time.NewTicker(time.Second)
+			f, err := os.Create("sockets.ticker")
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "Failed to create sockets.ticker file: %v\n", err)
+				return
+			}
+			for {
+				<-t.C
+				fmt.Fprintf(f, "Connections count: %v\n", atomic.LoadInt64(&connsCount))
+				f.Sync()
+			}
+		}()
+	})
+
+	st := string(debug.Stack())
+	st = st[strings.Index(st, "\n")+1:]
+	st = reg.ReplaceAllString(st, "")
+
+	mu.Lock()
+	_, has := createdStacks[st]
+	if !has {
+		createdStacks[st] = 1
+	} else {
+		createdStacks[st]++
+	}
+	printStacks()
+	mu.Unlock()
+
+	return c
+}
+
+func (c *connCounting) Close() error {
+	if !c.closed {
+		atomic.AddInt64(&connsCount, -1)
+		c.closed = true
+	}
+	return c.Conn.Close()
+}
+
+func printStacks() {
+	f, _ := os.Create("sockets.created")
+	for s, c := range createdStacks {
+		fmt.Fprintf(f, "%v:\n%v\n\n", c, s)
+	}
+	f.Sync()
+	f.Close()
+}