98 lines
1.8 KiB
Go
98 lines
1.8 KiB
Go
package packets
|
|
|
|
import (
|
|
"bufio"
|
|
"bytes"
|
|
"errors"
|
|
"io"
|
|
|
|
"badat.dev/maeqtt/v2/mqtt/properties"
|
|
"badat.dev/maeqtt/v2/mqtt/types"
|
|
)
|
|
|
|
type PublishPacket struct {
|
|
Dup bool
|
|
Retain bool
|
|
QOSLevel byte
|
|
TopicName string
|
|
Payload []byte
|
|
PacketId *uint16
|
|
Properties properties.PublishPacketProperties
|
|
}
|
|
|
|
func (p PublishPacket) Visit(v PacketVisitor) {
|
|
v.VisitPublish(p)
|
|
}
|
|
|
|
func parsePublishPacket(control controlPacket) (PublishPacket, error) {
|
|
var err error
|
|
r := bufio.NewReader(control.reader)
|
|
packet := PublishPacket{}
|
|
|
|
if control.packetType != PacketTypePublish {
|
|
return packet, errors.New("Wrong packet type for parseConnectPacket")
|
|
}
|
|
|
|
packet.Retain = control.flags&1 == 1
|
|
packet.QOSLevel = byte((control.flags >> 1) & 0b11)
|
|
packet.Dup = (control.flags>>3)&1 == 0
|
|
|
|
packet.TopicName, err = types.DecodeUTF8String(r)
|
|
if err != nil {
|
|
return packet, err
|
|
}
|
|
|
|
if packet.QOSLevel != 0 {
|
|
packId, err := types.DecodeUint16(r)
|
|
if err != nil {
|
|
return packet, err
|
|
}
|
|
packet.PacketId = &packId
|
|
}
|
|
err = properties.ParseProperties(r, packet.Properties.ArrayOf())
|
|
if err != nil {
|
|
return packet, err
|
|
}
|
|
|
|
packet.Payload, err = io.ReadAll(r)
|
|
if err != nil {
|
|
return packet, err
|
|
}
|
|
|
|
return packet, nil
|
|
}
|
|
|
|
func (p PublishPacket) Write(w io.Writer) error {
|
|
buf := bytes.NewBuffer([]byte{})
|
|
|
|
err := types.WriteUTF8String(buf, p.TopicName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if p.PacketId != nil {
|
|
err := types.WriteUint16(buf, *p.PacketId)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
|
|
err = properties.WriteProps(buf, p.Properties.ArrayOf())
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
buf.Write(p.Payload)
|
|
|
|
flags := types.BoolToUint(p.Retain)
|
|
flags += uint(p.QOSLevel) << 1
|
|
flags += types.BoolToUint(p.Dup) << 3
|
|
conPack := controlPacket{
|
|
packetType: PacketTypePublish,
|
|
flags: flags,
|
|
reader: buf,
|
|
}
|
|
|
|
return conPack.write(w)
|
|
}
|