Sergey Krashevich
2b7682cdb3
- replace traditional for loop with range-based for loop for clarity
refactor(ffmpeg): simplify cut function loop
- utilize range-based for loop instead of traditional for loop
refactor(ring): update API response mapping type
- change map type from `interface{}` to `any` for better type safety
refactor(stream): handle nil source in NewStream
- add nil check for source elements before processing
refactor(webrtc): unify payload handling in GetToken
- change map type from `interface{}` to `any` for consistency
refactor(ascii): optimize nested loops in Write function
- replace traditional for loops with range-based for loops for readability
refactor(bits): enhance readability in Reader methods
- replace traditional for loops with range-based for loops in Read functions
refactor(h264): modernize loop structures in DecodeConfig
- switch to range-based for loops for cleaner code
refactor(h265): streamline profile_tier_level loops
- utilize range-based for loops instead of traditional for loops
chore(core): remove commented-out test function for clarity
refactor(core): simplify RandString function loop
- change traditional for loop to range-based for loop
refactor(flvt): optimize timestamp handling in TestTimeToRTP
- switch to range-based for loop for iterating frames
refactor(nest): improve error handling in ExchangeSDP
- format error message with printf-style formatting for clarity
refactor(tapo): enhance securityEncode function
- change traditional for loop to range-based for loop for readability
fix(tcp): correct masking in websocket Write method
- replace traditional for loop with range-based for loop
refactor(tutk): modernize encoding loops in crypto functions
- utilize range-based for loops for better readability
refactor(tuya): unify data types in MQTT message struct
- change map type from `interface{}` to `any` for consistency
refactor(webrtc): standardize codec registration
- change map type from `interface{}` to `any` for type safety
refactor(yaml): simplify Unmarshal function signature
- update parameter type from `interface{}` to `any` for better clarity
123 lines
2.1 KiB
Go
123 lines
2.1 KiB
Go
package mqtt
|
|
|
|
import (
|
|
"io"
|
|
)
|
|
|
|
type Message struct {
|
|
b []byte
|
|
}
|
|
|
|
// https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html
|
|
const (
|
|
CONNECT = 0x10
|
|
CONNACK = 0x20
|
|
PUBLISH = 0x30
|
|
PUBACK = 0x40
|
|
SUBSCRIBE = 0x82
|
|
SUBACK = 0x90
|
|
QOS1 = 0x02
|
|
)
|
|
|
|
func (m *Message) WriteUint8(b byte) {
|
|
m.b = append(m.b, b)
|
|
}
|
|
|
|
func (m *Message) WriteBytes(b []byte) {
|
|
m.b = append(m.b, b...)
|
|
}
|
|
|
|
func (m *Message) WriteUint16(i uint16) {
|
|
m.b = append(m.b, byte(i>>8), byte(i))
|
|
}
|
|
|
|
func (m *Message) WriteLen(i int) {
|
|
for i > 0 {
|
|
b := byte(i % 128)
|
|
if i /= 128; i > 0 {
|
|
b |= 0x80
|
|
}
|
|
m.WriteUint8(b)
|
|
}
|
|
}
|
|
|
|
func (m *Message) WriteString(s string) {
|
|
m.WriteUint16(uint16(len(s)))
|
|
m.b = append(m.b, s...)
|
|
}
|
|
|
|
func (m *Message) Bytes() []byte {
|
|
return m.b
|
|
}
|
|
|
|
const (
|
|
flagCleanStart = 0x02
|
|
flagUsername = 0x80
|
|
flagPassword = 0x40
|
|
)
|
|
|
|
func NewConnect(clientID, username, password string) *Message {
|
|
m := &Message{}
|
|
m.WriteUint8(CONNECT)
|
|
m.WriteLen(16 + len(clientID) + len(username) + len(password))
|
|
|
|
m.WriteString("MQTT")
|
|
m.WriteUint8(4) // MQTT version
|
|
m.WriteUint8(flagCleanStart | flagUsername | flagPassword)
|
|
m.WriteUint16(30) // keepalive
|
|
|
|
m.WriteString(clientID)
|
|
m.WriteString(username)
|
|
m.WriteString(password)
|
|
return m
|
|
}
|
|
|
|
func NewSubscribe(mid uint16, topic string, qos byte) *Message {
|
|
m := &Message{}
|
|
m.WriteUint8(SUBSCRIBE)
|
|
m.WriteLen(5 + len(topic))
|
|
|
|
m.WriteUint16(mid)
|
|
m.WriteString(topic)
|
|
m.WriteUint8(qos)
|
|
return m
|
|
}
|
|
|
|
func NewPublish(topic string, payload []byte) *Message {
|
|
m := &Message{}
|
|
m.WriteUint8(PUBLISH)
|
|
m.WriteLen(2 + len(topic) + len(payload))
|
|
|
|
m.WriteString(topic)
|
|
m.WriteBytes(payload)
|
|
return m
|
|
}
|
|
|
|
func NewPublishQOS1(mid uint16, topic string, payload []byte) *Message {
|
|
m := &Message{}
|
|
m.WriteUint8(PUBLISH | QOS1)
|
|
m.WriteLen(4 + len(topic) + len(payload))
|
|
|
|
m.WriteString(topic)
|
|
m.WriteUint16(mid)
|
|
m.WriteBytes(payload)
|
|
return m
|
|
}
|
|
|
|
func ReadLen(r io.Reader) (uint32, error) {
|
|
var i uint32
|
|
var shift byte
|
|
|
|
b := []byte{0x80}
|
|
for b[0]&0x80 != 0 {
|
|
if _, err := r.Read(b); err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
i += uint32(b[0]&0x7F) << shift
|
|
shift += 7
|
|
}
|
|
|
|
return i, nil
|
|
}
|