all: modern build environment and dependencies

.github/workflows/ci.yml

@@ -0,0 +1,23 @@
+on: [push, pull_request]
+name: Test
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+    - name: Install Go
+      # github.com/actions/setup-go/tags
+      uses: actions/setup-go@v5
+      with:
+        go-version: 1.22.x
+      # github.com/actions/checkout/tags
+    - uses: actions/checkout@v4
+      with:
+         path: './src/github.com/segmentio/nsq-go'
+    - run: echo "PATH=$GITHUB_WORKSPACE/bin:$PATH" >> $GITHUB_ENV
+    - name: Run tests
+      run: |
+        docker-compose up -d
+        go vet ./...
+        go run honnef.co/go/tools/cmd/staticcheck@latest ./...
+        go test -race -v ./...
+      working-directory: './src/github.com/segmentio/nsq-go'

.gitignore

@@ -30,4 +30,4 @@ _testmain.go
 /vendor/*/
 nsq-to-nsq
 nsq-to-http
-nsqlookup-proxy
+/nsqlookup-proxy

cmd/nsq-to-http/main.go

@@ -2,7 +2,7 @@ package main
 
 import (
 	"bytes"
-	"io/ioutil"
+	"io"
 	"log"
 	"math"
 	"math/rand"
@@ -76,6 +76,9 @@ func main() {
 		MaxInFlight: config.MaxInFlight,
 		Identify:    nsq.Identify{UserAgent: config.UserAgent},
 	})
+	if err != nil {
+		log.Fatalf("could not start consumer: %v", err)
+	}
 
 	wg := sync.WaitGroup{}
 	wg.Add(config.MaxInFlight)
@@ -87,7 +90,7 @@ func main() {
 		}()
 	}
 
-	sigchan := make(chan os.Signal)
+	sigchan := make(chan os.Signal, 1)
 	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
 
 	<-sigchan
@@ -118,7 +121,7 @@ func forward(dst *url.URL, contentType, userAgent string, timeout time.Duration,
 				"Content-Type": {contentType},
 				"User-Agent":   {userAgent},
 			},
-			Body:          ioutil.NopCloser(bytes.NewReader(msg.Body)),
+			Body:          io.NopCloser(bytes.NewReader(msg.Body)),
 			ContentLength: int64(len(msg.Body)),
 		}).WithContext(timers.LowRes.Timeout(timeout)))
 

cmd/nsq-to-nsq/main.go

@@ -67,7 +67,7 @@ func main() {
 		go forward(producer, messages, done)
 	}
 
-	sigchan := make(chan os.Signal)
+	sigchan := make(chan os.Signal, 1)
 	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)
 	<-sigchan
 

cmd/nsqlookup-proxy/main.go

@@ -8,10 +8,10 @@ import (
 	"time"
 
 	"github.com/segmentio/conf"
-	"github.com/segmentio/events"
-	_ "github.com/segmentio/events/ecslogs"
-	"github.com/segmentio/events/httpevents"
-	_ "github.com/segmentio/events/text"
+	"github.com/segmentio/events/v2"
+	_ "github.com/segmentio/events/v2/ecslogs"
+	"github.com/segmentio/events/v2/httpevents"
+	_ "github.com/segmentio/events/v2/text"
 	nsq "github.com/segmentio/nsq-go"
 	"github.com/segmentio/nsq-go/nsqlookup"
 )

consumer.go

@@ -218,11 +218,11 @@ func (c *Consumer) run() {
 				go func(cm connMeta) {
 					start := time.Now()
 					for len(cm.CmdChan) > 0 {
-						if time.Now().Sub(start) > c.drainTimeout {
-							log.Print("failed to drain CmdChan for connection, closing now")
+						if time.Since(start) > c.drainTimeout {
+							log.Println("failed to drain CmdChan for connection, closing now")
 							break
 						}
-						log.Println("awaiting for write channel to flush any requeue commands")
+						log.Println("waiting for write channel to flush any requeue commands")
 						time.Sleep(time.Millisecond * 500)
 					}
 					closeCommand(cm.CmdChan)
@@ -354,7 +354,6 @@ func (c *Consumer) close() {
 		sendCommand(cm.CmdChan, Cls{})
 	}
 	c.mtx.Unlock()
-	return
 }
 
 func (c *Consumer) closeConn(addr string) {

docker-compose.yml

@@ -4,7 +4,7 @@ version: '2'
 # tests.
 services:
   consul:
-    image: consul:latest
+    image: hashicorp/consul:latest
     command: agent -server -dev -log-level debug -client 0.0.0.0
     ports:
       - 8500:8500

error.go

@@ -94,10 +94,3 @@ func appendError(errList error, err error) error {
 		return errorList{err}
 	}
 }
-
-func isTimeout(err error) bool {
-	e, ok := err.(interface {
-		Timeout() bool
-	})
-	return ok && e.Timeout()
-}

go.mod

@@ -1,15 +1,21 @@
 module github.com/segmentio/nsq-go
 
-go 1.11
+go 1.18
 
 require (
-	github.com/pkg/errors v0.8.0
-	github.com/segmentio/conf v1.0.0
-	github.com/segmentio/events v2.1.0+incompatible
-	github.com/segmentio/go-snakecase v1.0.0 // indirect
+	github.com/pkg/errors v0.9.1
+	github.com/segmentio/conf v1.3.0
+	github.com/segmentio/events/v2 v2.6.0
+	github.com/segmentio/timers v1.1.1
+)
+
+require (
+	github.com/segmentio/asm v1.1.3 // indirect
+	github.com/segmentio/encoding v0.3.6 // indirect
+	github.com/segmentio/go-snakecase v1.2.0 // indirect
 	github.com/segmentio/objconv v1.0.1 // indirect
-	github.com/segmentio/timers v0.0.0-20180605162245-8ad1428b010e
-	gopkg.in/go-playground/assert.v1 v1.2.1 // indirect
+	golang.org/x/sys v0.18.0 // indirect
 	gopkg.in/go-playground/mold.v2 v2.2.0 // indirect
-	gopkg.in/validator.v2 v2.0.0-20180514200540-135c24b11c19 // indirect
+	gopkg.in/validator.v2 v2.0.1 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 )

go.sum

@@ -1,22 +1,31 @@
-github.com/pkg/errors v0.8.0 h1:WdK/asTD0HN+q6hsWO3/vpuAkAr+tw6aNJNDFFf0+qw=
-github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/segmentio/conf v1.0.0 h1:oRF4BtoJbI/+I7fUngYMnMcKFbjqVUFi8hv4Pp0l88w=
-github.com/segmentio/conf v1.0.0/go.mod h1:y0VyxYAlU2slxCjm7XX7tGKFlN39bwHCZrbOpCcLsr8=
-github.com/segmentio/events v2.1.0+incompatible h1:7ns47dgRJMt/JgIXrNU0MiD/NtCGa55lKzWP15dcGnM=
-github.com/segmentio/events v2.1.0+incompatible/go.mod h1:npQUbmKYO33tlRpaQNZjgD2mXv0fb2hbOH0CNVs6g2Y=
-github.com/segmentio/go-snakecase v1.0.0 h1:FSeHpP0sBL3O+MCpxvQZrS5a51WAki6gposZuwVE9L4=
-github.com/segmentio/go-snakecase v1.0.0/go.mod h1:jk1miR5MS7Na32PZUykG89Arm+1BUSYhuGR6b7+hJto=
+github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc=
+github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg=
+github.com/segmentio/conf v1.3.0 h1:rdTRjia6Kzw0n69CGybG0n/Qq/88bCvpUO3uJnAmDo4=
+github.com/segmentio/conf v1.3.0/go.mod h1:DfjTISxmTT8oAhBAde5/PQaBQFj1bvg8EDbuGMeq1Q4=
+github.com/segmentio/encoding v0.3.6 h1:E6lVLyDPseWEulBmCmAKPanDd3jiyGDo5gMcugCRwZQ=
+github.com/segmentio/encoding v0.3.6/go.mod h1:n0JeuIqEQrQoPDGsjo8UNd1iA0U8d8+oHAA4E3G3OxM=
+github.com/segmentio/events/v2 v2.6.0 h1:MRPhHWwfLGFjkppjkdCDw2DZVBMyMb/JcBVnDVVNdzc=
+github.com/segmentio/events/v2 v2.6.0/go.mod h1:oDngvacuvZ/bPnAq8VKNzvk8ICS1pVVEtQESVcvSL/w=
+github.com/segmentio/go-snakecase v1.2.0 h1:4cTmEjPGi03WmyAHWBjX53viTpBkn/z+4DO++fqYvpw=
+github.com/segmentio/go-snakecase v1.2.0/go.mod h1:jk1miR5MS7Na32PZUykG89Arm+1BUSYhuGR6b7+hJto=
 github.com/segmentio/objconv v1.0.1 h1:QjfLzwriJj40JibCV3MGSEiAoXixbp4ybhwfTB8RXOM=
 github.com/segmentio/objconv v1.0.1/go.mod h1:auayaH5k3137Cl4SoXTgrzQcuQDmvuVtZgS0fb1Ahys=
-github.com/segmentio/timers v0.0.0-20180605162245-8ad1428b010e h1:GNlZqttb0RQTUYTYnSwAMyqgHzYouBGVImRJp8Y43Hg=
-github.com/segmentio/timers v0.0.0-20180605162245-8ad1428b010e/go.mod h1:/6SKE4F6LzTh32hPOJs0KRJ7MHHWpTZ7wb4PxdjgzaQ=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+github.com/segmentio/timers v1.1.1 h1:3RUmQKeNXiF+NS0P8OqIBLS8dLYjw/7IBQzJy2Aqrl8=
+github.com/segmentio/timers v1.1.1/go.mod h1:/6SKE4F6LzTh32hPOJs0KRJ7MHHWpTZ7wb4PxdjgzaQ=
+golang.org/x/sys v0.0.0-20211110154304-99a53858aa08/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4=
+golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/go-playground/assert.v1 v1.2.1 h1:xoYuJVE7KT85PYWrN730RguIQO0ePzVRfFMXadIrXTM=
-gopkg.in/go-playground/assert.v1 v1.2.1/go.mod h1:9RXL0bg/zibRAgZUYszZSwO/z8Y/a8bDuhia5mkpMnE=
 gopkg.in/go-playground/mold.v2 v2.2.0 h1:Y4IYB4/HYQfuq43zaKh6vs9cVelLE9qbqe2fkyfCTWQ=
 gopkg.in/go-playground/mold.v2 v2.2.0/go.mod h1:XMyyRsGtakkDPbxXbrA5VODo6bUXyvoDjLd5l3T0XoA=
-gopkg.in/validator.v2 v2.0.0-20180514200540-135c24b11c19 h1:WB265cn5OpO+hK3pikC9hpP1zI/KTwmyMFKloW9eOVc=
-gopkg.in/validator.v2 v2.0.0-20180514200540-135c24b11c19/go.mod h1:o4V0GXN9/CAmCsvJ0oXYZvrZOe7syiDZSN1GWGZTGzc=
-gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE=
+gopkg.in/validator.v2 v2.0.1 h1:xF0KWyGWXm/LM2G1TrEjqOu4pa6coO9AlWSf3msVfDY=
+gopkg.in/validator.v2 v2.0.1/go.mod h1:lIUZBlB3Im4s/eYp39Ry/wkR02yOPhZ9IwIRBjuPuG8=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=

lookup.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"encoding/json"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"net/url"
 	"time"
@@ -156,7 +155,7 @@ func (c *LookupClient) do(host string, method string, path string, query url.Val
 	}
 
 	if len(data) != 0 {
-		body = ioutil.NopCloser(bytes.NewReader(data))
+		body = io.NopCloser(bytes.NewReader(data))
 	}
 
 	if res, err = c.Do(&http.Request{
@@ -184,7 +183,7 @@ func (c *LookupClient) do(host string, method string, path string, query url.Val
 
 	defer res.Body.Close()
 
-	if ret, err = ioutil.ReadAll(res.Body); err != nil {
+	if ret, err = io.ReadAll(res.Body); err != nil {
 		err = errors.Wrap(err, "reading response body")
 		return
 	}

nsq.go

@@ -3,7 +3,6 @@ package nsq
 import (
 	"bytes"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"net/url"
 
@@ -125,7 +124,7 @@ func (c *Client) do(method string, path string, query url.Values, data []byte) (
 	}
 
 	if len(data) != 0 {
-		body = ioutil.NopCloser(bytes.NewReader(data))
+		body = io.NopCloser(bytes.NewReader(data))
 	}
 
 	if res, err = c.Do(&http.Request{
@@ -157,7 +156,7 @@ func (c *Client) do(method string, path string, query url.Values, data []byte) (
 		return
 	}
 
-	if ret, err = ioutil.ReadAll(res.Body); err != nil {
+	if ret, err = io.ReadAll(res.Body); err != nil {
 		err = errors.Wrapf(err, "%s %s://%s?%s", method, scheme, host, query.Encode())
 		return
 	}

nsqlookup/consul.go

@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"path"
 	"sort"
@@ -428,7 +427,7 @@ func (e *ConsulEngine) do(ctx context.Context, method string, url string, send i
 	defer res.Body.Close()
 
 	if res.StatusCode != http.StatusOK {
-		io.Copy(ioutil.Discard, res.Body)
+		io.Copy(io.Discard, res.Body)
 		err = consulError{
 			method: method,
 			url:    url,
@@ -443,7 +442,7 @@ func (e *ConsulEngine) do(ctx context.Context, method string, url string, send i
 			return
 		}
 	} else {
-		io.Copy(ioutil.Discard, res.Body)
+		io.Copy(io.Discard, res.Body)
 	}
 
 	return

nsqlookup/engine.go

@@ -232,10 +232,6 @@ func httpBroadcastAddress(info NodeInfo) string {
 	return makeBroadcastAddress(info.BroadcastAddress, info.HttpPort)
 }
 
-func tcpBroadcastAddress(info NodeInfo) string {
-	return makeBroadcastAddress(info.BroadcastAddress, info.TcpPort)
-}
-
 func makeBroadcastAddress(addr string, port int) string {
 	host, _, _ := net.SplitHostPort(addr)
 	if len(host) == 0 {