Use context to signal client open state

If SessionExpiryInterval was not set in CONNECT, SessionExpiryIntervalFlag is also not set. According to spec:
  If the Session Expiry Interval is absent the value 0 is used. If it is set to 0, or is absent, the Session ends when the Network Connection is closed.

.github/workflows/build.yml

@@ -7,37 +7,39 @@ jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v3
 
     - name: Set up Go
-      uses: actions/setup-go@v2
+      uses: actions/setup-go@v3
       with:
         go-version: 1.19
 
     - name: Vet
       run: go vet ./...
-    
+
     - name: Test
       run: go test -race ./... && echo true
       
 
   coverage:
+    name: Test with Coverage
     runs-on: ubuntu-latest
     steps:
-      - name: Install Go
-        if: success()
-        uses: actions/setup-go@v2
-        with:
-          go-version: 1.19.x
-      - name: Checkout code
-        uses: actions/checkout@v2
-      - name: Calc coverage
-        run: |
-          go test -v -covermode=count -coverprofile=coverage.out ./...
-      - name: Convert coverage.out to coverage.lcov
-        uses: jandelgado/gcov2lcov-action@v1.0.6
-      - name: Coveralls
-        uses: coverallsapp/github-action@v1.1.2
-        with:
-          github-token: ${{ secrets.github_token }}
-          path-to-lcov: coverage.lcov
+    - name: Set up Go
+      uses: actions/setup-go@v3
+      with:
+        go-version: '1.19'
+    - name: Check out code
+      uses: actions/checkout@v3
+    - name: Install dependencies
+      run: |
+        go mod download
+    - name: Run Unit tests
+      run: |
+        go test -race -covermode atomic -coverprofile=covprofile ./...
+    - name: Install goveralls
+      run: go install github.com/mattn/goveralls@latest
+    - name: Send coverage
+      env:
+        COVERALLS_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      run: goveralls -coverprofile=covprofile -service=github

README.md

@@ -304,6 +304,7 @@ The function signatures for all the hooks and `mqtt.Hook` interface can be found
 | OnQosPublish | Called when a publish packet with Qos >= 1 is issued to a subscriber. | 
 | OnQosComplete | Called when the Qos flow for a message has been completed. | 
 | OnQosDropped | Called when an inflight message expires before completion. | 
+| OnPacketIDExhausted | Called when a client runs out of unused packet ids to assign. | 
 | OnWill | Called when a client disconnects and intends to issue a will message. Allows packet modification. | 
 | OnWillSent | Called when an LWT message has been issued from a disconnecting client. | 
 | OnClientExpired | Called when a client session has expired and should be deleted. | 

clients.go

@@ -7,6 +7,7 @@ package mqtt
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"fmt"
 	"io"
 	"net"
@@ -87,7 +88,7 @@ func (cl *Clients) GetByListener(id string) []*Client {
 	defer cl.RUnlock()
 	clients := make([]*Client, 0, cl.Len())
 	for _, client := range cl.internal {
-		if client.Net.Listener == id && atomic.LoadUint32(&client.State.done) == 0 {
+		if client.Net.Listener == id && !client.Closed() {
 			clients = append(clients, client)
 		}
 	}
@@ -144,7 +145,7 @@ type ClientState struct {
 	endOnce       sync.Once            // only end once
 	isTakenOver   uint32               // used to identify orphaned clients
 	packetID      uint32               // the current highest packetID
-	done          uint32               // atomic counter which indicates that the client has closed
+	open          context.Context      // indicate that the client is open for packet exchange
 	outboundQty   int32                // number of messages currently in the outbound queue
 	keepalive     uint16               // the number of seconds the connection can wait
 }
@@ -158,6 +159,7 @@ func newClient(c net.Conn, o *ops) *Client {
 			Subscriptions: NewSubscriptions(),
 			TopicAliases:  NewTopicAliases(o.options.Capabilities.TopicAliasMaximum),
 			keepalive:     defaultKeepalive,
+			open:          context.Background(),
 			outbound:      make(chan *packets.Packet, o.options.Capabilities.MaximumClientWritesPending),
 		},
 		Properties: ClientProperties{
@@ -330,7 +332,7 @@ func (cl *Client) Read(packetHandler ReadFn) error {
 	var err error
 
 	for {
-		if atomic.LoadUint32(&cl.State.done) == 1 {
+		if cl.Closed() {
 			return nil
 		}
 
@@ -356,6 +358,9 @@ func (cl *Client) Read(packetHandler ReadFn) error {
 // Stop instructs the client to shut down all processing goroutines and disconnect.
 func (cl *Client) Stop(err error) {
 	cl.State.endOnce.Do(func() {
+		cl.Lock()
+		defer cl.Unlock()
+
 		if cl.Net.Conn != nil {
 			_ = cl.Net.Conn.Close() // omit close error
 		}
@@ -368,7 +373,12 @@ func (cl *Client) Stop(err error) {
 			close(cl.State.outbound)
 		}
 
-		atomic.StoreUint32(&cl.State.done, 1)
+		if cl.State.open != nil {
+			var cancel context.CancelFunc
+			cl.State.open, cancel = context.WithCancel(cl.State.open)
+			cancel()
+		}
+
 		atomic.StoreInt64(&cl.State.disconnected, time.Now().Unix())
 	})
 }
@@ -383,7 +393,7 @@ func (cl *Client) StopCause() error {
 
 // Closed returns true if client connection is closed.
 func (cl *Client) Closed() bool {
-	return atomic.LoadUint32(&cl.State.done) == 1
+	return cl.State.open == nil || cl.State.open.Err() != nil
 }
 
 // ReadFixedHeader reads in the values of the next packet's fixed header.
@@ -479,14 +489,6 @@ func (cl *Client) ReadPacket(fh *packets.FixedHeader) (pk packets.Packet, err er
 
 // WritePacket encodes and writes a packet to the client.
 func (cl *Client) WritePacket(pk packets.Packet) error {
-	if atomic.LoadUint32(&cl.State.done) == 1 {
-		return ErrConnectionClosed
-	}
-
-	if cl.Net.Conn == nil {
-		return nil
-	}
-
 	if pk.Expiry > 0 {
 		pk.Properties.MessageExpiryInterval = uint32(pk.Expiry - time.Now().Unix()) // [MQTT-3.3.2-6]
 	}
@@ -550,6 +552,17 @@ func (cl *Client) WritePacket(pk packets.Packet) error {
 		return packets.ErrPacketTooLarge // [MQTT-3.1.2-24] [MQTT-3.1.2-25]
 	}
 
+	cl.Lock()
+	defer cl.Unlock()
+
+	if cl.Closed() {
+		return ErrConnectionClosed
+	}
+
+	if cl.Net.Conn == nil {
+		return nil
+	}
+
 	nb := net.Buffers{buf.Bytes()}
 	n, err := nb.WriteTo(cl.Net.Conn)
 	if err != nil {

clients_test.go

@@ -5,6 +5,7 @@
 package mqtt
 
 import (
+	"context"
 	"errors"
 	"io"
 	"net"
@@ -114,8 +115,8 @@ func TestClientsDelete(t *testing.T) {
 
 func TestClientsGetByListener(t *testing.T) {
 	cl := NewClients()
-	cl.Add(&Client{ID: "t1", Net: ClientConnection{Listener: "tcp1"}})
-	cl.Add(&Client{ID: "t2", Net: ClientConnection{Listener: "ws1"}})
+	cl.Add(&Client{ID: "t1", State: ClientState{open: context.Background()}, Net: ClientConnection{Listener: "tcp1"}})
+	cl.Add(&Client{ID: "t2", State: ClientState{open: context.Background()}, Net: ClientConnection{Listener: "ws1"}})
 	require.Contains(t, cl.internal, "t1")
 	require.Contains(t, cl.internal, "t2")
 
@@ -466,7 +467,7 @@ func TestClientReadOK(t *testing.T) {
 func TestClientReadDone(t *testing.T) {
 	cl, _, _ := newTestClient()
 	defer cl.Stop(errClientStop)
-	cl.State.done = 1
+	cl.State.open = nil
 
 	o := make(chan error)
 	go func() {
@@ -483,7 +484,7 @@ func TestClientStop(t *testing.T) {
 	cl.Stop(nil)
 	require.Equal(t, nil, cl.State.stopCause.Load())
 	require.Equal(t, time.Now().Unix(), cl.State.disconnected)
-	require.Equal(t, uint32(1), cl.State.done)
+	require.True(t, cl.Closed())
 	require.Equal(t, nil, cl.StopCause())
 }
 

hooks.go

@@ -44,6 +44,7 @@ const (
 	OnQosPublish
 	OnQosComplete
 	OnQosDropped
+	OnPacketIDExhausted
 	OnWill
 	OnWillSent
 	OnClientExpired
@@ -93,6 +94,7 @@ type Hook interface {
 	OnQosPublish(cl *Client, pk packets.Packet, sent int64, resends int)
 	OnQosComplete(cl *Client, pk packets.Packet)
 	OnQosDropped(cl *Client, pk packets.Packet)
+	OnPacketIDExhausted(cl *Client, pk packets.Packet)
 	OnWill(cl *Client, will Will) (Will, error)
 	OnWillSent(cl *Client, pk packets.Packet)
 	OnClientExpired(cl *Client)
@@ -447,6 +449,16 @@ func (h *Hooks) OnQosDropped(cl *Client, pk packets.Packet) {
 	}
 }
 
+// OnPacketIDExhausted is called when the client runs out of unused packet ids to
+// assign to a packet.
+func (h *Hooks) OnPacketIDExhausted(cl *Client, pk packets.Packet) {
+	for _, hook := range h.GetAll() {
+		if hook.Provides(OnPacketIDExhausted) {
+			hook.OnPacketIDExhausted(cl, pk)
+		}
+	}
+}
+
 // OnWill is called when a client disconnects and publishes an LWT message. This method
 // differs from OnWillSent in that it allows you to modify the LWT message before it is
 // published. The return values of the hook methods are passed-through in the order
@@ -754,6 +766,9 @@ func (h *HookBase) OnQosComplete(cl *Client, pk packets.Packet) {}
 // OnQosDropped is called the Qos flow for a message expires.
 func (h *HookBase) OnQosDropped(cl *Client, pk packets.Packet) {}
 
+// OnPacketIDExhausted is called when the client runs out of unused packet ids to assign to a packet.
+func (h *HookBase) OnPacketIDExhausted(cl *Client, pk packets.Packet) {}
+
 // OnWill is called when a client disconnects and publishes an LWT message.
 func (h *HookBase) OnWill(cl *Client, will Will) (Will, error) {
 	return will, nil

hooks_test.go

@@ -241,6 +241,7 @@ func TestHooksNonReturns(t *testing.T) {
 			h.OnQosPublish(cl, packets.Packet{}, time.Now().Unix(), 0)
 			h.OnQosComplete(cl, packets.Packet{})
 			h.OnQosDropped(cl, packets.Packet{})
+			h.OnPacketIDExhausted(cl, packets.Packet{})
 			h.OnWillSent(cl, packets.Packet{})
 			h.OnClientExpired(cl)
 			h.OnRetainedExpired("a/b/c")

packets/codes.go

@@ -21,7 +21,7 @@ func (c Code) Error() string {
 }
 
 var (
-	// QosCodes indicicates the reason codes for each Qos byte.
+	// QosCodes indicates the reason codes for each Qos byte.
 	QosCodes = map[byte]Code{
 		0: CodeGrantedQos0,
 		1: CodeGrantedQos1,
@@ -120,7 +120,7 @@ var (
 	ErrQosNotSupported                        = Code{Code: 0x9B, Reason: "qos not supported"}
 	ErrUseAnotherServer                       = Code{Code: 0x9C, Reason: "use another server"}
 	ErrServerMoved                            = Code{Code: 0x9D, Reason: "server moved"}
-	ErrSharedSubscriptionsNotSupported        = Code{Code: 0x9E, Reason: "shared subscriptiptions not supported"}
+	ErrSharedSubscriptionsNotSupported        = Code{Code: 0x9E, Reason: "shared subscriptions not supported"}
 	ErrConnectionRateExceeded                 = Code{Code: 0x9F, Reason: "connection rate exceeded"}
 	ErrMaxConnectTime                         = Code{Code: 0xA0, Reason: "maximum connect time"}
 	ErrSubscriptionIdentifiersNotSupported    = Code{Code: 0xA1, Reason: "subscription identifiers not supported"}

packets/packets.go

@@ -16,22 +16,23 @@ import (
 
 // All of the valid packet types and their packet identifier.
 const (
-	Reserved    byte = iota // 0 - we use this in packet tests to indicate special-test or all packets.
-	Connect                 // 1
-	Connack                 // 2
-	Publish                 // 3
-	Puback                  // 4
-	Pubrec                  // 5
-	Pubrel                  // 6
-	Pubcomp                 // 7
-	Subscribe               // 8
-	Suback                  // 9
-	Unsubscribe             // 10
-	Unsuback                // 11
-	Pingreq                 // 12
-	Pingresp                // 13
-	Disconnect              // 14
-	Auth                    // 15
+	Reserved       byte = iota // 0 - we use this in packet tests to indicate special-test or all packets.
+	Connect                    // 1
+	Connack                    // 2
+	Publish                    // 3
+	Puback                     // 4
+	Pubrec                     // 5
+	Pubrel                     // 6
+	Pubcomp                    // 7
+	Subscribe                  // 8
+	Suback                     // 9
+	Unsubscribe                // 10
+	Unsuback                   // 11
+	Pingreq                    // 12
+	Pingresp                   // 13
+	Disconnect                 // 14
+	Auth                       // 15
+	WillProperties byte = 99   // Special byte for validating Will Properties.
 )
 
 var (
@@ -313,7 +314,7 @@ func (pk *Packet) ConnectEncode(buf *bytes.Buffer) error {
 
 	if pk.ProtocolVersion == 5 {
 		pb := bytes.NewBuffer([]byte{})
-		(&pk.Properties).Encode(pk, pb, 0)
+		(&pk.Properties).Encode(pk.FixedHeader.Type, pk.Mods, pb, 0)
 		nb.Write(pb.Bytes())
 	}
 
@@ -322,7 +323,7 @@ func (pk *Packet) ConnectEncode(buf *bytes.Buffer) error {
 	if pk.Connect.WillFlag {
 		if pk.ProtocolVersion == 5 {
 			pb := bytes.NewBuffer([]byte{})
-			(&pk.Connect).WillProperties.Encode(pk, pb, 0)
+			(&pk.Connect).WillProperties.Encode(WillProperties, pk.Mods, pb, 0)
 			nb.Write(pb.Bytes())
 		}
 
@@ -393,7 +394,7 @@ func (pk *Packet) ConnectDecode(buf []byte) error {
 
 	if pk.Connect.WillFlag { // [MQTT-3.1.2-7]
 		if pk.ProtocolVersion == 5 {
-			n, err := pk.Connect.WillProperties.Decode(pk.FixedHeader.Type, bytes.NewBuffer(buf[offset:]))
+			n, err := pk.Connect.WillProperties.Decode(WillProperties, bytes.NewBuffer(buf[offset:]))
 			if err != nil {
 				return ErrMalformedWillProperties
 			}
@@ -496,7 +497,7 @@ func (pk *Packet) ConnackEncode(buf *bytes.Buffer) error {
 
 	if pk.ProtocolVersion == 5 {
 		pb := bytes.NewBuffer([]byte{})
-		pk.Properties.Encode(pk, pb, nb.Len()+2) // +SessionPresent +ReasonCode
+		pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len()+2) // +SessionPresent +ReasonCode
 		nb.Write(pb.Bytes())
 	}
 
@@ -539,7 +540,7 @@ func (pk *Packet) DisconnectEncode(buf *bytes.Buffer) error {
 		nb.WriteByte(pk.ReasonCode)
 
 		pb := bytes.NewBuffer([]byte{})
-		pk.Properties.Encode(pk, pb, nb.Len())
+		pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len())
 		nb.Write(pb.Bytes())
 	}
 
@@ -608,7 +609,7 @@ func (pk *Packet) PublishEncode(buf *bytes.Buffer) error {
 
 	if pk.ProtocolVersion == 5 {
 		pb := bytes.NewBuffer([]byte{})
-		pk.Properties.Encode(pk, pb, nb.Len()+len(pk.Payload))
+		pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len()+len(pk.Payload))
 		nb.Write(pb.Bytes())
 	}
 
@@ -692,7 +693,7 @@ func (pk *Packet) encodePubAckRelRecComp(buf *bytes.Buffer) error {
 
 	if pk.ProtocolVersion == 5 {
 		pb := bytes.NewBuffer([]byte{})
-		pk.Properties.Encode(pk, pb, nb.Len())
+		pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len())
 		if pk.ReasonCode >= ErrUnspecifiedError.Code || pb.Len() > 1 {
 			nb.WriteByte(pk.ReasonCode)
 		}
@@ -833,7 +834,7 @@ func (pk *Packet) SubackEncode(buf *bytes.Buffer) error {
 
 	if pk.ProtocolVersion == 5 {
 		pb := bytes.NewBuffer([]byte{})
-		pk.Properties.Encode(pk, pb, nb.Len()+len(pk.ReasonCodes))
+		pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len()+len(pk.ReasonCodes))
 		nb.Write(pb.Bytes())
 	}
 
@@ -890,7 +891,7 @@ func (pk *Packet) SubscribeEncode(buf *bytes.Buffer) error {
 
 	if pk.ProtocolVersion == 5 {
 		pb := bytes.NewBuffer([]byte{})
-		pk.Properties.Encode(pk, pb, nb.Len()+xb.Len())
+		pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len()+xb.Len())
 		nb.Write(pb.Bytes())
 	}
 
@@ -985,7 +986,7 @@ func (pk *Packet) UnsubackEncode(buf *bytes.Buffer) error {
 
 	if pk.ProtocolVersion == 5 {
 		pb := bytes.NewBuffer([]byte{})
-		pk.Properties.Encode(pk, pb, nb.Len())
+		pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len())
 		nb.Write(pb.Bytes())
 	}
 
@@ -1038,7 +1039,7 @@ func (pk *Packet) UnsubscribeEncode(buf *bytes.Buffer) error {
 
 	if pk.ProtocolVersion == 5 {
 		pb := bytes.NewBuffer([]byte{})
-		pk.Properties.Encode(pk, pb, nb.Len()+xb.Len())
+		pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len()+xb.Len())
 		nb.Write(pb.Bytes())
 	}
 
@@ -1101,7 +1102,7 @@ func (pk *Packet) AuthEncode(buf *bytes.Buffer) error {
 	nb.WriteByte(pk.ReasonCode)
 
 	pb := bytes.NewBuffer([]byte{})
-	pk.Properties.Encode(pk, pb, nb.Len())
+	pk.Properties.Encode(pk.FixedHeader.Type, pk.Mods, pb, nb.Len())
 	nb.Write(pb.Bytes())
 
 	pk.FixedHeader.Remaining = nb.Len()

packets/properties.go

@@ -42,11 +42,11 @@ const (
 
 // validPacketProperties indicates which properties are valid for which packet types.
 var validPacketProperties = map[byte]map[byte]byte{
-	PropPayloadFormat:          {Publish: 1},
-	PropMessageExpiryInterval:  {Publish: 1},
-	PropContentType:            {Publish: 1},
-	PropResponseTopic:          {Publish: 1},
-	PropCorrelationData:        {Publish: 1},
+	PropPayloadFormat:          {Publish: 1, WillProperties: 1},
+	PropMessageExpiryInterval:  {Publish: 1, WillProperties: 1},
+	PropContentType:            {Publish: 1, WillProperties: 1},
+	PropResponseTopic:          {Publish: 1, WillProperties: 1},
+	PropCorrelationData:        {Publish: 1, WillProperties: 1},
 	PropSubscriptionIdentifier: {Publish: 1, Subscribe: 1},
 	PropSessionExpiryInterval:  {Connect: 1, Connack: 1, Disconnect: 1},
 	PropAssignedClientID:       {Connack: 1},
@@ -54,7 +54,7 @@ var validPacketProperties = map[byte]map[byte]byte{
 	PropAuthenticationMethod:   {Connect: 1, Connack: 1, Auth: 1},
 	PropAuthenticationData:     {Connect: 1, Connack: 1, Auth: 1},
 	PropRequestProblemInfo:     {Connect: 1},
-	PropWillDelayInterval:      {Connect: 1},
+	PropWillDelayInterval:      {WillProperties: 1},
 	PropRequestResponseInfo:    {Connect: 1},
 	PropResponseInfo:           {Connack: 1},
 	PropServerReference:        {Connack: 1, Disconnect: 1},
@@ -64,7 +64,7 @@ var validPacketProperties = map[byte]map[byte]byte{
 	PropTopicAlias:             {Publish: 1},
 	PropMaximumQos:             {Connack: 1},
 	PropRetainAvailable:        {Connack: 1},
-	PropUser:                   {Connect: 1, Connack: 1, Publish: 1, Puback: 1, Pubrec: 1, Pubrel: 1, Pubcomp: 1, Subscribe: 1, Suback: 1, Unsubscribe: 1, Unsuback: 1, Disconnect: 1, Auth: 1},
+	PropUser:                   {Connect: 1, Connack: 1, Publish: 1, Puback: 1, Pubrec: 1, Pubrel: 1, Pubcomp: 1, Subscribe: 1, Suback: 1, Unsubscribe: 1, Unsuback: 1, Disconnect: 1, Auth: 1, WillProperties: 1},
 	PropMaximumPacketSize:      {Connect: 1, Connack: 1},
 	PropWildcardSubAvailable:   {Connack: 1},
 	PropSubIDAvailable:         {Connack: 1},
@@ -194,14 +194,12 @@ func (p *Properties) canEncode(pkt byte, k byte) bool {
 }
 
 // Encode encodes properties into a bytes buffer.
-func (p *Properties) Encode(pk *Packet, b *bytes.Buffer, n int) {
+func (p *Properties) Encode(pkt byte, mods Mods, b *bytes.Buffer, n int) {
 	if p == nil {
 		return
 	}
 
 	var buf bytes.Buffer
-	pkt := pk.FixedHeader.Type
-
 	if p.canEncode(pkt, PropPayloadFormat) && p.PayloadFormatFlag {
 		buf.WriteByte(PropPayloadFormat)
 		buf.WriteByte(p.PayloadFormat)
@@ -217,13 +215,13 @@ func (p *Properties) Encode(pk *Packet, b *bytes.Buffer, n int) {
 		buf.Write(encodeString(p.ContentType)) // [MQTT-3.3.2-19]
 	}
 
-	if pk.Mods.AllowResponseInfo && p.canEncode(pkt, PropResponseTopic) && //  [MQTT-3.3.2-14]
+	if mods.AllowResponseInfo && p.canEncode(pkt, PropResponseTopic) && //  [MQTT-3.3.2-14]
 		p.ResponseTopic != "" && !strings.ContainsAny(p.ResponseTopic, "+#") { // [MQTT-3.1.2-28]
 		buf.WriteByte(PropResponseTopic)
 		buf.Write(encodeString(p.ResponseTopic)) // [MQTT-3.3.2-13]
 	}
 
-	if pk.Mods.AllowResponseInfo && p.canEncode(pkt, PropCorrelationData) && len(p.CorrelationData) > 0 { // [MQTT-3.1.2-28]
+	if mods.AllowResponseInfo && p.canEncode(pkt, PropCorrelationData) && len(p.CorrelationData) > 0 { // [MQTT-3.1.2-28]
 		buf.WriteByte(PropCorrelationData)
 		buf.Write(encodeBytes(p.CorrelationData))
 	}
@@ -277,7 +275,7 @@ func (p *Properties) Encode(pk *Packet, b *bytes.Buffer, n int) {
 		buf.WriteByte(p.RequestResponseInfo)
 	}
 
-	if pk.Mods.AllowResponseInfo && p.canEncode(pkt, PropResponseInfo) && len(p.ResponseInfo) > 0 { // [MQTT-3.1.2-28]
+	if mods.AllowResponseInfo && p.canEncode(pkt, PropResponseInfo) && len(p.ResponseInfo) > 0 { // [MQTT-3.1.2-28]
 		buf.WriteByte(PropResponseInfo)
 		buf.Write(encodeString(p.ResponseInfo))
 	}
@@ -289,9 +287,9 @@ func (p *Properties) Encode(pk *Packet, b *bytes.Buffer, n int) {
 
 	// [MQTT-3.2.2-19] [MQTT-3.14.2-3] [MQTT-3.4.2-2] [MQTT-3.5.2-2]
 	// [MQTT-3.6.2-2] [MQTT-3.9.2-1] [MQTT-3.11.2-1] [MQTT-3.15.2-2]
-	if !pk.Mods.DisallowProblemInfo && p.canEncode(pkt, PropReasonString) && p.ReasonString != "" {
+	if !mods.DisallowProblemInfo && p.canEncode(pkt, PropReasonString) && p.ReasonString != "" {
 		b := encodeString(p.ReasonString)
-		if pk.Mods.MaxSize == 0 || uint32(n+len(b)+1) < pk.Mods.MaxSize {
+		if mods.MaxSize == 0 || uint32(n+len(b)+1) < mods.MaxSize {
 			buf.WriteByte(PropReasonString)
 			buf.Write(b)
 		}
@@ -322,7 +320,7 @@ func (p *Properties) Encode(pk *Packet, b *bytes.Buffer, n int) {
 		buf.WriteByte(p.RetainAvailable)
 	}
 
-	if !pk.Mods.DisallowProblemInfo && p.canEncode(pkt, PropUser) {
+	if !mods.DisallowProblemInfo && p.canEncode(pkt, PropUser) {
 		pb := bytes.NewBuffer([]byte{})
 		for _, v := range p.User {
 			pb.WriteByte(PropUser)
@@ -331,7 +329,7 @@ func (p *Properties) Encode(pk *Packet, b *bytes.Buffer, n int) {
 		}
 		// [MQTT-3.2.2-20] [MQTT-3.14.2-4] [MQTT-3.4.2-3] [MQTT-3.5.2-3]
 		// [MQTT-3.6.2-3] [MQTT-3.9.2-2] [MQTT-3.11.2-2] [MQTT-3.15.2-3]
-		if pk.Mods.MaxSize == 0 || uint32(n+pb.Len()+1) < pk.Mods.MaxSize {
+		if mods.MaxSize == 0 || uint32(n+pb.Len()+1) < mods.MaxSize {
 			buf.Write(pb.Bytes())
 		}
 	}
@@ -361,7 +359,7 @@ func (p *Properties) Encode(pk *Packet, b *bytes.Buffer, n int) {
 }
 
 // Decode decodes property bytes into a properties struct.
-func (p *Properties) Decode(pk byte, b *bytes.Buffer) (n int, err error) {
+func (p *Properties) Decode(pkt byte, b *bytes.Buffer) (n int, err error) {
 	if p == nil {
 		return 0, nil
 	}
@@ -384,8 +382,8 @@ func (p *Properties) Decode(pk byte, b *bytes.Buffer) (n int, err error) {
 			return n + bu, err
 		}
 
-		if _, ok := validPacketProperties[k][pk]; !ok {
-			return n + bu, fmt.Errorf("property type %v not valid for packet type %v: %w", k, pk, ErrProtocolViolationUnsupportedProperty)
+		if _, ok := validPacketProperties[k][pkt]; !ok {
+			return n + bu, fmt.Errorf("property type %v not valid for packet type %v: %w", k, pkt, ErrProtocolViolationUnsupportedProperty)
 		}
 
 		switch k {

packets/properties_test.go

@@ -202,14 +202,14 @@ func init() {
 func TestEncodeProperties(t *testing.T) {
 	props := propertiesStruct
 	b := bytes.NewBuffer([]byte{})
-	props.Encode(&Packet{FixedHeader: FixedHeader{Type: Reserved}, Mods: Mods{AllowResponseInfo: true}}, b, 0)
+	props.Encode(Reserved, Mods{AllowResponseInfo: true}, b, 0)
 	require.Equal(t, propertiesBytes, b.Bytes())
 }
 
 func TestEncodePropertiesDisallowProblemInfo(t *testing.T) {
 	props := propertiesStruct
 	b := bytes.NewBuffer([]byte{})
-	props.Encode(&Packet{FixedHeader: FixedHeader{Type: Reserved}, Mods: Mods{DisallowProblemInfo: true}}, b, 0)
+	props.Encode(Reserved, Mods{DisallowProblemInfo: true}, b, 0)
 	require.NotEqual(t, propertiesBytes, b.Bytes())
 	require.False(t, bytes.Contains(b.Bytes(), []byte{31, 0, 6}))
 	require.False(t, bytes.Contains(b.Bytes(), []byte{38, 0, 5}))
@@ -219,7 +219,7 @@ func TestEncodePropertiesDisallowProblemInfo(t *testing.T) {
 func TestEncodePropertiesDisallowResponseInfo(t *testing.T) {
 	props := propertiesStruct
 	b := bytes.NewBuffer([]byte{})
-	props.Encode(&Packet{FixedHeader: FixedHeader{Type: Reserved}, Mods: Mods{AllowResponseInfo: false}}, b, 0)
+	props.Encode(Reserved, Mods{AllowResponseInfo: false}, b, 0)
 	require.NotEqual(t, propertiesBytes, b.Bytes())
 	require.NotContains(t, b.Bytes(), []byte{8, 0, 5})
 	require.NotContains(t, b.Bytes(), []byte{9, 0, 4})
@@ -232,7 +232,7 @@ func TestEncodePropertiesNil(t *testing.T) {
 
 	pr := tmp{}
 	b := bytes.NewBuffer([]byte{})
-	pr.p.Encode(&Packet{FixedHeader: FixedHeader{Type: Reserved}}, b, 0)
+	pr.p.Encode(Reserved, Mods{}, b, 0)
 	require.Equal(t, []byte{}, b.Bytes())
 }
 
@@ -240,7 +240,7 @@ func TestEncodeZeroProperties(t *testing.T) {
 	// [MQTT-2.2.2-1] If there are no properties, this MUST be indicated by including a Property Length of zero.
 	props := new(Properties)
 	b := bytes.NewBuffer([]byte{})
-	props.Encode(&Packet{FixedHeader: FixedHeader{Type: Reserved}, Mods: Mods{AllowResponseInfo: true}}, b, 0)
+	props.Encode(Reserved, Mods{AllowResponseInfo: true}, b, 0)
 	require.Equal(t, []byte{0x00}, b.Bytes())
 }
 
@@ -250,7 +250,7 @@ func TestDecodeProperties(t *testing.T) {
 	props := new(Properties)
 	n, err := props.Decode(Reserved, b)
 	require.NoError(t, err)
-	require.Equal(t, 172 + 2, n)
+	require.Equal(t, 172+2, n)
 	require.EqualValues(t, propertiesStruct, *props)
 }
 

server.go

@@ -26,7 +26,7 @@ import (
 )
 
 const (
-	Version                       = "2.2.7" // the current server version.
+	Version                       = "2.2.8" // the current server version.
 	defaultSysTopicInterval int64 = 1       // the interval between $SYS topic publishes
 )
 
@@ -372,7 +372,7 @@ func (s *Server) attachClient(cl *Client, listener string) error {
 	}
 
 	s.Log.Debug().Str("client", cl.ID).Err(err).Str("remote", cl.Net.Remote).Str("listener", listener).Msg("client disconnected")
-	expire := (cl.Properties.ProtocolVersion == 5 && cl.Properties.Props.SessionExpiryIntervalFlag && cl.Properties.Props.SessionExpiryInterval == 0) || (cl.Properties.ProtocolVersion < 5 && cl.Properties.Clean)
+	expire := (cl.Properties.ProtocolVersion == 5 && cl.Properties.Props.SessionExpiryInterval == 0) || (cl.Properties.ProtocolVersion < 5 && cl.Properties.Clean)
 	s.hooks.OnDisconnect(cl, err, expire)
 
 	if expire && atomic.LoadUint32(&cl.State.isTakenOver) == 0 {
@@ -826,6 +826,7 @@ func (s *Server) publishToClient(cl *Client, sub packets.Subscription, pk packet
 	if out.FixedHeader.Qos > 0 {
 		i, err := cl.NextPacketID() // [MQTT-4.3.2-1] [MQTT-4.3.3-1]
 		if err != nil {
+			s.hooks.OnPacketIDExhausted(cl, pk)
 			s.Log.Warn().Err(err).Str("client", cl.ID).Str("listener", cl.Net.Listener).Msg("packet ids exhausted")
 			return out, packets.ErrQuotaExceeded
 		}
@@ -846,7 +847,7 @@ func (s *Server) publishToClient(cl *Client, sub packets.Subscription, pk packet
 		}
 	}
 
-	if cl.Net.Conn == nil || atomic.LoadUint32(&cl.State.done) == 1 {
+	if cl.Net.Conn == nil || cl.Closed() {
 		return out, packets.CodeDisconnect
 	}
 

server_test.go

@@ -2476,7 +2476,7 @@ func TestServerProcessPacketDisconnect(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Equal(t, 0, s.loop.willDelayed.Len())
-	require.Equal(t, uint32(1), atomic.LoadUint32(&cl.State.done))
+	require.True(t, cl.Closed())
 	require.Equal(t, time.Now().Unix(), atomic.LoadInt64(&cl.State.disconnected))
 }
 
@@ -2806,7 +2806,7 @@ func TestServerClearExpiredClients(t *testing.T) {
 	cl0, _, _ := newTestClient()
 	cl0.ID = "c0"
 	cl0.State.disconnected = n - 10
-	cl0.State.done = 1
+	cl0.State.open = nil
 	cl0.Properties.ProtocolVersion = 5
 	cl0.Properties.Props.SessionExpiryInterval = 12
 	cl0.Properties.Props.SessionExpiryIntervalFlag = true
@@ -2816,7 +2816,7 @@ func TestServerClearExpiredClients(t *testing.T) {
 	cl1, _, _ := newTestClient()
 	cl1.ID = "c1"
 	cl1.State.disconnected = n - 10
-	cl1.State.done = 1
+	cl1.State.open = nil
 	cl1.Properties.ProtocolVersion = 5
 	cl1.Properties.Props.SessionExpiryInterval = 8
 	cl1.Properties.Props.SessionExpiryIntervalFlag = true
@@ -2826,7 +2826,7 @@ func TestServerClearExpiredClients(t *testing.T) {
 	cl2, _, _ := newTestClient()
 	cl2.ID = "c2"
 	cl2.State.disconnected = n - 10
-	cl2.State.done = 1
+	cl2.State.open = nil
 	cl2.Properties.ProtocolVersion = 5
 	cl2.Properties.Props.SessionExpiryInterval = 0
 	cl2.Properties.Props.SessionExpiryIntervalFlag = true