.github/workflows/test.yml

@@ -6,26 +6,26 @@ on:
 
 jobs:
   test:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     timeout-minutes: 30
     strategy:
       fail-fast: false
       matrix:
         imgtag:
-          - "golang:1.15-buster"
-          - "golang:1.16-buster"
-          - "golang:1.17-buster"
+          - "golang:1.23-bullseye"
+          - "golang:1.24-bullseye"
         goarch:
           - "amd64"
         nsq_ver:
           - "nsq-1.1.0.linux-amd64.go1.10.3"
           - "nsq-1.2.0.linux-amd64.go1.12.9"
           - "nsq-1.2.1.linux-amd64.go1.16.6"
+          - "nsq-1.3.0.linux-amd64.go1.21.5"
         include:
           # test 386 only against latest version of NSQ
-          - imgtag: "golang:1.17-buster"
+          - imgtag: "golang:1.24-bullseye"
             goarch: "386"
-            nsq_ver: "nsq-1.2.1.linux-amd64.go1.16.6"
+            nsq_ver: "nsq-1.3.0.linux-amd64.go1.21.5"
 
     container: "${{matrix.imgtag}}"
     env:

api_request.go

@@ -3,7 +3,7 @@ package nsq
 import (
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net"
 	"net/http"
 	"time"
@@ -47,7 +47,7 @@ func apiRequestNegotiateV1(httpclient *http.Client, method string, endpoint stri
 		return err
 	}
 
-	respBody, err := ioutil.ReadAll(resp.Body)
+	respBody, err := io.ReadAll(resp.Body)
 	resp.Body.Close()
 	if err != nil {
 		return err

config.go

@@ -5,7 +5,6 @@ import (
 	"crypto/x509"
 	"errors"
 	"fmt"
-	"io/ioutil"
 	"log"
 	"math"
 	"math/rand"
@@ -140,6 +139,10 @@ type Config struct {
 	Hostname  string `opt:"hostname"`
 	UserAgent string `opt:"user_agent"`
 
+	// Topology hints allow nsqd to prefer same zone and same region consumers
+	TopologyRegion string `opt:"topology_region"`
+	TopologyZone   string `opt:"topology_zone"`
+
 	// Duration of time between heartbeats. This must be less than ReadTimeout
 	HeartbeatInterval time.Duration `opt:"heartbeat_interval" default:"30s"`
 	// Integer percentage to sample the channel (requires nsqd 0.2.25+)
@@ -177,6 +180,9 @@ type Config struct {
 	// The server-side message timeout for messages delivered to this client
 	MsgTimeout time.Duration `opt:"msg_timeout" min:"0"`
 
+	// Maximum size of a single message in bytes (0 means no limit)
+	MaxMsgSize int32 `opt:"max_msg_size" min:"0" default:"0"`
+
 	// Secret for nsqd authentication (requires nsqd 0.2.29+)
 	AuthSecret string `opt:"auth_secret"`
 	// Use AuthSecret as 'Authorization: Bearer {AuthSecret}' on lookupd queries
@@ -205,15 +211,15 @@ func NewConfig() *Config {
 //
 // Calls to Set() that take a time.Duration as an argument can be input as:
 //
-// 	"1000ms" (a string parsed by time.ParseDuration())
-// 	1000 (an integer interpreted as milliseconds)
-// 	1000*time.Millisecond (a literal time.Duration value)
+//	"1000ms" (a string parsed by time.ParseDuration())
+//	1000 (an integer interpreted as milliseconds)
+//	1000*time.Millisecond (a literal time.Duration value)
 //
 // Calls to Set() that take bool can be input as:
 //
-// 	"true" (a string parsed by strconv.ParseBool())
-// 	true (a boolean)
-// 	1 (an int where 1 == true and 0 == false)
+//	"true" (a string parsed by strconv.ParseBool())
+//	true (a boolean)
+//	1 (an int where 1 == true and 0 == false)
 //
 // It returns an error for an invalid option or value.
 func (c *Config) Set(option string, value interface{}) error {
@@ -431,7 +437,7 @@ func (t *tlsConfig) Set(c *Config, option string, value interface{}) error {
 			return fmt.Errorf("ERROR: %v is not a string", value)
 		}
 		tlsCertPool := x509.NewCertPool()
-		caCertFile, err := ioutil.ReadFile(filename)
+		caCertFile, err := os.ReadFile(filename)
 		if err != nil {
 			return fmt.Errorf("ERROR: failed to read custom Certificate Authority file %s", err)
 		}

conn.go

@@ -119,8 +119,7 @@ func NewConn(addr string, config *Config, delegate ConnDelegate) *Conn {
 // The logger parameter is an interface that requires the following
 // method to be implemented (such as the the stdlib log.Logger):
 //
-//    Output(calldepth int, s string)
-//
+//	Output(calldepth int, s string)
 func (c *Conn) SetLogger(l logger, lvl LogLevel, format string) {
 	c.logGuard.Lock()
 	defer c.logGuard.Unlock()
@@ -345,6 +344,8 @@ func (c *Conn) identify() (*IdentifyResponse, error) {
 		ci["output_buffer_timeout"] = int64(c.config.OutputBufferTimeout / time.Millisecond)
 	}
 	ci["msg_timeout"] = int64(c.config.MsgTimeout / time.Millisecond)
+	ci["topology_region"] = c.config.TopologyRegion
+	ci["topology_zone"] = c.config.TopologyZone
 	cmd, err := Identify(ci)
 	if err != nil {
 		return nil, ErrIdentify{err.Error()}
@@ -355,7 +356,7 @@ func (c *Conn) identify() (*IdentifyResponse, error) {
 		return nil, ErrIdentify{err.Error()}
 	}
 
-	frameType, data, err := ReadUnpackedResponse(c)
+	frameType, data, err := ReadUnpackedResponse(c, c.config.MaxMsgSize)
 	if err != nil {
 		return nil, ErrIdentify{err.Error()}
 	}
@@ -434,7 +435,7 @@ func (c *Conn) upgradeTLS(tlsConf *tls.Config) error {
 	}
 	c.r = c.tlsConn
 	c.w = c.tlsConn
-	frameType, data, err := ReadUnpackedResponse(c)
+	frameType, data, err := ReadUnpackedResponse(c, c.config.MaxMsgSize)
 	if err != nil {
 		return err
 	}
@@ -452,7 +453,7 @@ func (c *Conn) upgradeDeflate(level int) error {
 	fw, _ := flate.NewWriter(conn, level)
 	c.r = flate.NewReader(conn)
 	c.w = fw
-	frameType, data, err := ReadUnpackedResponse(c)
+	frameType, data, err := ReadUnpackedResponse(c, c.config.MaxMsgSize)
 	if err != nil {
 		return err
 	}
@@ -468,8 +469,8 @@ func (c *Conn) upgradeSnappy() error {
 		conn = c.tlsConn
 	}
 	c.r = snappy.NewReader(conn)
-	c.w = snappy.NewWriter(conn)
-	frameType, data, err := ReadUnpackedResponse(c)
+	c.w = snappy.NewBufferedWriter(conn)
+	frameType, data, err := ReadUnpackedResponse(c, c.config.MaxMsgSize)
 	if err != nil {
 		return err
 	}
@@ -490,7 +491,7 @@ func (c *Conn) auth(secret string) error {
 		return err
 	}
 
-	frameType, data, err := ReadUnpackedResponse(c)
+	frameType, data, err := ReadUnpackedResponse(c, c.config.MaxMsgSize)
 	if err != nil {
 		return err
 	}
@@ -518,7 +519,7 @@ func (c *Conn) readLoop() {
 			goto exit
 		}
 
-		frameType, data, err := ReadUnpackedResponse(c)
+		frameType, data, err := ReadUnpackedResponse(c, c.config.MaxMsgSize)
 		if err != nil {
 			if err == io.EOF && atomic.LoadInt32(&c.closeFlag) == 1 {
 				goto exit
@@ -692,7 +693,7 @@ func (c *Conn) cleanup() {
 			msgsInFlight = atomic.LoadInt64(&c.messagesInFlight)
 		}
 		if msgsInFlight > 0 {
-			if time.Now().Sub(lastWarning) > time.Second {
+			if time.Since(lastWarning) > time.Second {
 				c.log(LogLevelWarning, "draining... waiting for %d messages in flight", msgsInFlight)
 				lastWarning = time.Now()
 			}
@@ -701,7 +702,7 @@ func (c *Conn) cleanup() {
 		// until the readLoop has exited we cannot be sure that there
 		// still won't be a race
 		if atomic.LoadInt32(&c.readLoopRunning) == 1 {
-			if time.Now().Sub(lastWarning) > time.Second {
+			if time.Since(lastWarning) > time.Second {
 				c.log(LogLevelWarning, "draining... readLoop still running")
 				lastWarning = time.Now()
 			}

consumer.go

@@ -33,9 +33,9 @@ type Handler interface {
 // HandlerFunc is a convenience type to avoid having to declare a struct
 // to implement the Handler interface, it can be used like this:
 //
-// 	consumer.AddHandler(nsq.HandlerFunc(func(m *Message) error {
-// 		// handle the message
-// 	}))
+//	consumer.AddHandler(nsq.HandlerFunc(func(m *Message) error {
+//		// handle the message
+//	}))
 type HandlerFunc func(message *Message) error
 
 // HandleMessage implements the Handler interface
@@ -220,8 +220,7 @@ func (r *Consumer) conns() []*Conn {
 // The logger parameter is an interface that requires the following
 // method to be implemented (such as the the stdlib log.Logger):
 //
-//    Output(calldepth int, s string) error
-//
+//	Output(calldepth int, s string) error
 func (r *Consumer) SetLogger(l logger, lvl LogLevel) {
 	r.logGuard.Lock()
 	defer r.logGuard.Unlock()
@@ -266,8 +265,7 @@ func (r *Consumer) getLogLevel() LogLevel {
 // of the following interfaces that modify the behavior
 // of the `Consumer`:
 //
-//    DiscoveryFilter
-//
+//	DiscoveryFilter
 func (r *Consumer) SetBehaviorDelegate(cb interface{}) {
 	matched := false
 
@@ -312,7 +310,7 @@ func (r *Consumer) getMaxInFlight() int32 {
 // ChangeMaxInFlight sets a new maximum number of messages this comsumer instance
 // will allow in-flight, and updates all existing connections as appropriate.
 //
-// For example, ChangeMaxInFlight(0) would pause message flow
+// # For example, ChangeMaxInFlight(0) would pause message flow
 //
 // If already connected, it updates the reader RDY state for each connection.
 func (r *Consumer) ChangeMaxInFlight(maxInFlight int) {
@@ -609,8 +607,11 @@ func (r *Consumer) ConnectToNSQD(addr string) error {
 
 	// pre-emptive signal to existing connections to lower their RDY count
 	for _, c := range r.conns() {
-		r.maybeUpdateRDY(c)
+		if c != conn {
+			r.maybeUpdateRDY(c)
+		}
 	}
+	r.maybeUpdateRDY(conn)
 
 	return nil
 }
@@ -912,7 +913,9 @@ func (r *Consumer) maybeUpdateRDY(conn *Conn) {
 
 	count := r.perConnMaxInFlight()
 	r.log(LogLevelDebug, "(%s) sending RDY %d", conn, count)
-	r.updateRDY(conn, count)
+	if err := r.updateRDY(conn, count); err != nil {
+		r.log(LogLevelWarning, "(%s) error sending RDY %d: %v", conn, count, err)
+	}
 }
 
 func (r *Consumer) rdyLoop() {
@@ -1028,8 +1031,8 @@ func (r *Consumer) redistributeRDY() {
 
 	possibleConns := make([]*Conn, 0, len(conns))
 	for _, c := range conns {
-		lastMsgDuration := time.Now().Sub(c.LastMessageTime())
-		lastRdyDuration := time.Now().Sub(c.LastRdyTime())
+		lastMsgDuration := time.Since(c.LastMessageTime())
+		lastRdyDuration := time.Since(c.LastRdyTime())
 		rdyCount := c.RDY()
 		r.log(LogLevelDebug, "(%s) rdy: %d (last message received %s)",
 			c.String(), rdyCount, lastMsgDuration)
@@ -1104,7 +1107,7 @@ func (r *Consumer) stopHandlers() {
 // AddHandler sets the Handler for messages received by this Consumer. This can be called
 // multiple times to add additional handlers. Handler will have a 1:1 ratio to message handling goroutines.
 //
-// This panics if called after connecting to NSQD or NSQ Lookupd
+// # This panics if called after connecting to NSQD or NSQ Lookupd
 //
 // (see Handler or HandlerFunc for details on implementing this interface)
 func (r *Consumer) AddHandler(handler Handler) {
@@ -1115,7 +1118,7 @@ func (r *Consumer) AddHandler(handler Handler) {
 // takes a second argument which indicates the number of goroutines to spawn for
 // message handling.
 //
-// This panics if called after connecting to NSQD or NSQ Lookupd
+// # This panics if called after connecting to NSQD or NSQ Lookupd
 //
 // (see Handler or HandlerFunc for details on implementing this interface)
 func (r *Consumer) AddConcurrentHandlers(handler Handler, concurrency int) {
@@ -1223,7 +1226,7 @@ func buildLookupAddr(addr, topic string) (string, error) {
 		u.Path = "/lookup"
 	}
 
-	v, err := url.ParseQuery(u.RawQuery)
+	v, _ := url.ParseQuery(u.RawQuery)
 	v.Add("topic", topic)
 	u.RawQuery = v.Encode()
 	return u.String(), nil

consumer_test.go

@@ -6,7 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"log"
 	"net"
 	"net/http"
@@ -25,7 +25,7 @@ type MyTestHandler struct {
 	messagesFailed   int
 }
 
-var nullLogger = log.New(ioutil.Discard, "", log.LstdFlags)
+var nullLogger = log.New(io.Discard, "", log.LstdFlags)
 
 func (h *MyTestHandler) LogFailedMessage(message *Message) {
 	h.messagesFailed++
@@ -58,7 +58,7 @@ func (h *MyTestHandler) HandleMessage(message *Message) error {
 func SendMessage(t *testing.T, port int, topic string, method string, body []byte) {
 	httpclient := &http.Client{}
 	endpoint := fmt.Sprintf("http://127.0.0.1:%d/%s?topic=%s", port, method, topic)
-	req, err := http.NewRequest("POST", endpoint, bytes.NewBuffer(body))
+	req, _ := http.NewRequest("POST", endpoint, bytes.NewBuffer(body))
 	resp, err := httpclient.Do(req)
 	if err != nil {
 		t.Fatalf(err.Error())

go.mod

@@ -1,5 +1,5 @@
 module github.com/nsqio/go-nsq
 
-go 1.11
+go 1.17
 
-require github.com/golang/snappy v0.0.1
+require github.com/golang/snappy v0.0.4

go.sum

@@ -1,2 +1,2 @@
-github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
-github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=

mock_test.go

@@ -184,7 +184,7 @@ func framedResponse(frameType int32, data []byte) []byte {
 		return nil
 	}
 
-	_, err = w.Write(data)
+	_, _ = w.Write(data)
 	return w.Bytes()
 }
 

producer.go

@@ -92,7 +92,7 @@ func NewProducer(addr string, config *Config) (*Producer, error) {
 
 	// Set default logger for all log levels
 	l := log.New(os.Stderr, "", log.Flags())
-	for index, _ := range p.logger {
+	for index := range p.logger {
 		p.logger[index] = l
 	}
 	return p, nil
@@ -120,8 +120,7 @@ func (w *Producer) Ping() error {
 // The logger parameter is an interface that requires the following
 // method to be implemented (such as the the stdlib log.Logger):
 //
-//    Output(calldepth int, s string)
-//
+//	Output(calldepth int, s string)
 func (w *Producer) SetLogger(l logger, lvl LogLevel) {
 	w.logGuard.Lock()
 	defer w.logGuard.Unlock()
@@ -367,6 +366,17 @@ exit:
 }
 
 func (w *Producer) popTransaction(frameType int32, data []byte) {
+	if len(w.transactions) == 0 {
+		dataLen := len(data)
+		if dataLen > 32 {
+			data = data[:32]
+		}
+		w.log(LogLevelError,
+			"(%s) unexpected response type=%d len=%d data[:32]=0x%x",
+			w.conn.String(), frameType, dataLen, data)
+		w.close()
+		return
+	}
 	t := w.transactions[0]
 	w.transactions = w.transactions[1:]
 	if frameType == FrameTypeError {