apps/mochi-mqtt
1
0
Fork 0
mirror of https://github.com/mochi-mqtt/server.git synced 2025-11-03 02:23:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

Ensure msg doesn't exceed subscription QoS (#253)

Browse Source 
Co-authored-by: JB <28275108+mochi-co@users.noreply.github.com>
This commit is contained in:
thedevop
2023-07-13 13:19:22 -04:00
committed by GitHub
parent 6f42c3fd65
commit c77d1c0331
2 changed files with 34 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
server.go
View File

@@ -818,6 +818,10 @@ func (s *Server) publishToClient(cl *Client, sub packets.Subscription, pk packet
sort.Ints(out.Properties.SubscriptionIdentifier)
}
if out.FixedHeader.Qos > sub.Qos {
out.FixedHeader.Qos = sub.Qos
}
if out.FixedHeader.Qos > s.Options.Capabilities.MaximumQos {
out.FixedHeader.Qos = s.Options.Capabilities.MaximumQos // [MQTT-3.2.2-9]
}

31
server_test.go
View File

@@ -1566,6 +1566,35 @@ func TestPublishToClientServerDowngradeQos(t *testing.T) {
require.Equal(t, packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).RawBytes, <-receiverBuf)
}
func TestPublishToClientSubscriptionDowngradeQos(t *testing.T) {
s := newServer()
s.Options.Capabilities.MaximumQos = 2
cl, r, w := newTestClient()
s.Clients.Add(cl)
_, ok := cl.State.Inflight.Get(1)
require.False(t, ok)
cl.State.packetID = 6 // just to match the same packet id (7) in the fixtures
go func() {
pkx := *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet
pkx.FixedHeader.Qos = 2
s.publishToClient(cl, packets.Subscription{Filter: "a/b/c", Qos: 1}, pkx)
time.Sleep(time.Microsecond * 100)
w.Close()
}()
receiverBuf := make(chan []byte)
go func() {
buf, err := io.ReadAll(r)
require.NoError(t, err)
receiverBuf <- buf
}()
require.Equal(t, packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).RawBytes, <-receiverBuf)
}
func TestPublishToClientExceedClientWritesPending(t *testing.T) {
s := newServer()
@@ -1630,7 +1659,7 @@ func TestPublishToClientExhaustedPacketID(t *testing.T) {
cl.State.Inflight.Set(packets.Packet{PacketID: uint16(i)})
}
_, err := s.publishToClient(cl, packets.Subscription{Filter: "a/b/c"}, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet)
_, err := s.publishToClient(cl, packets.Subscription{Filter: "a/b/c", Qos: 1}, *packets.TPacketData[packets.Publish].Get(packets.TPublishQos1).Packet)
require.Error(t, err)
require.ErrorIs(t, err, packets.ErrQuotaExceeded)
}