mqtt 5

Author: everesio

Makefile

@@ -2,7 +2,7 @@
 
 .PHONY: clean build fmt test
 
-TAG           ?= v0.1.1
+TAG           ?= v0.2.0
 
 BUILD_FLAGS   ?=
 BINARY        ?= mqtt-proxy

README.md

@@ -14,7 +14,7 @@ MQTT Proxy allows MQTT clients to send messages to other messaging systems
 
 * MQTT protocol
     * [x] [MQTT 3.1.1](http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/mqtt-v3.1.1.html)
-    * [ ] [MQTT 5.0](https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html)
+    * [x] [MQTT 5.0](https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html)
 * Publisher
     * [x] Noop
     * [x] [Apache Kafka](https://kafka.apache.org/)
@@ -95,6 +95,9 @@ prerequisites
     docker exec -it mqtt-client mosquitto_pub -L mqtt://mqtt-proxy:1883/dummy -m "test qos 0" --repeat 1 -q 0
     docker exec -it mqtt-client mosquitto_pub -L mqtt://mqtt-proxy:1883/dummy -m "test qos 1" --repeat 1 -q 1
     docker exec -it mqtt-client mosquitto_pub -L mqtt://mqtt-proxy:1883/dummy -m "test qos 2" --repeat 1 -q 2
+    docker exec -it mqtt-client mosquitto_pub -L mqtt://mqtt-proxy:1883/dummy -m "test qos 0 / v5" --repeat 1 -q 0 -V mqttv5
+    docker exec -it mqtt-client mosquitto_pub -L mqtt://mqtt-proxy:1883/dummy -m "test qos 1 / v5" --repeat 1 -q 1 -V mqttv5
+    docker exec -it mqtt-client mosquitto_pub -L mqtt://mqtt-proxy:1883/dummy -m "test qos 2 / v5" --repeat 1 -q 2 -V mqttv5
     ```
 
     * proxy using Kafka SSL listener

apis/auth.go

@@ -8,7 +8,7 @@ import (
 
 const (
 	AuthAccepted     = mqttproto.Accepted
-	AuthUnauthorized = mqttproto.RefusedNotAuthorized
+	AuthUnauthorized = mqttproto.RefusedBadUserNameOrPassword
 )
 
 type UserPasswordAuthRequest struct {

pkg/mqtt/codec/packets.go

@@ -7,6 +7,7 @@ import (
 
 	mqttproto "github.com/grepplabs/mqtt-proxy/pkg/mqtt/codec/proto"
 	mqtt311 "github.com/grepplabs/mqtt-proxy/pkg/mqtt/codec/v311"
+	mqtt5 "github.com/grepplabs/mqtt-proxy/pkg/mqtt/codec/v5"
 )
 
 func ReadPacket(r io.Reader, protocolVersion byte) (mqttproto.ControlPacket, error) {
@@ -26,7 +27,7 @@ func ReadPacket(r io.Reader, protocolVersion byte) (mqttproto.ControlPacket, err
 	case mqttproto.MQTT_3_1_1:
 		return mqtt311.ReadPacket(r)
 	case mqttproto.MQTT_5:
-		return nil, mqtt311.NewConnAckError(mqttproto.RefusedUnacceptableProtocolVersion, "mqtt5 is not supported yet")
+		return mqtt5.ReadPacket(r)
 	default:
 		return nil, mqtt311.NewConnAckError(mqttproto.RefusedUnacceptableProtocolVersion, fmt.Sprintf("unsupported protocol version %v", protocolVersion))
 	}

pkg/mqtt/codec/proto/constants.go

@@ -78,6 +78,7 @@ const (
 
 // MQTT 5 - 3.2.2.2 Connect Reason Code
 const (
-	RefusedUnspecifiedError           byte = 0x80 // 128
-	RefusedUnsupportedProtocolVersion byte = 0x84 // 132
+	RefusedV5UnspecifiedError           byte = 0x80 // 128
+	RefusedV5UnsupportedProtocolVersion byte = 0x84 // 132
+	RefusedV5BadUserNameOrPassword      byte = 0x86 // 134
 )

pkg/mqtt/codec/proto/encoding.go

@@ -62,20 +62,20 @@ func EncodeUint16(v uint16) []byte {
 	return b
 }
 
-func DecodeLength(r io.Reader) (int, error) {
+func DecodeUvarint(r io.Reader) (int, error) {
 	byteReader := newByteReader(r)
 	length, err := binary.ReadUvarint(byteReader)
 	if err != nil {
 		return 0, err
 	}
 	if byteReader.bytesRead > 4 {
-		return 0, fmt.Errorf("the maximum number of bytes in the length is 4, but was %d", byteReader.bytesRead)
+		return 0, fmt.Errorf("the maximum number of bytes in the variable byte integer is 4, but was %d", byteReader.bytesRead)
 	}
 	return int(length), nil
 }
 
-// WriteLength is a modified binary.PutUvarint
-func WriteLength(buffer *bytes.Buffer, x uint32) int {
+// WriteUvarint is a modified binary.PutUvarint
+func WriteUvarint(buffer *bytes.Buffer, x uint32) int {
 	i := 0
 	for x >= 0x80 {
 		buffer.WriteByte(byte(x) | 0x80)
@@ -104,3 +104,14 @@ func (r *byteReader) ReadByte() (byte, error) {
 	r.bytesRead += n
 	return r.buf[0], nil
 }
+
+type CountingReader struct {
+	Reader    io.Reader
+	BytesRead int
+}
+
+func (r *CountingReader) Read(p []byte) (n int, err error) {
+	n, err = r.Reader.Read(p)
+	r.BytesRead += n
+	return n, err
+}

pkg/mqtt/codec/proto/fixedheader.go

@@ -44,14 +44,14 @@ func (fh *FixedHeader) Unpack(typeAndFlags byte, r io.Reader) (err error) {
 	fh.Dup = (typeAndFlags & 0x08) == 0x08
 	fh.Qos = (typeAndFlags & 0x06) >> 1
 	fh.Retain = (typeAndFlags & 0x01) != 0
-	fh.RemainingLength, err = DecodeLength(r)
+	fh.RemainingLength, err = DecodeUvarint(r)
 	return err
 }
 
 func (fh *FixedHeader) Pack() bytes.Buffer {
 	var header bytes.Buffer
 	header.WriteByte(fh.getFixedHeaderByte1())
-	WriteLength(&header, uint32(fh.RemainingLength))
+	WriteUvarint(&header, uint32(fh.RemainingLength))
 	return header
 }
 

pkg/mqtt/codec/proto/fixedheader_test.go

@@ -145,12 +145,12 @@ func TestFixedHeaderDecodeError(t *testing.T) {
 		{
 			name:     "remaining length too long",
 			input:    "008080808001",
-			expected: "the maximum number of bytes in the length is 4, but was 5",
+			expected: "the maximum number of bytes in the variable byte integer is 4, but was 5",
 		},
 		{
 			name:     "remaining length too long",
 			input:    "00808080808001",
-			expected: "the maximum number of bytes in the length is 4, but was 6",
+			expected: "the maximum number of bytes in the variable byte integer is 4, but was 6",
 		},
 	}
 	for _, tc := range tests {

pkg/mqtt/codec/v311/connack_test.go

@@ -15,6 +15,7 @@ func TestNewConnackPacket(t *testing.T) {
 	packet := NewControlPacket(mqttproto.CONNACK).(*ConnackPacket)
 	a.Equal(mqttproto.CONNACK, packet.MessageType)
 	a.Equal(mqttproto.MqttMessageTypeNames[packet.MessageType], packet.Name())
+	a.Equal(mqttproto.MQTT_3_1_1, packet.Version())
 	t.Log(packet)
 }
 
@@ -78,6 +79,7 @@ func TestConnackPacketCodec(t *testing.T) {
 			}
 			packet := decoded.(*ConnackPacket)
 			a.Equal(*tc.packet, *packet)
+			a.Equal(mqttproto.MQTT_3_1_1, packet.Version())
 
 			// encode
 			var output bytes.Buffer

pkg/mqtt/codec/v311/connect_test.go

@@ -15,6 +15,7 @@ func TestNewConnectPacket(t *testing.T) {
 	packet := NewControlPacket(mqttproto.CONNECT).(*ConnectPacket)
 	a.Equal(mqttproto.CONNECT, packet.MessageType)
 	a.Equal(mqttproto.MqttMessageTypeNames[packet.MessageType], packet.Name())
+	a.Equal(mqttproto.MQTT_3_1_1, packet.Version())
 	t.Log(packet)
 }
 
@@ -115,6 +116,7 @@ func TestDecodeConnectPacket(t *testing.T) {
 			}
 			packet := decoded.(*ConnectPacket)
 			a.Equal(*tc.packet, *packet)
+			a.Equal(mqttproto.MQTT_3_1_1, packet.Version())
 
 			// encode
 			var output bytes.Buffer