collab/hmq
3
0
Fork 0
mirror of https://github.com/fhmq/hmq.git synced 2026-04-24 10:38:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

packet

Browse Source
This commit is contained in:
zhouyuyan
2017-09-11 17:00:02 +08:00
parent cc1b3239ad
commit 5601632a33
22 changed files with 28 additions and 1251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@@ -15,6 +15,7 @@ $ go run main.go
### broker.config
~~~
{
"workerNum": 4096,
"port": "1883",
"host": "0.0.0.0",
"cluster": {

27
broker/broker.go
View File

@@ -3,13 +3,15 @@ package broker
import (
"crypto/tls"
"hmq/lib/acl"
"hmq/packets"
"net"
"net/http"
"runtime/debug"
"sync"
"sync/atomic"
"time"
"github.com/eclipse/paho.mqtt.golang/packets"
"golang.org/x/net/websocket"
log "github.com/cihub/seelog"
@@ -77,20 +79,21 @@ func (b *Broker) Start() {
if len(b.config.Cluster.Routes) > 0 {
b.ConnectToRouters()
}
// go StateMonitor()
go StateMonitor()
}
// func StateMonitor() {
// v, _ := mem.VirtualMemory()
// timeSticker := time.NewTicker(time.Second * 5)
// for {
// select {
// case <-timeSticker.C:
// fmt.Printf("Total: %v, Free:%v, UsedPercent:%f%%\n", v.Total, v.Free, v.UsedPercent)
// }
// }
// }
func StateMonitor() {
// v, _ := mem.VirtualMemory()
timeSticker := time.NewTicker(time.Second * 30)
for {
select {
case <-timeSticker.C:
debug.FreeOSMemory()
// fmt.Printf("Total: %v, Free:%v, UsedPercent:%f%%\n", v.Total, v.Free, v.UsedPercent)
}
}
}
func (b *Broker) StartWebsocketListening() {
path := b.config.WsPath

7
broker/client.go
View File

@@ -1,12 +1,13 @@
package broker
import (
"hmq/packets"
"net"
"strings"
"sync"
"time"
"github.com/eclipse/paho.mqtt.golang/packets"
log "github.com/cihub/seelog"
)
@@ -20,6 +21,10 @@ const (
//REMOTE is the router connect to other cluster
REMOTE = 2
)
const (
Connected = 1
Disconnected = 2
)
type client struct {
typ int

3
broker/info.go
View File

@@ -2,9 +2,10 @@ package broker
import (
"fmt"
"hmq/packets"
"time"
"github.com/eclipse/paho.mqtt.golang/packets"
simplejson "github.com/bitly/go-simplejson"
log "github.com/cihub/seelog"
)

3
broker/msgpool.go
View File

@@ -1,8 +1,9 @@
package broker
import (
"hmq/packets"
"sync"
"github.com/eclipse/paho.mqtt.golang/packets"
)
const (

3
broker/retain.go
View File

@@ -1,8 +1,9 @@
package broker
import (
"hmq/packets"
"sync"
"github.com/eclipse/paho.mqtt.golang/packets"
)
type RetainList struct {

51
packets/connack.go
View File

@@ -1,51 +0,0 @@
package packets
import (
"bytes"
"fmt"
"io"
)
//ConnackPacket is an internal representation of the fields of the
//Connack MQTT packet
type ConnackPacket struct {
FixedHeader
SessionPresent bool
ReturnCode byte
}
func (ca *ConnackPacket) String() string {
str := fmt.Sprintf("%s", ca.FixedHeader)
str += " "
str += fmt.Sprintf("sessionpresent: %t returncode: %d", ca.SessionPresent, ca.ReturnCode)
return str
}
func (ca *ConnackPacket) Write(w io.Writer) error {
var body bytes.Buffer
var err error
body.WriteByte(boolToByte(ca.SessionPresent))
body.WriteByte(ca.ReturnCode)
ca.FixedHeader.RemainingLength = 2
packet := ca.FixedHeader.pack()
packet.Write(body.Bytes())
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (ca *ConnackPacket) Unpack(b io.Reader) error {
ca.SessionPresent = 1&decodeByte(b) > 0
ca.ReturnCode = decodeByte(b)
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (ca *ConnackPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

126
packets/connect.go
View File

@@ -1,126 +0,0 @@
package packets
import (
"bytes"
"fmt"
"io"
)
//ConnectPacket is an internal representation of the fields of the
//Connect MQTT packet
type ConnectPacket struct {
FixedHeader
ProtocolName string
ProtocolVersion byte
CleanSession bool
WillFlag bool
WillQos byte
WillRetain bool
UsernameFlag bool
PasswordFlag bool
ReservedBit byte
Keepalive uint16
ClientIdentifier string
WillTopic string
WillMessage []byte
Username string
Password []byte
}
func (c *ConnectPacket) String() string {
str := fmt.Sprintf("%s", c.FixedHeader)
str += " "
str += fmt.Sprintf("protocolversion: %d protocolname: %s cleansession: %t willflag: %t WillQos: %d WillRetain: %t Usernameflag: %t Passwordflag: %t keepalive: %d clientId: %s willtopic: %s willmessage: %s Username: %s Password: %s", c.ProtocolVersion, c.ProtocolName, c.CleanSession, c.WillFlag, c.WillQos, c.WillRetain, c.UsernameFlag, c.PasswordFlag, c.Keepalive, c.ClientIdentifier, c.WillTopic, c.WillMessage, c.Username, c.Password)
return str
}
func (c *ConnectPacket) Write(w io.Writer) error {
var body bytes.Buffer
var err error
body.Write(encodeString(c.ProtocolName))
body.WriteByte(c.ProtocolVersion)
body.WriteByte(boolToByte(c.CleanSession)<<1 | boolToByte(c.WillFlag)<<2 | c.WillQos<<3 | boolToByte(c.WillRetain)<<5 | boolToByte(c.PasswordFlag)<<6 | boolToByte(c.UsernameFlag)<<7)
body.Write(encodeUint16(c.Keepalive))
body.Write(encodeString(c.ClientIdentifier))
if c.WillFlag {
body.Write(encodeString(c.WillTopic))
body.Write(encodeBytes(c.WillMessage))
}
if c.UsernameFlag {
body.Write(encodeString(c.Username))
}
if c.PasswordFlag {
body.Write(encodeBytes(c.Password))
}
c.FixedHeader.RemainingLength = body.Len()
packet := c.FixedHeader.pack()
packet.Write(body.Bytes())
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (c *ConnectPacket) Unpack(b io.Reader) error {
c.ProtocolName = decodeString(b)
c.ProtocolVersion = decodeByte(b)
options := decodeByte(b)
c.ReservedBit = 1 & options
c.CleanSession = 1&(options>>1) > 0
c.WillFlag = 1&(options>>2) > 0
c.WillQos = 3 & (options >> 3)
c.WillRetain = 1&(options>>5) > 0
c.PasswordFlag = 1&(options>>6) > 0
c.UsernameFlag = 1&(options>>7) > 0
c.Keepalive = decodeUint16(b)
c.ClientIdentifier = decodeString(b)
if c.WillFlag {
c.WillTopic = decodeString(b)
c.WillMessage = decodeBytes(b)
}
if c.UsernameFlag {
c.Username = decodeString(b)
}
if c.PasswordFlag {
c.Password = decodeBytes(b)
}
return nil
}
//Validate performs validation of the fields of a Connect packet
func (c *ConnectPacket) Validate() byte {
if c.PasswordFlag && !c.UsernameFlag {
return ErrRefusedBadUsernameOrPassword
}
if c.ReservedBit != 0 {
//Bad reserved bit
return ErrProtocolViolation
}
if (c.ProtocolName == "MQIsdp" && c.ProtocolVersion != 3) || (c.ProtocolName == "MQTT" && c.ProtocolVersion != 4) {
//Mismatched or unsupported protocol version
return ErrRefusedBadProtocolVersion
}
if c.ProtocolName != "MQIsdp" && c.ProtocolName != "MQTT" {
//Bad protocol name
return ErrProtocolViolation
}
if len(c.ClientIdentifier) > 65535 || len(c.Username) > 65535 || len(c.Password) > 65535 {
//Bad size field
return ErrProtocolViolation
}
if len(c.ClientIdentifier) == 0 && !c.CleanSession {
//Bad client identifier
return ErrRefusedIDRejected
}
return Accepted
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (c *ConnectPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

36
packets/disconnect.go
View File

@@ -1,36 +0,0 @@
package packets
import (
"fmt"
"io"
)
//DisconnectPacket is an internal representation of the fields of the
//Disconnect MQTT packet
type DisconnectPacket struct {
FixedHeader
}
func (d *DisconnectPacket) String() string {
str := fmt.Sprintf("%s", d.FixedHeader)
return str
}
func (d *DisconnectPacket) Write(w io.Writer) error {
packet := d.FixedHeader.pack()
_, err := packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (d *DisconnectPacket) Unpack(b io.Reader) error {
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (d *DisconnectPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

322
packets/packets.go
View File

@@ -1,322 +0,0 @@
package packets
import (
"bytes"
"encoding/binary"
"errors"
"fmt"
"io"
)
//ControlPacket defines the interface for structs intended to hold
//decoded MQTT packets, either from being read or before being
//written
type ControlPacket interface {
Write(io.Writer) error
Unpack(io.Reader) error
String() string
Details() Details
}
//PacketNames maps the constants for each of the MQTT packet types
//to a string representation of their name.
var PacketNames = map[uint8]string{
1: "CONNECT",
2: "CONNACK",
3: "PUBLISH",
4: "PUBACK",
5: "PUBREC",
6: "PUBREL",
7: "PUBCOMP",
8: "SUBSCRIBE",
9: "SUBACK",
10: "UNSUBSCRIBE",
11: "UNSUBACK",
12: "PINGREQ",
13: "PINGRESP",
14: "DISCONNECT",
}
//Below are the constants assigned to each of the MQTT packet types
const (
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
)
//Below are the const definitions for error codes returned by
//Connect()
const (
Accepted = 0x00
ErrRefusedBadProtocolVersion = 0x01
ErrRefusedIDRejected = 0x02
ErrRefusedServerUnavailable = 0x03
ErrRefusedBadUsernameOrPassword = 0x04
ErrRefusedNotAuthorised = 0x05
ErrNetworkError = 0xFE
ErrProtocolViolation = 0xFF
)
//ConnackReturnCodes is a map of the error codes constants for Connect()
//to a string representation of the error
var ConnackReturnCodes = map[uint8]string{
0: "Connection Accepted",
1: "Connection Refused: Bad Protocol Version",
2: "Connection Refused: Client Identifier Rejected",
3: "Connection Refused: Server Unavailable",
4: "Connection Refused: Username or Password in unknown format",
5: "Connection Refused: Not Authorised",
254: "Connection Error",
255: "Connection Refused: Protocol Violation",
}
//ConnErrors is a map of the errors codes constants for Connect()
//to a Go error
var ConnErrors = map[byte]error{
Accepted: nil,
ErrRefusedBadProtocolVersion: errors.New("Unnacceptable protocol version"),
ErrRefusedIDRejected: errors.New("Identifier rejected"),
ErrRefusedServerUnavailable: errors.New("Server Unavailable"),
ErrRefusedBadUsernameOrPassword: errors.New("Bad user name or password"),
ErrRefusedNotAuthorised: errors.New("Not Authorized"),
ErrNetworkError: errors.New("Network Error"),
ErrProtocolViolation: errors.New("Protocol Violation"),
}
//ReadPacket takes an instance of an io.Reader (such as net.Conn) and attempts
//to read an MQTT packet from the stream. It returns a ControlPacket
//representing the decoded MQTT packet and an error. One of these returns will
//always be nil, a nil ControlPacket indicating an error occurred.
func ReadPacket(r io.Reader) (cp ControlPacket, err error) {
var fh FixedHeader
b := make([]byte, 1)
_, err = io.ReadFull(r, b)
if err != nil {
return nil, err
}
fh.unpack(b[0], r)
cp = NewControlPacketWithHeader(fh)
if cp == nil {
return nil, errors.New("Bad data from client")
}
packetBytes := make([]byte, fh.RemainingLength)
_, err = io.ReadFull(r, packetBytes)
if err != nil {
return nil, err
}
err = cp.Unpack(bytes.NewBuffer(packetBytes))
return cp, err
}
//NewControlPacket is used to create a new ControlPacket of the type specified
//by packetType, this is usually done by reference to the packet type constants
//defined in packets.go. The newly created ControlPacket is empty and a pointer
//is returned.
func NewControlPacket(packetType byte) (cp ControlPacket) {
switch packetType {
case Connect:
cp = &ConnectPacket{FixedHeader: FixedHeader{MessageType: Connect}}
case Connack:
cp = &ConnackPacket{FixedHeader: FixedHeader{MessageType: Connack}}
case Disconnect:
cp = &DisconnectPacket{FixedHeader: FixedHeader{MessageType: Disconnect}}
case Publish:
cp = &PublishPacket{FixedHeader: FixedHeader{MessageType: Publish}}
case Puback:
cp = &PubackPacket{FixedHeader: FixedHeader{MessageType: Puback}}
case Pubrec:
cp = &PubrecPacket{FixedHeader: FixedHeader{MessageType: Pubrec}}
case Pubrel:
cp = &PubrelPacket{FixedHeader: FixedHeader{MessageType: Pubrel, Qos: 1}}
case Pubcomp:
cp = &PubcompPacket{FixedHeader: FixedHeader{MessageType: Pubcomp}}
case Subscribe:
cp = &SubscribePacket{FixedHeader: FixedHeader{MessageType: Subscribe, Qos: 1}}
case Suback:
cp = &SubackPacket{FixedHeader: FixedHeader{MessageType: Suback}}
case Unsubscribe:
cp = &UnsubscribePacket{FixedHeader: FixedHeader{MessageType: Unsubscribe, Qos: 1}}
case Unsuback:
cp = &UnsubackPacket{FixedHeader: FixedHeader{MessageType: Unsuback}}
case Pingreq:
cp = &PingreqPacket{FixedHeader: FixedHeader{MessageType: Pingreq}}
case Pingresp:
cp = &PingrespPacket{FixedHeader: FixedHeader{MessageType: Pingresp}}
default:
return nil
}
return cp
}
//NewControlPacketWithHeader is used to create a new ControlPacket of the type
//specified within the FixedHeader that is passed to the function.
//The newly created ControlPacket is empty and a pointer is returned.
func NewControlPacketWithHeader(fh FixedHeader) (cp ControlPacket) {
switch fh.MessageType {
case Connect:
cp = &ConnectPacket{FixedHeader: fh}
case Connack:
cp = &ConnackPacket{FixedHeader: fh}
case Disconnect:
cp = &DisconnectPacket{FixedHeader: fh}
case Publish:
cp = &PublishPacket{FixedHeader: fh}
case Puback:
cp = &PubackPacket{FixedHeader: fh}
case Pubrec:
cp = &PubrecPacket{FixedHeader: fh}
case Pubrel:
cp = &PubrelPacket{FixedHeader: fh}
case Pubcomp:
cp = &PubcompPacket{FixedHeader: fh}
case Subscribe:
cp = &SubscribePacket{FixedHeader: fh}
case Suback:
cp = &SubackPacket{FixedHeader: fh}
case Unsubscribe:
cp = &UnsubscribePacket{FixedHeader: fh}
case Unsuback:
cp = &UnsubackPacket{FixedHeader: fh}
case Pingreq:
cp = &PingreqPacket{FixedHeader: fh}
case Pingresp:
cp = &PingrespPacket{FixedHeader: fh}
default:
return nil
}
return cp
}
//Details struct returned by the Details() function called on
//ControlPackets to present details of the Qos and MessageID
//of the ControlPacket
type Details struct {
Qos byte
MessageID uint16
}
//FixedHeader is a struct to hold the decoded information from
//the fixed header of an MQTT ControlPacket
type FixedHeader struct {
MessageType byte
Dup bool
Qos byte
Retain bool
RemainingLength int
}
func (fh FixedHeader) String() string {
return fmt.Sprintf("%s: dup: %t qos: %d retain: %t rLength: %d", PacketNames[fh.MessageType], fh.Dup, fh.Qos, fh.Retain, fh.RemainingLength)
}
func boolToByte(b bool) byte {
switch b {
case true:
return 1
default:
return 0
}
}
func (fh *FixedHeader) pack() bytes.Buffer {
var header bytes.Buffer
header.WriteByte(fh.MessageType<<4 | boolToByte(fh.Dup)<<3 | fh.Qos<<1 | boolToByte(fh.Retain))
header.Write(encodeLength(fh.RemainingLength))
return header
}
func (fh *FixedHeader) unpack(typeAndFlags byte, r io.Reader) {
fh.MessageType = typeAndFlags >> 4
fh.Dup = (typeAndFlags>>3)&0x01 > 0
fh.Qos = (typeAndFlags >> 1) & 0x03
fh.Retain = typeAndFlags&0x01 > 0
fh.RemainingLength = decodeLength(r)
}
func decodeByte(b io.Reader) byte {
num := make([]byte, 1)
b.Read(num)
return num[0]
}
func decodeUint16(b io.Reader) uint16 {
num := make([]byte, 2)
b.Read(num)
return binary.BigEndian.Uint16(num)
}
func encodeUint16(num uint16) []byte {
bytes := make([]byte, 2)
binary.BigEndian.PutUint16(bytes, num)
return bytes
}
func encodeString(field string) []byte {
fieldLength := make([]byte, 2)
binary.BigEndian.PutUint16(fieldLength, uint16(len(field)))
return append(fieldLength, []byte(field)...)
}
func decodeString(b io.Reader) string {
fieldLength := decodeUint16(b)
field := make([]byte, fieldLength)
b.Read(field)
return string(field)
}
func decodeBytes(b io.Reader) []byte {
fieldLength := decodeUint16(b)
field := make([]byte, fieldLength)
b.Read(field)
return field
}
func encodeBytes(field []byte) []byte {
fieldLength := make([]byte, 2)
binary.BigEndian.PutUint16(fieldLength, uint16(len(field)))
return append(fieldLength, field...)
}
func encodeLength(length int) []byte {
var encLength []byte
for {
digit := byte(length % 128)
length /= 128
if length > 0 {
digit |= 0x80
}
encLength = append(encLength, digit)
if length == 0 {
break
}
}
return encLength
}
func decodeLength(r io.Reader) int {
var rLength uint32
var multiplier uint32
b := make([]byte, 1)
for multiplier < 27 { //fix: Infinite '(digit & 128) == 1' will cause the dead loop
io.ReadFull(r, b)
digit := b[0]
rLength |= uint32(digit&127) << multiplier
if (digit & 128) == 0 {
break
}
multiplier += 7
}
return int(rLength)
}

159
packets/packets_test.go
View File

@@ -1,159 +0,0 @@
package packets
import (
"bytes"
"testing"
)
func TestPacketNames(t *testing.T) {
if PacketNames[1] != "CONNECT" {
t.Errorf("PacketNames[1] is %s, should be %s", PacketNames[1], "CONNECT")
}
if PacketNames[2] != "CONNACK" {
t.Errorf("PacketNames[2] is %s, should be %s", PacketNames[2], "CONNACK")
}
if PacketNames[3] != "PUBLISH" {
t.Errorf("PacketNames[3] is %s, should be %s", PacketNames[3], "PUBLISH")
}
if PacketNames[4] != "PUBACK" {
t.Errorf("PacketNames[4] is %s, should be %s", PacketNames[4], "PUBACK")
}
if PacketNames[5] != "PUBREC" {
t.Errorf("PacketNames[5] is %s, should be %s", PacketNames[5], "PUBREC")
}
if PacketNames[6] != "PUBREL" {
t.Errorf("PacketNames[6] is %s, should be %s", PacketNames[6], "PUBREL")
}
if PacketNames[7] != "PUBCOMP" {
t.Errorf("PacketNames[7] is %s, should be %s", PacketNames[7], "PUBCOMP")
}
if PacketNames[8] != "SUBSCRIBE" {
t.Errorf("PacketNames[8] is %s, should be %s", PacketNames[8], "SUBSCRIBE")
}
if PacketNames[9] != "SUBACK" {
t.Errorf("PacketNames[9] is %s, should be %s", PacketNames[9], "SUBACK")
}
if PacketNames[10] != "UNSUBSCRIBE" {
t.Errorf("PacketNames[10] is %s, should be %s", PacketNames[10], "UNSUBSCRIBE")
}
if PacketNames[11] != "UNSUBACK" {
t.Errorf("PacketNames[11] is %s, should be %s", PacketNames[11], "UNSUBACK")
}
if PacketNames[12] != "PINGREQ" {
t.Errorf("PacketNames[12] is %s, should be %s", PacketNames[12], "PINGREQ")
}
if PacketNames[13] != "PINGRESP" {
t.Errorf("PacketNames[13] is %s, should be %s", PacketNames[13], "PINGRESP")
}
if PacketNames[14] != "DISCONNECT" {
t.Errorf("PacketNames[14] is %s, should be %s", PacketNames[14], "DISCONNECT")
}
}
func TestPacketConsts(t *testing.T) {
if Connect != 1 {
t.Errorf("Const for Connect is %d, should be %d", Connect, 1)
}
if Connack != 2 {
t.Errorf("Const for Connack is %d, should be %d", Connack, 2)
}
if Publish != 3 {
t.Errorf("Const for Publish is %d, should be %d", Publish, 3)
}
if Puback != 4 {
t.Errorf("Const for Puback is %d, should be %d", Puback, 4)
}
if Pubrec != 5 {
t.Errorf("Const for Pubrec is %d, should be %d", Pubrec, 5)
}
if Pubrel != 6 {
t.Errorf("Const for Pubrel is %d, should be %d", Pubrel, 6)
}
if Pubcomp != 7 {
t.Errorf("Const for Pubcomp is %d, should be %d", Pubcomp, 7)
}
if Subscribe != 8 {
t.Errorf("Const for Subscribe is %d, should be %d", Subscribe, 8)
}
if Suback != 9 {
t.Errorf("Const for Suback is %d, should be %d", Suback, 9)
}
if Unsubscribe != 10 {
t.Errorf("Const for Unsubscribe is %d, should be %d", Unsubscribe, 10)
}
if Unsuback != 11 {
t.Errorf("Const for Unsuback is %d, should be %d", Unsuback, 11)
}
if Pingreq != 12 {
t.Errorf("Const for Pingreq is %d, should be %d", Pingreq, 12)
}
if Pingresp != 13 {
t.Errorf("Const for Pingresp is %d, should be %d", Pingresp, 13)
}
if Disconnect != 14 {
t.Errorf("Const for Disconnect is %d, should be %d", Disconnect, 14)
}
}
func TestConnackConsts(t *testing.T) {
if Accepted != 0x00 {
t.Errorf("Const for Accepted is %d, should be %d", Accepted, 0)
}
if ErrRefusedBadProtocolVersion != 0x01 {
t.Errorf("Const for RefusedBadProtocolVersion is %d, should be %d", ErrRefusedBadProtocolVersion, 1)
}
if ErrRefusedIDRejected != 0x02 {
t.Errorf("Const for RefusedIDRejected is %d, should be %d", ErrRefusedIDRejected, 2)
}
if ErrRefusedServerUnavailable != 0x03 {
t.Errorf("Const for RefusedServerUnavailable is %d, should be %d", ErrRefusedServerUnavailable, 3)
}
if ErrRefusedBadUsernameOrPassword != 0x04 {
t.Errorf("Const for RefusedBadUsernameOrPassword is %d, should be %d", ErrRefusedBadUsernameOrPassword, 4)
}
if ErrRefusedNotAuthorised != 0x05 {
t.Errorf("Const for RefusedNotAuthorised is %d, should be %d", ErrRefusedNotAuthorised, 5)
}
}
func TestConnectPacket(t *testing.T) {
connectPacketBytes := bytes.NewBuffer([]byte{16, 52, 0, 4, 77, 81, 84, 84, 4, 204, 0, 0, 0, 0, 0, 4, 116, 101, 115, 116, 0, 12, 84, 101, 115, 116, 32, 80, 97, 121, 108, 111, 97, 100, 0, 8, 116, 101, 115, 116, 117, 115, 101, 114, 0, 8, 116, 101, 115, 116, 112, 97, 115, 115})
packet, err := ReadPacket(connectPacketBytes)
if err != nil {
t.Fatalf("Error reading packet: %s", err.Error())
}
cp := packet.(*ConnectPacket)
if cp.ProtocolName != "MQTT" {
t.Errorf("Connect Packet ProtocolName is %s, should be %s", cp.ProtocolName, "MQTT")
}
if cp.ProtocolVersion != 4 {
t.Errorf("Connect Packet ProtocolVersion is %d, should be %d", cp.ProtocolVersion, 4)
}
if cp.UsernameFlag != true {
t.Errorf("Connect Packet UsernameFlag is %t, should be %t", cp.UsernameFlag, true)
}
if cp.Username != "testuser" {
t.Errorf("Connect Packet Username is %s, should be %s", cp.Username, "testuser")
}
if cp.PasswordFlag != true {
t.Errorf("Connect Packet PasswordFlag is %t, should be %t", cp.PasswordFlag, true)
}
if string(cp.Password) != "testpass" {
t.Errorf("Connect Packet Password is %s, should be %s", string(cp.Password), "testpass")
}
if cp.WillFlag != true {
t.Errorf("Connect Packet WillFlag is %t, should be %t", cp.WillFlag, true)
}
if cp.WillTopic != "test" {
t.Errorf("Connect Packet WillTopic is %s, should be %s", cp.WillTopic, "test")
}
if cp.WillQos != 1 {
t.Errorf("Connect Packet WillQos is %d, should be %d", cp.WillQos, 1)
}
if cp.WillRetain != false {
t.Errorf("Connect Packet WillRetain is %t, should be %t", cp.WillRetain, false)
}
if string(cp.WillMessage) != "Test Payload" {
t.Errorf("Connect Packet WillMessage is %s, should be %s", string(cp.WillMessage), "Test Payload")
}
}

36
packets/pingreq.go
View File

@@ -1,36 +0,0 @@
package packets
import (
"fmt"
"io"
)
//PingreqPacket is an internal representation of the fields of the
//Pingreq MQTT packet
type PingreqPacket struct {
FixedHeader
}
func (pr *PingreqPacket) String() string {
str := fmt.Sprintf("%s", pr.FixedHeader)
return str
}
func (pr *PingreqPacket) Write(w io.Writer) error {
packet := pr.FixedHeader.pack()
_, err := packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (pr *PingreqPacket) Unpack(b io.Reader) error {
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (pr *PingreqPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

36
packets/pingresp.go
View File

@@ -1,36 +0,0 @@
package packets
import (
"fmt"
"io"
)
//PingrespPacket is an internal representation of the fields of the
//Pingresp MQTT packet
type PingrespPacket struct {
FixedHeader
}
func (pr *PingrespPacket) String() string {
str := fmt.Sprintf("%s", pr.FixedHeader)
return str
}
func (pr *PingrespPacket) Write(w io.Writer) error {
packet := pr.FixedHeader.pack()
_, err := packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (pr *PingrespPacket) Unpack(b io.Reader) error {
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (pr *PingrespPacket) Details() Details {
return Details{Qos: 0, MessageID: 0}
}

44
packets/puback.go
View File

@@ -1,44 +0,0 @@
package packets
import (
"fmt"
"io"
)
//PubackPacket is an internal representation of the fields of the
//Puback MQTT packet
type PubackPacket struct {
FixedHeader
MessageID uint16
}
func (pa *PubackPacket) String() string {
str := fmt.Sprintf("%s", pa.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", pa.MessageID)
return str
}
func (pa *PubackPacket) Write(w io.Writer) error {
var err error
pa.FixedHeader.RemainingLength = 2
packet := pa.FixedHeader.pack()
packet.Write(encodeUint16(pa.MessageID))
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (pa *PubackPacket) Unpack(b io.Reader) error {
pa.MessageID = decodeUint16(b)
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (pa *PubackPacket) Details() Details {
return Details{Qos: pa.Qos, MessageID: pa.MessageID}
}

44
packets/pubcomp.go
View File

@@ -1,44 +0,0 @@
package packets
import (
"fmt"
"io"
)
//PubcompPacket is an internal representation of the fields of the
//Pubcomp MQTT packet
type PubcompPacket struct {
FixedHeader
MessageID uint16
}
func (pc *PubcompPacket) String() string {
str := fmt.Sprintf("%s", pc.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", pc.MessageID)
return str
}
func (pc *PubcompPacket) Write(w io.Writer) error {
var err error
pc.FixedHeader.RemainingLength = 2
packet := pc.FixedHeader.pack()
packet.Write(encodeUint16(pc.MessageID))
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (pc *PubcompPacket) Unpack(b io.Reader) error {
pc.MessageID = decodeUint16(b)
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (pc *PubcompPacket) Details() Details {
return Details{Qos: pc.Qos, MessageID: pc.MessageID}
}

80
packets/publish.go
View File

@@ -1,80 +0,0 @@
package packets
import (
"bytes"
"fmt"
"io"
)
//PublishPacket is an internal representation of the fields of the
//Publish MQTT packet
type PublishPacket struct {
FixedHeader
TopicName string
MessageID uint16
Payload []byte
}
func (p *PublishPacket) String() string {
str := fmt.Sprintf("%s", p.FixedHeader)
str += " "
str += fmt.Sprintf("topicName: %s MessageID: %d", p.TopicName, p.MessageID)
str += " "
str += fmt.Sprintf("payload: %s", string(p.Payload))
return str
}
func (p *PublishPacket) Write(w io.Writer) error {
var body bytes.Buffer
var err error
body.Write(encodeString(p.TopicName))
if p.Qos > 0 {
body.Write(encodeUint16(p.MessageID))
}
p.FixedHeader.RemainingLength = body.Len() + len(p.Payload)
packet := p.FixedHeader.pack()
packet.Write(body.Bytes())
packet.Write(p.Payload)
_, err = w.Write(packet.Bytes())
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (p *PublishPacket) Unpack(b io.Reader) error {
var payloadLength = p.FixedHeader.RemainingLength
p.TopicName = decodeString(b)
if p.Qos > 0 {
p.MessageID = decodeUint16(b)
payloadLength -= len(p.TopicName) + 4
} else {
payloadLength -= len(p.TopicName) + 2
}
if payloadLength < 0 {
return fmt.Errorf("Error upacking publish, payload length < 0")
}
p.Payload = make([]byte, payloadLength)
_, err := b.Read(p.Payload)
return err
}
//Copy creates a new PublishPacket with the same topic and payload
//but an empty fixed header, useful for when you want to deliver
//a message with different properties such as Qos but the same
//content
func (p *PublishPacket) Copy() *PublishPacket {
newP := NewControlPacket(Publish).(*PublishPacket)
newP.TopicName = p.TopicName
newP.Payload = p.Payload
return newP
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (p *PublishPacket) Details() Details {
return Details{Qos: p.Qos, MessageID: p.MessageID}
}

44
packets/pubrec.go
View File

@@ -1,44 +0,0 @@
package packets
import (
"fmt"
"io"
)
//PubrecPacket is an internal representation of the fields of the
//Pubrec MQTT packet
type PubrecPacket struct {
FixedHeader
MessageID uint16
}
func (pr *PubrecPacket) String() string {
str := fmt.Sprintf("%s", pr.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", pr.MessageID)
return str
}
func (pr *PubrecPacket) Write(w io.Writer) error {
var err error
pr.FixedHeader.RemainingLength = 2
packet := pr.FixedHeader.pack()
packet.Write(encodeUint16(pr.MessageID))
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (pr *PubrecPacket) Unpack(b io.Reader) error {
pr.MessageID = decodeUint16(b)
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (pr *PubrecPacket) Details() Details {
return Details{Qos: pr.Qos, MessageID: pr.MessageID}
}

44
packets/pubrel.go
View File

@@ -1,44 +0,0 @@
package packets
import (
"fmt"
"io"
)
//PubrelPacket is an internal representation of the fields of the
//Pubrel MQTT packet
type PubrelPacket struct {
FixedHeader
MessageID uint16
}
func (pr *PubrelPacket) String() string {
str := fmt.Sprintf("%s", pr.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", pr.MessageID)
return str
}
func (pr *PubrelPacket) Write(w io.Writer) error {
var err error
pr.FixedHeader.RemainingLength = 2
packet := pr.FixedHeader.pack()
packet.Write(encodeUint16(pr.MessageID))
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (pr *PubrelPacket) Unpack(b io.Reader) error {
pr.MessageID = decodeUint16(b)
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (pr *PubrelPacket) Details() Details {
return Details{Qos: pr.Qos, MessageID: pr.MessageID}
}

52
packets/suback.go
View File

@@ -1,52 +0,0 @@
package packets
import (
"bytes"
"fmt"
"io"
)
//SubackPacket is an internal representation of the fields of the
//Suback MQTT packet
type SubackPacket struct {
FixedHeader
MessageID uint16
ReturnCodes []byte
}
func (sa *SubackPacket) String() string {
str := fmt.Sprintf("%s", sa.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", sa.MessageID)
return str
}
func (sa *SubackPacket) Write(w io.Writer) error {
var body bytes.Buffer
var err error
body.Write(encodeUint16(sa.MessageID))
body.Write(sa.ReturnCodes)
sa.FixedHeader.RemainingLength = body.Len()
packet := sa.FixedHeader.pack()
packet.Write(body.Bytes())
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (sa *SubackPacket) Unpack(b io.Reader) error {
var qosBuffer bytes.Buffer
sa.MessageID = decodeUint16(b)
qosBuffer.ReadFrom(b)
sa.ReturnCodes = qosBuffer.Bytes()
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (sa *SubackPacket) Details() Details {
return Details{Qos: 0, MessageID: sa.MessageID}
}

62
packets/subscribe.go
View File

@@ -1,62 +0,0 @@
package packets
import (
"bytes"
"fmt"
"io"
)
//SubscribePacket is an internal representation of the fields of the
//Subscribe MQTT packet
type SubscribePacket struct {
FixedHeader
MessageID uint16
Topics []string
Qoss []byte
}
func (s *SubscribePacket) String() string {
str := fmt.Sprintf("%s", s.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d topics: %s", s.MessageID, s.Topics)
return str
}
func (s *SubscribePacket) Write(w io.Writer) error {
var body bytes.Buffer
var err error
body.Write(encodeUint16(s.MessageID))
for i, topic := range s.Topics {
body.Write(encodeString(topic))
body.WriteByte(s.Qoss[i])
}
s.FixedHeader.RemainingLength = body.Len()
packet := s.FixedHeader.pack()
packet.Write(body.Bytes())
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (s *SubscribePacket) Unpack(b io.Reader) error {
s.MessageID = decodeUint16(b)
payloadLength := s.FixedHeader.RemainingLength - 2
for payloadLength > 0 {
topic := decodeString(b)
s.Topics = append(s.Topics, topic)
qos := decodeByte(b)
s.Qoss = append(s.Qoss, qos)
payloadLength -= 2 + len(topic) + 1 //2 bytes of string length, plus string, plus 1 byte for Qos
}
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (s *SubscribePacket) Details() Details {
return Details{Qos: 1, MessageID: s.MessageID}
}

44
packets/unsuback.go
View File

@@ -1,44 +0,0 @@
package packets
import (
"fmt"
"io"
)
//UnsubackPacket is an internal representation of the fields of the
//Unsuback MQTT packet
type UnsubackPacket struct {
FixedHeader
MessageID uint16
}
func (ua *UnsubackPacket) String() string {
str := fmt.Sprintf("%s", ua.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", ua.MessageID)
return str
}
func (ua *UnsubackPacket) Write(w io.Writer) error {
var err error
ua.FixedHeader.RemainingLength = 2
packet := ua.FixedHeader.pack()
packet.Write(encodeUint16(ua.MessageID))
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (ua *UnsubackPacket) Unpack(b io.Reader) error {
ua.MessageID = decodeUint16(b)
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (ua *UnsubackPacket) Details() Details {
return Details{Qos: 0, MessageID: ua.MessageID}
}

55
packets/unsubscribe.go
View File

@@ -1,55 +0,0 @@
package packets
import (
"bytes"
"fmt"
"io"
)
//UnsubscribePacket is an internal representation of the fields of the
//Unsubscribe MQTT packet
type UnsubscribePacket struct {
FixedHeader
MessageID uint16
Topics []string
}
func (u *UnsubscribePacket) String() string {
str := fmt.Sprintf("%s", u.FixedHeader)
str += " "
str += fmt.Sprintf("MessageID: %d", u.MessageID)
return str
}
func (u *UnsubscribePacket) Write(w io.Writer) error {
var body bytes.Buffer
var err error
body.Write(encodeUint16(u.MessageID))
for _, topic := range u.Topics {
body.Write(encodeString(topic))
}
u.FixedHeader.RemainingLength = body.Len()
packet := u.FixedHeader.pack()
packet.Write(body.Bytes())
_, err = packet.WriteTo(w)
return err
}
//Unpack decodes the details of a ControlPacket after the fixed
//header has been read
func (u *UnsubscribePacket) Unpack(b io.Reader) error {
u.MessageID = decodeUint16(b)
var topic string
for topic = decodeString(b); topic != ""; topic = decodeString(b) {
u.Topics = append(u.Topics, topic)
}
return nil
}
//Details returns a Details struct containing the Qos and
//MessageID of this ControlPacket
func (u *UnsubscribePacket) Details() Details {
return Details{Qos: 1, MessageID: u.MessageID}
}