mqtt_test.go

// Copyright 2020-2024 The NATS Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//go:build !skip_mqtt_tests

package server

import (
"bufio"
"bytes"
"crypto/tls"
"encoding/json"
"errors"
"fmt"
"io"
"math/rand"
"net"
"os"
"reflect"
"strings"
"sync"
"testing"
"time"

"github.com/nats-io/jwt/v2"
"github.com/nats-io/nats.go"
"github.com/nats-io/nkeys"
"github.com/nats-io/nuid"
)

vartestMQTTTimeout=10*time.Second

varjsClusterTemplWithLeafAndMQTT=`
 listen: 127.0.0.1:-1
 server_name: %s
 jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s'}

 {{leaf}}

 cluster {
 name: %s
 listen: 127.0.0.1:%d
 routes = [%s]
 }

 mqtt {
 listen: 127.0.0.1:-1
 }

 # For access to system account.
 accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
`

typemqttWrapAsWsstruct {
 net.Conn
t testing.TB
br*bufio.Reader
tmp []byte
}

func (c*mqttWrapAsWs) Write(p []byte) (int, error) {
proto:=testWSCreateClientMsg(wsBinaryMessage, 1, true, false, p)
returnc.Conn.Write(proto)
}

func (c*mqttWrapAsWs) Read(p []byte) (int, error) {
for {
ifn:=len(c.tmp); n>0 {
iflen(p) <n {
n=len(p)
 }
copy(p, c.tmp[:n])
c.tmp=c.tmp[n:]
returnn, nil
 }
c.tmp=testWSReadFrame(c.t, c.br)
 }
}

functestMQTTReadPacket(t testing.TB, r*mqttReader) (byte, int) {
t.Helper()

varbbyte
varplint
varerrerror

rd:=r.reader

fill:=func() []byte {
t.Helper()
varbuf [512]byte

n, err:=rd.Read(buf[:])
iferr!=nil {
t.Fatalf("Error reading data: %v", err)
 }
returncopyBytes(buf[:n])
 }

rd.SetReadDeadline(time.Now().Add(testMQTTTimeout))
for {
r.pstart=r.pos
if!r.hasMore() {
r.reset(fill())
continue
 }
b, err=r.readByte("packet type")
iferr!=nil {
t.Fatalf("Error reading packet: %v", err)
 }
varcompletebool
pl, complete, err=r.readPacketLen()
iferr!=nil {
t.Fatalf("Error reading packet: %v", err)
 }
if!complete {
r.reset(fill())
continue
 }
break
 }
rd.SetReadDeadline(time.Time{})
returnb, pl
}

functestMQTTReadPIPacket(expectedTypebyte, t testing.TB, r*mqttReader, expectedPIuint16) {
t.Helper()
b, _:=testMQTTReadPacket(t, r)
ifpt:=b&mqttPacketMask; pt!=expectedType {
t.Fatalf("Expected packet %x, got %x", expectedType, pt)
 }
rpi, err:=r.readUint16("packet identifier")
iferr!=nil||rpi!=expectedPI {
t.Fatalf("Expected PI %v got: %v, err=%v", expectedPI, rpi, err)
 }
}

funcTestMQTTReader(t*testing.T) {
r:=&mqttReader{}
r.reset([]byte{0, 2, 'a', 'b'})
bs, err:=r.readBytes("", false)
iferr!=nil {
t.Fatal(err)
 }
sbs:=string(bs)
ifsbs!="ab" {
t.Fatalf(`expected "ab", got %q`, sbs)
 }

r.reset([]byte{0, 2, 'a', 'b'})
bs, err=r.readBytes("", true)
iferr!=nil {
t.Fatal(err)
 }
bs[0], bs[1] ='c', 'd'
ifbytes.Equal(bs, r.buf[2:]) {
t.Fatal("readBytes should have returned a copy")
 }

r.reset([]byte{'a', 'b'})
ifb, err:=r.readByte(""); err!=nil||b!='a' {
t.Fatalf("Error reading byte: b=%v err=%v", b, err)
 }
if!r.hasMore() {
t.Fatal("expected to have more, did not")
 }
ifb, err:=r.readByte(""); err!=nil||b!='b' {
t.Fatalf("Error reading byte: b=%v err=%v", b, err)
 }
ifr.hasMore() {
t.Fatal("expected to not have more")
 }
if_, err:=r.readByte("test"); err==nil||!strings.Contains(err.Error(), "error reading test") {
t.Fatalf("unexpected error: %v", err)
 }

r.reset([]byte{0, 2, 'a', 'b'})
ifs, err:=r.readString(""); err!=nil||s!="ab" {
t.Fatalf("Error reading string: s=%q err=%v", s, err)
 }

r.reset([]byte{10})
if_, err:=r.readUint16("uint16"); err==nil||!strings.Contains(err.Error(), "error reading uint16") {
t.Fatalf("unexpected error: %v", err)
 }

r.reset([]byte{0x82, 0xff, 0x3})
l, _, err:=r.readPacketLenWithCheck(false)
iferr!=nil {
t.Fatal("error getting packet len")
 }
ifl!=0xff82 {
t.Fatalf("expected length 0xff82 got 0x%x", l)
 }
r.reset([]byte{0xff, 0xff, 0xff, 0xff, 0xff})
if_, _, err:=r.readPacketLenWithCheck(false); err==nil||!strings.Contains(err.Error(), "malformed") {
t.Fatalf("unexpected error: %v", err)
 }

r.reset([]byte{0, 2, 'a', 'b', mqttPacketPub, 0x82, 0xff, 0x3})
r.readString("")
fori:=0; i<2; i++ {
r.pstart=r.pos
b, err:=r.readByte("")
iferr!=nil {
t.Fatalf("Error reading byte: %v", err)
 }
ifpt:=b&mqttPacketMask; pt!=mqttPacketPub {
t.Fatalf("Unexpected byte: %v", b)
 }
pl, complete, err:=r.readPacketLen()
iferr!=nil {
t.Fatalf("Unexpected error: %v", err)
 }
ifcomplete {
t.Fatal("Expected to be incomplete")
 }
ifpl!=0 {
t.Fatalf("Expected pl to be 0, got %v", pl)
 }
ifi>0 {
break
 }
if!bytes.Equal(r.pbuf, []byte{mqttPacketPub, 0x82, 0xff, 0x3}) {
t.Fatalf("Invalid recorded partial: %v", r.pbuf)
 }
r.reset([]byte{'a', 'b', 'c'})
if!bytes.Equal(r.buf, []byte{mqttPacketPub, 0x82, 0xff, 0x3, 'a', 'b', 'c'}) {
t.Fatalf("Invalid buffer: %v", r.buf)
 }
ifr.pbuf!=nil {
t.Fatalf("Partial buffer should have been reset, got %v", r.pbuf)
 }
ifr.pos!=0 {
t.Fatalf("Pos should have been reset, got %v", r.pos)
 }
ifr.pstart!=0 {
t.Fatalf("Pstart should have been reset, got %v", r.pstart)
 }
 }
// On second pass, the pbuf should have been extended with 'abc'
if!bytes.Equal(r.pbuf, []byte{mqttPacketPub, 0x82, 0xff, 0x3, 'a', 'b', 'c'}) {
t.Fatalf("Invalid recorded partial: %v", r.pbuf)
 }
}

funcTestMQTTWriter(t*testing.T) {
w:=newMQTTWriter(0)
w.WriteUint16(1234)

r:=&mqttReader{}
r.reset(w.Bytes())
ifv, err:=r.readUint16(""); err!=nil||v!=1234 {
t.Fatalf("unexpected value: v=%v err=%v", v, err)
 }

w.Reset()
w.WriteString("test")
r.reset(w.Bytes())
iflen(r.buf) !=6 {
t.Fatalf("Expected 2 bytes size before string, got %v", r.buf)
 }

w.Reset()
w.WriteBytes([]byte("test"))
r.reset(w.Bytes())
iflen(r.buf) !=6 {
t.Fatalf("Expected 2 bytes size before bytes, got %v", r.buf)
 }

ints:= []int{
0, 1, 127, 128, 16383, 16384, 2097151, 2097152, 268435455,
 }
lens:= []int{
1, 1, 1, 2, 2, 3, 3, 4, 4,
 }

tl:=0
w.Reset()
fori, v:=rangeints {
w.WriteVarInt(v)
tl+=lens[i]
iftl!=w.Len() {
t.Fatalf("expected len %d, got %d", tl, w.Len())
 }
 }

r.reset(w.Bytes())
for_, v:=rangeints {
x, _, _:=r.readPacketLenWithCheck(false)
ifv!=x {
t.Fatalf("expected %d, got %d", v, x)
 }
 }
}

functestMQTTDefaultOptions() *Options {
o:=DefaultOptions()
o.ServerName=nuid.Next()
o.Cluster.Port=0
o.Gateway.Name=""
o.Gateway.Port=0
o.LeafNode.Port=0
o.Websocket.Port=0
o.MQTT.Host="127.0.0.1"
o.MQTT.Port=-1
o.JetStream=true
returno
}

functestMQTTRunServer(t testing.TB, o*Options) *Server {
t.Helper()
o.NoLog=false
ifo.StoreDir==_EMPTY_ {
o.StoreDir=t.TempDir()
 }
s, err:=NewServer(o)
iferr!=nil {
t.Fatalf("Error creating server: %v", err)
 }
l:=&DummyLogger{}
s.SetLogger(l, true, true)
s.Start()
iferr:=s.readyForConnections(3*time.Second); err!=nil {
testMQTTShutdownServer(s)
t.Fatal(err)
 }
returns
}

functestMQTTShutdownRestartedServer(s**Server) {
srv:=*s
testMQTTShutdownServer(srv)
*s=nil
}

functestMQTTShutdownServer(s*Server) {
ifc:=s.JetStreamConfig(); c!=nil {
dir:=strings.TrimSuffix(c.StoreDir, JetStreamStoreDir)
deferos.RemoveAll(dir)
 }
s.Shutdown()
}

functestMQTTDefaultTLSOptions(t*testing.T, verifybool) *Options {
t.Helper()
o:=testMQTTDefaultOptions()
tc:=&TLSConfigOpts{
CertFile: "../test/configs/certs/server-cert.pem",
KeyFile: "../test/configs/certs/server-key.pem",
CaFile: "../test/configs/certs/ca.pem",
Verify: verify,
 }
varerrerror
o.MQTT.TLSConfig, err=GenTLSConfig(tc)
o.MQTT.TLSTimeout=2.0
iferr!=nil {
t.Fatalf("Error creating tls config: %v", err)
 }
returno
}

funcTestMQTTServerNameRequired(t*testing.T) {
conf:=createConfFile(t, []byte(`
 cluster {
 port: -1
 }
 mqtt {
 port: -1
 }
 `))
o, err:=ProcessConfigFile(conf)
iferr!=nil {
t.Fatalf("Error processing config file: %v", err)
 }
if_, err:=NewServer(o); err==nil||err.Error() !=errMQTTServerNameMustBeSet.Error() {
t.Fatalf("Expected error about requiring server name to be set, got %v", err)
 }

conf=createConfFile(t, []byte(`
 mqtt {
 port: -1
 }
 `))
o, err=ProcessConfigFile(conf)
iferr!=nil {
t.Fatalf("Error processing config file: %v", err)
 }
if_, err:=NewServer(o); err==nil||err.Error() !=errMQTTStandaloneNeedsJetStream.Error() {
t.Fatalf(`Expected errMQTTStandaloneNeedsJetStream ("next in line"), got %v`, err)
 }
}

funcTestMQTTStandaloneRequiresJetStream(t*testing.T) {
conf:=createConfFile(t, []byte(`
 server_name: mqtt
 mqtt {
 port: -1
 tls {
 cert_file: "./configs/certs/server.pem"
 key_file: "./configs/certs/key.pem"
 }
 }
 `))
o, err:=ProcessConfigFile(conf)
iferr!=nil {
t.Fatalf("Error processing config file: %v", err)
 }
if_, err:=NewServer(o); err==nil||err.Error() !=errMQTTStandaloneNeedsJetStream.Error() {
t.Fatalf("Expected error about requiring JetStream in standalone mode, got %v", err)
 }
}

funcTestMQTTConfig(t*testing.T) {
conf:=createConfFile(t, []byte(fmt.Sprintf(`
 jetstream {
 store_dir = %q
 }
 server_name: mqtt
 mqtt {
 port: -1
 tls {
 cert_file: "./configs/certs/server.pem"
 key_file: "./configs/certs/key.pem"
 }
 }
 `, t.TempDir())))
s, o:=RunServerWithConfig(conf)
defertestMQTTShutdownServer(s)
ifo.MQTT.TLSConfig==nil {
t.Fatal("expected TLS config to be set")
 }
}

funcTestMQTTValidateOptions(t*testing.T) {
nmqtto:=DefaultOptions()
mqtto:=testMQTTDefaultOptions()
for_, test:=range []struct {
namestring
getOptsfunc() *Options
errerror
 }{
 {"mqtt disabled", func() *Options { returnnmqtto.Clone() }, nil},
 {"mqtt username not allowed if users specified", func() *Options {
o:=mqtto.Clone()
o.Users= []*User{{Username: "abc", Password: "pwd"}}
o.MQTT.Username="b"
o.MQTT.Password="pwd"
returno
 }, errMQTTUserMixWithUsersNKeys},
 {"mqtt token not allowed if users specified", func() *Options {
o:=mqtto.Clone()
o.Nkeys= []*NkeyUser{{Nkey: "abc"}}
o.MQTT.Token="mytoken"
returno
 }, errMQTTTokenMixWIthUsersNKeys},
 {"ack wait should be >=0", func() *Options {
o:=mqtto.Clone()
o.MQTT.AckWait=-10*time.Second
returno
 }, errMQTTAckWaitMustBePositive},
 } {
t.Run(test.name, func(t*testing.T) {
err:=validateMQTTOptions(test.getOpts())
iftest.err==nil&&err!=nil {
t.Fatalf("Unexpected error: %v", err)
 } elseiftest.err!=nil&& (err==nil||err.Error() !=test.err.Error()) {
t.Fatalf("Expected error to contain %q, got %v", test.err, err)
 }
 })
 }
}

funcTestMQTTParseOptions(t*testing.T) {
for_, test:=range []struct {
namestring
contentstring
checkOptfunc(*MQTTOpts) error
errstring
 }{
// Negative tests
 {"bad type", "mqtt: []", nil, "to be a map"},
 {"bad listen", "mqtt: { listen: [] }", nil, "port or host:port"},
 {"bad port", `mqtt: { port: "abc" }`, nil, "not int64"},
 {"bad host", `mqtt: { host: 123 }`, nil, "not string"},
 {"bad tls", `mqtt: { tls: 123 }`, nil, "not map[string]interface {}"},
 {"unknown field", `mqtt: { this_does_not_exist: 123 }`, nil, "unknown"},
 {"ack wait", `mqtt: {ack_wait: abc}`, nil, "invalid duration"},
 {"max ack pending", `mqtt: {max_ack_pending: abc}`, nil, "not int64"},
 {"max ack pending too high", `mqtt: {max_ack_pending: 12345678}`, nil, "invalid value"},
// Positive tests
 {"tls gen fails", `
 mqtt {
 tls {
 cert_file: "./configs/certs/server.pem"
 }
 }`, nil, "missing 'key_file'"},
 {"listen port only", `mqtt { listen: 1234 }`, func(o*MQTTOpts) error {
ifo.Port!=1234 {
returnfmt.Errorf("expected 1234, got %v", o.Port)
 }
returnnil
 }, ""},
 {"listen host and port", `mqtt { listen: "localhost:1234" }`, func(o*MQTTOpts) error {
ifo.Host!="localhost"||o.Port!=1234 {
returnfmt.Errorf("expected localhost:1234, got %v:%v", o.Host, o.Port)
 }
returnnil
 }, ""},
 {"host", `mqtt { host: "localhost" }`, func(o*MQTTOpts) error {
ifo.Host!="localhost" {
returnfmt.Errorf("expected localhost, got %v", o.Host)
 }
returnnil
 }, ""},
 {"port", `mqtt { port: 1234 }`, func(o*MQTTOpts) error {
ifo.Port!=1234 {
returnfmt.Errorf("expected 1234, got %v", o.Port)
 }
returnnil
 }, ""},
 {"tls config",
`
 mqtt {
 tls {
 cert_file: "./configs/certs/server.pem"
 key_file: "./configs/certs/key.pem"
 }
 }
 `, func(o*MQTTOpts) error {
ifo.TLSConfig==nil {
returnfmt.Errorf("TLSConfig should have been set")
 }
returnnil
 }, ""},
 {"no auth user",
`
 mqtt {
 no_auth_user: "noauthuser"
 }
 `, func(o*MQTTOpts) error {
ifo.NoAuthUser!="noauthuser" {
returnfmt.Errorf("Invalid NoAuthUser value: %q", o.NoAuthUser)
 }
returnnil
 }, ""},
 {"auth block",
`
 mqtt {
 authorization {
 user: "mqttuser"
 password: "pwd"
 token: "token"
 timeout: 2.0
 }
 }
 `, func(o*MQTTOpts) error {
ifo.Username!="mqttuser"||o.Password!="pwd"||o.Token!="token"||o.AuthTimeout!=2.0 {
returnfmt.Errorf("Invalid auth block: %+v", o)
 }
returnnil
 }, ""},
 {"auth timeout as int",
`
 mqtt {
 authorization {
 timeout: 2
 }
 }
 `, func(o*MQTTOpts) error {
ifo.AuthTimeout!=2.0 {
returnfmt.Errorf("Invalid auth timeout: %v", o.AuthTimeout)
 }
returnnil
 }, ""},
 {"ack wait",
`
 mqtt {
 ack_wait: "10s"
 }
 `, func(o*MQTTOpts) error {
ifo.AckWait!=10*time.Second {
returnfmt.Errorf("Invalid ack wait: %v", o.AckWait)
 }
returnnil
 }, ""},
 {"max ack pending",
`
 mqtt {
 max_ack_pending: 123
 }
 `, func(o*MQTTOpts) error {
ifo.MaxAckPending!=123 {
returnfmt.Errorf("Invalid max ack pending: %v", o.MaxAckPending)
 }
returnnil
 }, ""},
 {"reject_qos2_publish",
`
 mqtt {
 reject_qos2_publish: true
 }
 `, func(o*MQTTOpts) error {
if!o.rejectQoS2Pub {
returnfmt.Errorf("Invalid: expected rejectQoS2Pub to be set")
 }
returnnil
 }, ""},
 {"downgrade_qos2_subscribe",
`
 mqtt {
 downgrade_qos2_subscribe: true
 }
 `, func(o*MQTTOpts) error {
if!o.downgradeQoS2Sub {
returnfmt.Errorf("Invalid: expected downgradeQoS2Sub to be set")
 }
returnnil
 }, ""},
 } {
t.Run(test.name, func(t*testing.T) {
conf:=createConfFile(t, []byte(test.content))
o, err:=ProcessConfigFile(conf)
iftest.err!=_EMPTY_ {
iferr==nil||!strings.Contains(err.Error(), test.err) {
t.Fatalf("For content: %q, expected error about %q, got %v", test.content, test.err, err)
 }
return
 } elseiferr!=nil {
t.Fatalf("Unexpected error for content %q: %v", test.content, err)
 }
iferr:=test.checkOpt(&o.MQTT); err!=nil {
t.Fatalf("Incorrect option for content %q: %v", test.content, err.Error())
 }
 })
 }
}

funcTestMQTTStart(t*testing.T) {
o:=testMQTTDefaultOptions()
s:=testMQTTRunServer(t, o)
defertestMQTTShutdownServer(s)

nc, err:=net.Dial("tcp", fmt.Sprintf("%s:%d", o.MQTT.Host, o.MQTT.Port))
iferr!=nil {
t.Fatalf("Unable to create tcp connection to mqtt port: %v", err)
 }
nc.Close()

// Check failure to start due to port in use
o2:=testMQTTDefaultOptions()
o2.MQTT.Port=o.MQTT.Port
s2, err:=NewServer(o2)
iferr!=nil {
t.Fatalf("Error creating server: %v", err)
 }
defers2.Shutdown()
l:=&captureFatalLogger{fatalCh: make(chanstring, 1)}
s2.SetLogger(l, false, false)

wg:= sync.WaitGroup{}
wg.Add(1)
gofunc() {
s2.Start()
wg.Done()
 }()

select {
casee:=<-l.fatalCh:
if!strings.Contains(e, "Unable to listen for MQTT connections") {
t.Fatalf("Unexpected error: %q", e)
 }
case<-time.After(time.Second):
t.Fatal("Should have gotten a fatal error")
 }
}

funcTestMQTTTLS(t*testing.T) {
o:=testMQTTDefaultTLSOptions(t, false)
s:=testMQTTRunServer(t, o)
defertestMQTTShutdownServer(s)

c, _, err:=testMQTTConnectRetryWithError(t, &mqttConnInfo{tls: true}, o.MQTT.Host, o.MQTT.Port, 0)
iferr!=nil {
t.Fatal(err)
 }
c.Close()
c=nil
testMQTTShutdownServer(s)

// Force client cert verification
o=testMQTTDefaultTLSOptions(t, true)
s=testMQTTRunServer(t, o)
defertestMQTTShutdownServer(s)

c, _, err=testMQTTConnectRetryWithError(t, &mqttConnInfo{tls: true}, o.MQTT.Host, o.MQTT.Port, 0)
iferr==nil||c!=nil {
ifc!=nil {
c.Close()
 }
t.Fatal("Handshake expected to fail since client did not provide cert")
 }

// Add client cert.
tc:=&TLSConfigOpts{
CertFile: "../test/configs/certs/client-cert.pem",
KeyFile: "../test/configs/certs/client-key.pem",
 }
tlsc, err:=GenTLSConfig(tc)
iferr!=nil {
t.Fatalf("Error generating tls config: %v", err)
 }
tlsc.InsecureSkipVerify=true
c, _, err=testMQTTConnectRetryWithError(t, &mqttConnInfo{
tls: true,
tlsc: tlsc,
 }, o.MQTT.Host, o.MQTT.Port, 0)
iferr!=nil {
t.Fatal(err)
 }
c.Close()
c=nil
testMQTTShutdownServer(s)

// Lower TLS timeout so low that we should fail
o.MQTT.TLSTimeout=0.001
s=testMQTTRunServer(t, o)
defertestMQTTShutdownServer(s)

nc, err:=net.Dial("tcp", fmt.Sprintf("%s:%d", o.MQTT.Host, o.MQTT.Port))
iferr!=nil {
t.Fatalf("Unable to create tcp connection to mqtt port: %v", err)
 }
defernc.Close()
time.Sleep(100*time.Millisecond)
tlsConn:=tls.Client(nc, tlsc)
tlsConn.SetDeadline(time.Now().Add(time.Second))
iferr:=tlsConn.Handshake(); err==nil {
t.Fatal("Expected failure, did not get one")
 }
}

typemqttConnInfostruct {
clientIDstring
cleanSessbool
keepAliveuint16
will*mqttWill
userstring
passstring
wsbool
tlsbool
tlsc*tls.Config
}

functestMQTTGetClient(t testing.TB, s*Server, clientIDstring) *client {
t.Helper()
varmc*client
s.mu.Lock()
for_, c:=ranges.clients {
c.mu.Lock()
ifc.isMqtt() &&c.mqtt.cid==clientID {
mc=c
 }
c.mu.Unlock()
ifmc!=nil {
break
 }
 }
s.mu.Unlock()
ifmc==nil {
t.Fatalf("Did not find client %q", clientID)
 }
returnmc
}

functestMQTTRead(c net.Conn) ([]byte, error) {
varbuf [512]byte
// Make sure that test does not block
c.SetReadDeadline(time.Now().Add(testMQTTTimeout))
n, err:=c.Read(buf[:])
iferr!=nil {
returnnil, err
 }
c.SetReadDeadline(time.Time{})
returncopyBytes(buf[:n]), nil
}

functestMQTTWrite(c net.Conn, buf []byte) (int, error) {
c.SetWriteDeadline(time.Now().Add(testMQTTTimeout))
n, err:=c.Write(buf)
c.SetWriteDeadline(time.Time{})
returnn, err
}

functestMQTTConnect(t testing.TB, ci*mqttConnInfo, hoststring, portint) (net.Conn, *mqttReader) {
t.Helper()
returntestMQTTConnectRetry(t, ci, host, port, 0)
}

functestMQTTConnectRetry(t testing.TB, ci*mqttConnInfo, hoststring, portint, retryCountint) (net.Conn, *mqttReader) {
t.Helper()
c, r, err:=testMQTTConnectRetryWithError(t, ci, host, port, retryCount)
iferr!=nil {
t.Fatal(err)
 }
returnc, r
}

functestMQTTConnectRetryWithError(t testing.TB, ci*mqttConnInfo, hoststring, portint, retryCountint) (net.Conn, *mqttReader, error) {
retry:=func(c net.Conn) bool {
ifc!=nil {
c.Close()
 }
ifretryCount==0 {
returnfalse
 }
time.Sleep(time.Second)
retryCount--
returntrue
 }

addr:=fmt.Sprintf("%s:%d", host, port)
varc net.Conn
varerrerror
RETRY:
ifci.ws {
varbr*bufio.Reader
c, br, _, err=testNewWSClientWithError(t, testWSClientOptions{
host: host,
port: port,
noTLS: !ci.tls,
path: mqttWSPath,
 })
iferr==nil {
c=&mqttWrapAsWs{Conn: c, t: t, br: br}
 }
 } else {
c, err=net.Dial("tcp", addr)
iferr==nil&&ci.tls {
tc:=ci.tlsc
iftc==nil {
tc=&tls.Config{InsecureSkipVerify: true}
 }
c=tls.Client(c, tc)
c.SetDeadline(time.Now().Add(time.Second))
err=c.(*tls.Conn).Handshake()
c.SetDeadline(time.Time{})
 }
 }
iferr!=nil {
ifretry(c) {
goto RETRY
 }
returnnil, nil, fmt.Errorf("Error creating mqtt connection: %v", err)
 }

proto:=mqttCreateConnectProto(ci)
if_, err:=testMQTTWrite(c, proto); err!=nil {
ifretry(c) {
goto RETRY
 }
returnnil, nil, fmt.Errorf("Error writing connect: %v", err)
 }

buf, err:=testMQTTRead(c)
iferr!=nil {
ifretry(c) {
goto RETRY
 }
returnnil, nil, fmt.Errorf("Error reading: %v", err)
 }
mr:=&mqttReader{reader: c}
mr.reset(buf)

returnc, mr, nil
}

funcmqttCreateConnectProto(ci*mqttConnInfo) []byte {
flags:=byte(0)
ifci.cleanSess {
flags|=mqttConnFlagCleanSession
 }
ifci.will!=nil {
flags|=mqttConnFlagWillFlag| (ci.will.qos<<3)
ifci.will.retain {
flags|=mqttConnFlagWillRetain
 }
 }
ifci.user!=_EMPTY_ {
flags|=mqttConnFlagUsernameFlag
 }
ifci.pass!=_EMPTY_ {
flags|=mqttConnFlagPasswordFlag
 }

pkLen:=2+len(mqttProtoName) +
1+// proto level
1+// flags
2+// keepAlive
2+len(ci.clientID)

ifci.will!=nil {
pkLen+=2+len(ci.will.topic)
pkLen+=2+len(ci.will.message)
 }
ifci.user!=_EMPTY_ {
pkLen+=2+len(ci.user)
 }
ifci.pass!=_EMPTY_ {
pkLen+=2+len(ci.pass)
 }

w:=newMQTTWriter(0)
w.WriteByte(mqttPacketConnect)
w.WriteVarInt(pkLen)
w.WriteString(string(mqttProtoName))
w.WriteByte(0x4)
w.WriteByte(flags)
w.WriteUint16(ci.keepAlive)
w.WriteString(ci.clientID)
ifci.will!=nil {
w.WriteBytes(ci.will.topic)
w.WriteBytes(ci.will.message)
 }
ifci.user!=_EMPTY_ {
w.WriteString(ci.user)
 }
ifci.pass!=_EMPTY_ {
w.WriteBytes([]byte(ci.pass))
 }
returnw.Bytes()
}

functestMQTTCheckConnAck(t testing.TB, r*mqttReader, rcbyte, sessionPresentbool) {
t.Helper()
b, pl:=testMQTTReadPacket(t, r)
pt:=b&mqttPacketMask
ifpt!=mqttPacketConnectAck {
t.Fatalf("Expected ConnAck (%x), got %x", mqttPacketConnectAck, pt)
 }
ifpl!=2 {
t.Fatalf("ConnAck packet length should be 2, got %v", pl)
 }
caf, err:=r.readByte("connack flags")
iferr!=nil {
t.Fatalf("Error reading packet length: %v", err)
 }
ifcaf&0xfe!=0 {
t.Fatalf("ConnAck flag bits 7-1 should all be 0, got %x", caf>>1)
 }
ifsp:=caf==1; sp!=sessionPresent {
t.Fatalf("Expected session present flag=%v got %v", sessionPresent, sp)
 }
carc, err:=r.readByte("connack return code")
iferr!=nil {
t.Fatalf("Error reading returned code: %v", err)
 }
ifcarc!=rc {
t.Fatalf("Expected return code to be %v, got %v", rc, carc)
 }
}

functestMQTTCheckPubAck(t testing.TB, r*mqttReader, packetTypebyte) {
t.Helper()
b, pl:=testMQTTReadPacket(t, r)
pt:=b&mqttPacketMask
ifpt!=packetType {
t.Fatalf("Expected %x, got %x", packetType, pt)
 }
r.pos+=pl
}

funcTestMQTTRequiresJSEnabled(t*testing.T) {
o:=testMQTTDefaultOptions()
acc:=NewAccount("mqtt")
o.Accounts= []*Account{acc}
o.Users= []*User{{Username: "mqtt", Account: acc}}
s:=testMQTTRunServer(t, o)
defertestMQTTShutdownServer(s)

addr:=fmt.Sprintf("%s:%d", o.MQTT.Host, o.MQTT.Port)
c, err:=net.Dial("tcp", addr)
iferr!=nil {
t.Fatalf("Error creating mqtt connection: %v", err)
 }
deferc.Close()

proto:=mqttCreateConnectProto(&mqttConnInfo{cleanSess: true, user: "mqtt"})
if_, err:=testMQTTWrite(c, proto); err!=nil {
t.Fatalf("Error writing connect: %v", err)