<|start_filename|>spec/utils.go<|end_filename|>
package spec

import (
	"time"

	"github.com/256dpi/gomqtt/packet"
)

func safeReceive(ch chan struct{}) {
	select {
	case <-time.After(10 * time.Second):
		panic("nothing received")
	case <-ch:
	}
}

func lower(a, b packet.QOS) packet.QOS {
	if a < b {
		return a
	}

	return b
}


<|start_filename|>client/futures.go<|end_filename|>
package client

import (
	"time"

	"github.com/256dpi/gomqtt/client/future"
	"github.com/256dpi/gomqtt/packet"
)

// A GenericFuture is returned by publish and unsubscribe methods.
type GenericFuture interface {
	// Wait will wait the given amount of time and return whether the future has
	// been completed, canceled or the request timed out. If no time has been
	// provided the wait will never timeout.

	// Note: Wait will not return any Client related errors.
	Wait(timeout time.Duration) error
}

// A ConnectFuture is returned by the connect method.
type ConnectFuture interface {
	GenericFuture

	// SessionPresent will return whether a session was present.
	SessionPresent() bool

	// ReturnCode will return the connack code returned by the broker.
	ReturnCode() packet.ConnackCode
}

// A SubscribeFuture is returned by the subscribe methods.
type SubscribeFuture interface {
	GenericFuture

	// ReturnCodes will return the suback codes returned by the broker.
	ReturnCodes() []packet.QOS
}

type connectFuture struct {
	*future.Future
}

func (f *connectFuture) SessionPresent() bool {
	// get result
	connack := f.Result().(*packet.Connack)
	if connack == nil {
		return false
	}

	return connack.SessionPresent
}

func (f *connectFuture) ReturnCode() packet.ConnackCode {
	// get result
	connack := f.Result().(*packet.Connack)
	if connack == nil {
		return 0
	}

	return connack.ReturnCode
}

type subscribeFuture struct {
	*future.Future
}

func (f *subscribeFuture) ReturnCodes() []packet.QOS {
	// get result
	suback := f.Result().(*packet.Suback)
	if suback == nil {
		return nil
	}

	return suback.ReturnCodes
}


<|start_filename|>packet/suback_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestSubackInterface(t *testing.T) {
	pkt := NewSuback()
	pkt.ReturnCodes = []QOS{0, 1}

	assert.Equal(t, pkt.Type(), SUBACK)
	assert.Equal(t, "<Suback ID=0 ReturnCodes=[0, 1]>", pkt.String())
}

func TestSubackDecode(t *testing.T) {
	packet := []byte{
		byte(SUBACK << 4),
		6,
		0,    // packet ID MSB
		7,    // packet ID LSB
		0,    // return code 1
		1,    // return code 2
		2,    // return code 3
		0x80, // return code 4
	}

	pkt := NewSuback()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, 4, len(pkt.ReturnCodes))
}

func TestSubackDecodeError1(t *testing.T) {
	packet := []byte{
		byte(SUBACK << 4),
		1, // < wrong remaining length
		0, // packet ID MSB
		7, // packet ID LSB
		0, // return code 1
	}

	pkt := NewSuback()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubackDecodeError2(t *testing.T) {
	packet := []byte{
		byte(SUBACK << 4),
		6,
		0,    // packet ID MSB
		7,    // packet ID LSB
		0,    // return code 1
		1,    // return code 2
		2,    // return code 3
		0x81, // < wrong return code
	}

	pkt := NewSuback()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubackDecodeError3(t *testing.T) {
	packet := []byte{
		byte(SUBACK << 4),
		1, // < wrong remaining length
		0, // packet ID MSB
	}

	pkt := NewSuback()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubackDecodeError4(t *testing.T) {
	packet := []byte{
		byte(PUBCOMP << 4), // < wrong packet type
		3,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // return code 1
	}

	pkt := NewSuback()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubackDecodeError5(t *testing.T) {
	packet := []byte{
		byte(SUBACK << 4),
		3,
		0, // packet ID MSB
		0, // packet ID LSB < zero packet id
		0,
	}

	pkt := NewSuback()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubackEncode(t *testing.T) {
	packet := []byte{
		byte(SUBACK << 4),
		6,
		0,    // packet ID MSB
		7,    // packet ID LSB
		0,    // return code 1
		1,    // return code 2
		2,    // return code 3
		0x80, // return code 4
	}

	pkt := NewSuback()
	pkt.ID = 7
	pkt.ReturnCodes = []QOS{0, 1, 2, 0x80}

	dst := make([]byte, 10)
	n, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, packet, dst[:n])
}

func TestSubackEncodeError1(t *testing.T) {
	pkt := NewSuback()
	pkt.ID = 7
	pkt.ReturnCodes = []QOS{0x81}

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
}

func TestSubackEncodeError2(t *testing.T) {
	pkt := NewSuback()
	pkt.ID = 7
	pkt.ReturnCodes = []QOS{0x80}

	dst := make([]byte, pkt.Len()-1)
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
}

func TestSubackEncodeError3(t *testing.T) {
	pkt := NewSuback()
	pkt.ID = 0 // < zero packet id
	pkt.ReturnCodes = []QOS{0x80}

	dst := make([]byte, pkt.Len()-1)
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
}

func TestSubackEqualDecodeEncode(t *testing.T) {
	packet := []byte{
		byte(SUBACK << 4),
		6,
		0,    // packet ID MSB
		7,    // packet ID LSB
		0,    // return code 1
		1,    // return code 2
		2,    // return code 3
		0x80, // return code 4
	}

	pkt := NewSuback()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)

	dst := make([]byte, 100)
	n2, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n2)
	assert.Equal(t, packet, dst[:n2])

	n3, err := pkt.Decode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n3)
}

func BenchmarkSubackEncode(b *testing.B) {
	b.ReportAllocs()

	pkt := NewSuback()
	pkt.ID = 1
	pkt.ReturnCodes = []QOS{0}

	buf := make([]byte, pkt.Len())

	for i := 0; i < b.N; i++ {
		_, err := pkt.Encode(buf)
		if err != nil {
			panic(err)
		}
	}
}

func BenchmarkSubackDecode(b *testing.B) {
	b.ReportAllocs()

	packet := []byte{
		byte(SUBACK << 4),
		3,
		0, // packet ID MSB
		1, // packet ID LSB
		0, // return code 1
	}

	pkt := NewSuback()

	for i := 0; i < b.N; i++ {
		_, err := pkt.Decode(packet)
		if err != nil {
			panic(err)
		}
	}
}


<|start_filename|>transport/server_test.go<|end_filename|>
package transport

import (
	"fmt"
	"io"
	"testing"

	"github.com/256dpi/gomqtt/packet"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func abstractServerTest(t *testing.T, protocol string) {
	server, err := testLauncher.Launch(protocol + "://localhost:0")
	require.NoError(t, err)

	wait := make(chan struct{})

	go func() {
		conn1, err := server.Accept()
		require.NoError(t, err)

		pkt, err := conn1.Receive()
		assert.NoError(t, err)
		assert.Equal(t, pkt.Type(), packet.CONNECT)

		err = conn1.Send(packet.NewConnack(), false)
		assert.NoError(t, err)

		pkt, err = conn1.Receive()
		assert.Nil(t, pkt)
		assert.Equal(t, io.EOF, err)

		close(wait)
	}()

	conn2, err := testDialer.Dial(getURL(server, protocol))
	require.NoError(t, err)

	err = conn2.Send(packet.NewConnect(), false)
	assert.NoError(t, err)

	pkt, err := conn2.Receive()
	assert.NoError(t, err)
	assert.Equal(t, pkt.Type(), packet.CONNACK)

	err = conn2.Close()
	assert.NoError(t, err)

	safeReceive(wait)

	err = server.Close()
	assert.NoError(t, err)
}

func abstractServerLaunchErrorTest(t *testing.T, protocol string) {
	server, err := testLauncher.Launch(protocol + "://localhost:1")
	assert.Error(t, err)
	assert.Nil(t, server)
}

func abstractServerAcceptAfterCloseTest(t *testing.T, protocol string) {
	server, err := testLauncher.Launch(protocol + "://localhost:0")
	require.NoError(t, err)

	err = server.Close()
	assert.NoError(t, err)

	conn, err := server.Accept()
	assert.Nil(t, conn)
	assert.Error(t, err)
}

func abstractServerCloseAfterCloseTest(t *testing.T, protocol string) {
	server, err := testLauncher.Launch(protocol + "://localhost:0")
	require.NoError(t, err)

	err = server.Close()
	assert.NoError(t, err)

	err = server.Close()
	assert.Error(t, err)
}

func abstractServerAddrTest(t *testing.T, protocol string) {
	server, err := testLauncher.Launch(protocol + "://localhost:0")
	require.NoError(t, err)

	assert.Equal(t, fmt.Sprintf("127.0.0.1:%s", getPort(server)), server.Addr().String())

	err = server.Close()
	assert.NoError(t, err)
}


<|start_filename|>broker/backend_test.go<|end_filename|>
package broker

import (
	"testing"
	"time"

	"github.com/256dpi/gomqtt/client"
	"github.com/256dpi/gomqtt/packet"
	"github.com/256dpi/gomqtt/spec"

	"github.com/stretchr/testify/assert"
)

func TestBrokerWithMemoryBackend(t *testing.T) {
	backend := NewMemoryBackend()
	backend.Credentials = map[string]string{
		"allow": "allow",
	}

	port, quit, done := Run(NewEngine(backend), "tcp")

	config := spec.AllFeatures()
	config.URL = "tcp://allow:allow@localhost:" + port
	config.DenyURL = "tcp://deny:deny@localhost:" + port
	config.ProcessWait = 20 * time.Millisecond
	config.NoMessageWait = 50 * time.Millisecond
	config.MessageRetainWait = 50 * time.Millisecond

	spec.Run(t, config)

	close(quit)

	safeReceive(done)
}

func TestMemoryBackendClose(t *testing.T) {
	backend := NewMemoryBackend()

	port, quit, done := Run(NewEngine(backend), "tcp")

	options := client.NewConfigWithClientID("tcp://localhost:"+port, "close1")
	options.CleanSession = false

	wait1 := make(chan struct{})
	client1 := client.New()
	client1.Callback = func(msg *packet.Message, err error) error {
		assert.Nil(t, msg)
		assert.Error(t, err)
		close(wait1)

		return nil
	}

	cf, err := client1.Connect(options)
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.ConnectionAccepted, cf.ReturnCode())
	assert.False(t, cf.SessionPresent())

	options.ClientID = "close2"
	options.CleanSession = true

	wait2 := make(chan struct{})
	client2 := client.New()
	client2.Callback = func(msg *packet.Message, err error) error {
		assert.Nil(t, msg)
		assert.Error(t, err)
		close(wait2)

		return nil
	}

	cf, err = client2.Connect(options)
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.ConnectionAccepted, cf.ReturnCode())
	assert.False(t, cf.SessionPresent())

	ret := backend.Close(5 * time.Second)
	assert.True(t, ret)

	safeReceive(wait1)
	safeReceive(wait2)

	close(quit)

	safeReceive(done)
}


<|start_filename|>broker/engine_test.go<|end_filename|>
package broker

import (
	"testing"
	"time"

	"github.com/256dpi/gomqtt/client"
	"github.com/256dpi/gomqtt/packet"
	"github.com/256dpi/gomqtt/transport"

	"github.com/stretchr/testify/assert"
)

func TestConnectTimeout(t *testing.T) {
	engine := NewEngine(NewMemoryBackend())
	engine.ConnectTimeout = 10 * time.Millisecond

	port, quit, done := Run(engine, "tcp")

	conn, err := transport.Dial("tcp://localhost:" + port)
	assert.NoError(t, err)

	pkt, err := conn.Receive()
	assert.Nil(t, pkt)
	assert.Error(t, err)

	close(quit)
	safeReceive(done)
}

func TestDefaultReadLimit(t *testing.T) {
	engine := NewEngine(NewMemoryBackend())
	engine.ReadLimit = 1

	port, quit, done := Run(engine, "tcp")

	c := client.New()
	wait := make(chan struct{})

	c.Callback = func(msg *packet.Message, err error) error {
		assert.Error(t, err)
		close(wait)
		return nil
	}

	cf, err := c.Connect(client.NewConfig("tcp://localhost:" + port))
	assert.NoError(t, err)
	assert.Error(t, cf.Wait(10*time.Second))

	safeReceive(wait)
	close(quit)
	safeReceive(done)
}


<|start_filename|>broker/utils_test.go<|end_filename|>
package broker

import "time"

func safeReceive(ch chan struct{}) {
	select {
	case <-time.After(1 * time.Minute):
		panic("nothing received")
	case <-ch:
	}
}


<|start_filename|>packet/naked_test.go<|end_filename|>
package packet

import (
	"fmt"
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestNakedDecode(t *testing.T) {
	packet := []byte{
		byte(DISCONNECT << 4),
		0,
	}

	n, err := nakedDecode(packet, DISCONNECT)

	assert.NoError(t, err)
	assert.Equal(t, 2, n)
}

func TestNakedDecodeError1(t *testing.T) {
	packet := []byte{
		byte(DISCONNECT << 4),
		1, // < wrong remaining length
		0,
	}

	n, err := nakedDecode(packet, DISCONNECT)

	assert.Error(t, err)
	assert.Equal(t, 2, n)
}

func TestNakedEncode(t *testing.T) {
	packet := []byte{
		byte(DISCONNECT << 4),
		0,
	}

	dst := make([]byte, nakedLen())
	n, err := nakedEncode(dst, DISCONNECT)

	assert.NoError(t, err)
	assert.Equal(t, 2, n)
	assert.Equal(t, packet, dst[:n])
}

func TestNakedEqualDecodeEncode(t *testing.T) {
	packet := []byte{
		byte(DISCONNECT << 4),
		0,
	}

	n, err := nakedDecode(packet, DISCONNECT)

	assert.NoError(t, err)
	assert.Equal(t, 2, n)

	dst := make([]byte, nakedLen())
	n2, err := nakedEncode(dst, DISCONNECT)

	assert.NoError(t, err)
	assert.Equal(t, 2, n2)
	assert.Equal(t, packet, dst[:n2])

	n3, err := nakedDecode(dst, DISCONNECT)

	assert.NoError(t, err)
	assert.Equal(t, 2, n3)
}

func BenchmarkNakedEncode(b *testing.B) {
	b.ReportAllocs()

	buf := make([]byte, nakedLen())

	for i := 0; i < b.N; i++ {
		_, err := nakedEncode(buf, DISCONNECT)
		if err != nil {
			panic(err)
		}
	}
}

func BenchmarkNakedDecode(b *testing.B) {
	b.ReportAllocs()

	packet := []byte{
		byte(DISCONNECT << 4),
		0,
	}

	for i := 0; i < b.N; i++ {
		_, err := nakedDecode(packet, DISCONNECT)
		if err != nil {
			panic(err)
		}
	}
}

func testNakedImplementation(t *testing.T, _t Type) {
	pkt, err := _t.New()
	assert.NoError(t, err)
	assert.Equal(t, _t, pkt.Type())
	assert.Equal(t, fmt.Sprintf("<%s>", pkt.Type().String()), pkt.String())

	buf := make([]byte, pkt.Len())
	n, err := pkt.Encode(buf)
	assert.NoError(t, err)
	assert.Equal(t, 2, n)

	n, err = pkt.Decode(buf)
	assert.NoError(t, err)
	assert.Equal(t, 2, n)
}

func TestDisconnectImplementation(t *testing.T) {
	testNakedImplementation(t, DISCONNECT)
}

func TestPingreqImplementation(t *testing.T) {
	testNakedImplementation(t, PINGREQ)
}

func TestPingrespImplementation(t *testing.T) {
	testNakedImplementation(t, PINGRESP)
}


<|start_filename|>client/future/future_test.go<|end_filename|>
package future

import (
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
)

func TestFutureCompleteBefore(t *testing.T) {
	f := New()
	f.Complete(1)
	assert.NoError(t, f.Wait(10*time.Millisecond))
	assert.Equal(t, 1, f.Result())
}

func TestFutureCompleteAfter(t *testing.T) {
	f := New()

	time.AfterFunc(time.Millisecond, func() {
		f.Complete(1)
	})

	assert.NoError(t, f.Wait(10*time.Millisecond))
	assert.Equal(t, 1, f.Result())
}

func TestFutureCancelBefore(t *testing.T) {
	f := New()
	f.Cancel(1)
	assert.Equal(t, ErrCanceled, f.Wait(10*time.Millisecond))
	assert.Equal(t, 1, f.Result())
}

func TestFutureCancelAfter(t *testing.T) {
	f := New()

	time.AfterFunc(time.Millisecond, func() {
		f.Cancel(1)
	})

	assert.Equal(t, ErrCanceled, f.Wait(10*time.Millisecond))
	assert.Equal(t, 1, f.Result())
}

func TestFutureTimeout(t *testing.T) {
	f := New()
	assert.Equal(t, ErrTimeout, f.Wait(1*time.Millisecond))
}

func TestFutureBindBefore(t *testing.T) {
	f1 := New()
	f1.Cancel(1)

	f2 := New()
	f1.Attach(f2)

	err := f2.Wait(10 * time.Millisecond)
	assert.Equal(t, ErrCanceled, err)
	assert.Equal(t, 1, f2.Result())
}

func TestFutureBindAfter(t *testing.T) {
	f1 := New()

	time.AfterFunc(time.Millisecond, func() {
		f1.Cancel(1)
	})

	f2 := New()
	f1.Attach(f2)

	err := f2.Wait(10 * time.Millisecond)
	assert.Equal(t, ErrCanceled, err)
	assert.Equal(t, 1, f2.Result())
}


<|start_filename|>Makefile<|end_filename|>
check:
	go fmt ./...
	go vet ./...
	golint ./...

install:
	go install ...
	go mod tidy

cert:
	mkcert -install
	mkcert -cert-file example.crt -key-file example.key example.com localhost 127.0.0.1


<|start_filename|>packet/unsubscribe_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestUnsubscribeInterface(t *testing.T) {
	pkt := NewUnsubscribe()
	pkt.Topics = []string{"foo", "bar"}

	assert.Equal(t, pkt.Type(), UNSUBSCRIBE)
	assert.Equal(t, "<Unsubscribe Topics=[\"foo\", \"bar\"]>", pkt.String())
}

func TestUnsubscribeDecode(t *testing.T) {
	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		32,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // topic name MSB
		8, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c',
		0,  // topic name MSB
		10, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c', 'd', 'd',
	}

	pkt := NewUnsubscribe()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, 3, len(pkt.Topics))
	assert.Equal(t, "gomqtt", pkt.Topics[0])
	assert.Equal(t, "/a/b/#/c", pkt.Topics[1])
	assert.Equal(t, "/a/b/#/cdd", pkt.Topics[2])
}

func TestUnsubscribeDecodeError1(t *testing.T) {
	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		2,
		0, // packet ID MSB
		7, // packet ID LSB
		// empty topic list
	}

	pkt := NewUnsubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestUnsubscribeDecodeError2(t *testing.T) {
	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		6, // < wrong remaining length
		0, // packet ID MSB
		7, // packet ID LSB
	}

	pkt := NewUnsubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestUnsubscribeDecodeError3(t *testing.T) {
	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		0,
		// missing packet id
	}

	pkt := NewUnsubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestUnsubscribeDecodeError4(t *testing.T) {
	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		10,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		9, // topic name LSB < wrong size
		'g', 'o', 'm', 'q', 't', 't',
	}

	pkt := NewUnsubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestUnsubscribeDecodeError5(t *testing.T) {
	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		10,
		0, // packet ID MSB
		0, // packet ID LSB < zero packet id
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
	}

	pkt := NewUnsubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestUnsubscribeEncode(t *testing.T) {
	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		32,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // topic name MSB
		8, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c',
		0,  // topic name MSB
		10, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c', 'd', 'd',
	}

	pkt := NewUnsubscribe()
	pkt.ID = 7
	pkt.Topics = []string{
		"gomqtt",
		"/a/b/#/c",
		"/a/b/#/cdd",
	}

	dst := make([]byte, 100)
	n, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, packet, dst[:n])
}

func TestUnsubscribeEncodeError1(t *testing.T) {
	pkt := NewUnsubscribe()
	pkt.ID = 7
	pkt.Topics = []string{"gomqtt"}

	dst := make([]byte, 1) // < too small
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
}

func TestUnsubscribeEncodeError2(t *testing.T) {
	pkt := NewUnsubscribe()
	pkt.ID = 7
	pkt.Topics = []string{string(make([]byte, 65536))}

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 6, n)
}

func TestUnsubscribeEncodeError3(t *testing.T) {
	pkt := NewUnsubscribe()
	pkt.ID = 0 // < zero packet id

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
}

func TestUnsubscribeEqualDecodeEncode(t *testing.T) {
	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		32,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // topic name MSB
		8, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c',
		0,  // topic name MSB
		10, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c', 'd', 'd',
	}

	pkt := NewUnsubscribe()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)

	dst := make([]byte, 100)
	n2, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n2)
	assert.Equal(t, packet, dst[:n2])

	n3, err := pkt.Decode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n3)
}

func BenchmarkUnsubscribeEncode(b *testing.B) {
	b.ReportAllocs()

	pkt := NewUnsubscribe()
	pkt.ID = 1
	pkt.Topics = []string{"t"}

	buf := make([]byte, pkt.Len())

	for i := 0; i < b.N; i++ {
		_, err := pkt.Encode(buf)
		if err != nil {
			panic(err)
		}
	}
}

func BenchmarkUnsubscribeDecode(b *testing.B) {
	b.ReportAllocs()

	packet := []byte{
		byte(UNSUBSCRIBE<<4) | 2,
		5,
		0, // packet ID MSB
		1, // packet ID LSB
		0, // topic name MSB
		1, // topic name LSB
		't',
	}

	pkt := NewUnsubscribe()

	for i := 0; i < b.N; i++ {
		_, err := pkt.Decode(packet)
		if err != nil {
			panic(err)
		}
	}
}


<|start_filename|>cmd/gomqtt-demo/main.go<|end_filename|>
package main

import (
	"errors"
	"flag"
	"fmt"
	"io"
	"math/rand"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"time"

	"github.com/Pallinder/go-randomdata"
	"gopkg.in/tomb.v2"

	"github.com/256dpi/gomqtt/client"
)

var broker = flag.String("broker", "tcp://0.0.0.0:1883", "the broker url")

var numTopics = flag.Int("topic", 10, "the number of topics")
var numRobots = flag.Int("robots", 7, "the number of robots")
var numSubs = flag.Int("subs", 2, "the number of subscriptions")
var timeout = flag.Duration("timeout", 5*time.Second, "the publish timeout")

func main() {
	// parse flags
	flag.Parse()

	// print info
	fmt.Println("Launching robots...")

	// generate topics
	topics := generateTopics(*numTopics)

	// run robots
	fmt.Println("launching robots...")
	var robots []*robot
	for i := 1; i <= *numRobots; i++ {
		robots = append(robots, runRobot(*broker, strconv.Itoa(i), *numSubs, topics, *timeout))
	}

	// wait until interrupt
	interrupt := make(chan os.Signal, 1)
	signal.Notify(interrupt, os.Interrupt)
	<-interrupt

	// close all robots
	fmt.Println("closing...")
	for _, r := range robots {
		r.kill()
	}
}

func generateTopics(num int) []string {
	// prepare topic levels
	l1 := make([]string, 0, num)
	l2 := make([]string, 0, num)
	l3 := make([]string, 0, num)

	// generate segments
	for i := 0; i < num; i++ {
		l1 = append(l1, randomdata.Adjective())
		l2 = append(l2, randomdata.Adjective())
		l3 = append(l3, randomdata.Adjective())
	}

	// prepare full topics
	topics := make([]string, 0, num)

	// combine levels
	for i := 0; i < num; i++ {
		topics = append(topics, strings.Join([]string{
			l1[rand.Intn(num)],
			l2[rand.Intn(num)],
			l3[rand.Intn(num)],
		}, "/"))
	}

	return topics
}

func randPick(list []string) string {
	return list[rand.Intn(len(list))]
}

type robot struct {
	topics  []string
	timeout time.Duration
	service *client.Service
	tomb    *tomb.Tomb
}

func runRobot(url, id string, subs int, topics []string, timeout time.Duration) *robot {
	// prepare config
	config := client.NewConfigWithClientID(url, "robot-"+id)

	// create service
	service := client.NewService()
	service.ResubscribeAllSubscriptions = true

	// make subscriptions
	for i := 0; i < subs; i++ {
		service.Subscribe(randPick(topics), 0)
	}

	// prepare robot with new service
	rob := &robot{
		topics:  topics,
		timeout: timeout,
		service: service,
	}

	// set online callback
	rob.service.OnlineCallback = func(resumed bool) {
		fmt.Println(config.ClientID + ": online")

		// run loop in new tomb
		rob.tomb = new(tomb.Tomb)
		rob.tomb.Go(rob.loop)
	}

	// set error callback
	rob.service.ErrorCallback = func(err error) {
		if !errors.Is(err, io.EOF) {
			fmt.Println(config.ClientID + ": " + err.Error())
		}
	}

	// set offline callback
	rob.service.OfflineCallback = func() {
		fmt.Println(config.ClientID + ": offline")

		// kill loop
		rob.tomb.Kill(nil)
		_ = rob.tomb.Wait()
		rob.tomb = nil
	}

	// start service
	rob.service.Start(config)

	return rob
}

func (r *robot) kill() {
	// stop service
	r.service.Stop(true)
}

func (r *robot) loop() error {
	for {
		// await random timeout
		select {
		case <-time.After(time.Duration(rand.Int63n(int64(r.timeout)))):
			// prepare payload
			payload := strconv.AppendInt(nil, rand.Int63n(100), 10)

			// publish message
			err := r.service.Publish(randPick(r.topics), payload, 0, false).Wait(time.Minute)
			if err != nil {
				return err
			}
		case <-r.tomb.Dying():
			return tomb.ErrDying
		}
	}
}


<|start_filename|>packet/coding_test.go<|end_filename|>
package packet

import (
	"encoding/binary"
	"math"
	"testing"

	"github.com/stretchr/testify/assert"
)

func binaryVarintLen(n uint64) int {
	return binary.PutUvarint(make([]byte, binary.MaxVarintLen64), n)
}

func TestVarintLen(t *testing.T) {
	assert.Equal(t, 1, varintLen(1))
	assert.Equal(t, 1, varintLen(127))
	assert.Equal(t, 2, varintLen(128))
	assert.Equal(t, 2, varintLen(16383))
	assert.Equal(t, 3, varintLen(16384))
	assert.Equal(t, 3, varintLen(2097151))
	assert.Equal(t, 4, varintLen(2097152))
	assert.Equal(t, 4, varintLen(268435455))
	assert.Equal(t, 0, varintLen(268435456))
	assert.Equal(t, 0, varintLen(math.MaxUint64))

	assert.Equal(t, 1, binaryVarintLen(1))
	assert.Equal(t, 1, binaryVarintLen(127))
	assert.Equal(t, 2, binaryVarintLen(128))
	assert.Equal(t, 2, binaryVarintLen(16383))
	assert.Equal(t, 3, binaryVarintLen(16384))
	assert.Equal(t, 3, binaryVarintLen(2097151))
	assert.Equal(t, 4, binaryVarintLen(2097152))
	assert.Equal(t, 4, binaryVarintLen(268435455))
	assert.Equal(t, 5, binaryVarintLen(268435456))
	assert.Equal(t, 10, binaryVarintLen(math.MaxUint64))
}

func TestReadVarint(t *testing.T) {
	num, n, err := readVarint([]byte{}, CONNECT)
	assert.Error(t, err)
	assert.Zero(t, num)
	assert.Zero(t, n)

	num, n, err = readVarint([]byte{0xff, 0xff}, CONNECT)
	assert.Error(t, err)
	assert.Zero(t, num)
	assert.Equal(t, 0, n)

	num, n, err = readVarint([]byte{0xff, 0xff, 0xff, 0xff, 0x1}, CONNECT)
	assert.Error(t, err)
	assert.Zero(t, num)
	assert.Equal(t, 0, n)

	num, n, err = readVarint([]byte{0xff, 0x42}, CONNECT)
	assert.NoError(t, err)
	assert.Equal(t, 8575, int(num))
	assert.Equal(t, 2, n)
}

func TestWriteVarint(t *testing.T) {
	n, err := writeVarint([]byte{}, 42, CONNECT)
	assert.Error(t, err)
	assert.Zero(t, n)

	n, err = writeVarint([]byte{0}, 8575, CONNECT)
	assert.Error(t, err)
	assert.Zero(t, n)

	n, err = writeVarint([]byte{0, 0, 0, 0}, maxVarint+1, CONNECT)
	assert.Error(t, err)
	assert.Zero(t, n)

	n, err = writeVarint([]byte{0, 0}, 8575, CONNECT)
	assert.NoError(t, err)
	assert.Equal(t, 2, n)
}

func TestReadLPBString(t *testing.T) {
	str, n, err := readLPString([]byte{}, CONNECT)
	assert.Error(t, err)
	assert.Empty(t, str)
	assert.Zero(t, n)

	str, n, err = readLPString([]byte{0xff, 0xff, 0xff, 0xff}, CONNECT)
	assert.Error(t, err)
	assert.Empty(t, str)
	assert.Equal(t, 2, n)

	str, n, err = readLPString([]byte{0x0, 0x0}, CONNECT)
	assert.NoError(t, err)
	assert.Equal(t, "", str)
	assert.Equal(t, 2, n)

	str, n, err = readLPString([]byte{0x0, 0x5, 'H', 'e', 'l', 'l', 'o'}, CONNECT)
	assert.NoError(t, err)
	assert.Equal(t, "Hello", str)
	assert.Equal(t, 7, n)
}

func TestWriteLPString(t *testing.T) {
	n, err := writeLPString([]byte{}, string(make([]byte, 65536)), CONNECT)
	assert.Error(t, err)
	assert.Zero(t, n)

	n, err = writeLPString([]byte{}, string(make([]byte, 10)), CONNECT)
	assert.Error(t, err)
	assert.Zero(t, n)

	buf := make([]byte, 7)
	n, err = writeLPString(buf, "Hello", CONNECT)
	assert.NoError(t, err)
	assert.Equal(t, []byte{0x0, 0x5, 'H', 'e', 'l', 'l', 'o'}, buf)
	assert.Equal(t, 7, n)
}


<|start_filename|>transport/base_conn.go<|end_filename|>
package transport

import (
	"io"
	"sync"
	"time"

	"github.com/256dpi/gomqtt/packet"
)

// A Carrier is a generalized stream that can be used with BaseConn.
type Carrier interface {
	io.ReadWriteCloser
	SetReadDeadline(time.Time) error
}

// A BaseConn manages the low-level plumbing between the Carrier and the packet
// Stream.
type BaseConn struct {
	carrier      Carrier
	stream       *packet.Stream
	sendMutex    sync.Mutex
	receiveMutex sync.Mutex
	readTimeout  time.Duration
}

// NewBaseConn creates a new BaseConn using the specified Carrier.
func NewBaseConn(c Carrier) *BaseConn {
	return &BaseConn{
		carrier: c,
		stream:  packet.NewStream(c, c),
	}
}

// Send will write the packet to an internal buffer. It will either flush the
// internal buffer immediately or asynchronously in the background when it gets
// stale. Encoding errors are directly returned, but any network errors caught
// while flushing the buffer asynchronously will be returned on the next call.
//
// Note: Only one goroutine can send at the same time.
func (c *BaseConn) Send(pkt packet.Generic, async bool) error {
	// acquire mutex
	c.sendMutex.Lock()
	defer c.sendMutex.Unlock()

	// write packet
	err := c.stream.Write(pkt, async)
	if err != nil {
		// ensure carrier gets closed
		_ = c.carrier.Close()

		return err
	}

	return nil
}

// Receive will read from the underlying connection and return a fully read
// packet. It will return any error encountered while decoding or reading from
// the underlying connection.
//
// Note: Only one goroutine can receive at the same time.
func (c *BaseConn) Receive() (packet.Generic, error) {
	// acquire mutex
	c.receiveMutex.Lock()
	defer c.receiveMutex.Unlock()

	// read next packet
	pkt, err := c.stream.Read()
	if err != nil {
		// ensure carrier gets closed
		_ = c.carrier.Close()

		return nil, err
	}

	// reset timeout
	err = c.resetTimeout()
	if err != nil {
		// ensure carrier is closed
		_ = c.carrier.Close()

		return nil, err
	}

	return pkt, nil
}

// Close will close the underlying connection and cleanup resources. It will
// return any error encountered while closing the underlying connection.
func (c *BaseConn) Close() error {
	// acquire mutex
	c.sendMutex.Lock()
	defer c.sendMutex.Unlock()

	// flush buffer
	err1 := c.stream.Flush()

	// close carrier
	err2 := c.carrier.Close()

	// handle errors
	if err1 != nil {
		return err1
	} else if err2 != nil {
		return err2
	}

	return nil
}

// SetReadLimit sets the maximum size of a packet that can be received.
// If the limit is greater than zero, Receive will close the connection and
// return an error if receiving the next packet will exceed the limit.
func (c *BaseConn) SetReadLimit(limit int64) {
	c.stream.SetReadLimit(limit)
}

// SetReadTimeout sets the maximum time that can pass between reads.
// If no data is received in the set duration the connection will be closed
// and Read returns an error.
func (c *BaseConn) SetReadTimeout(timeout time.Duration) {
	// acquire mutex
	c.receiveMutex.Lock()
	defer c.receiveMutex.Unlock()

	// set new timeout
	c.readTimeout = timeout

	// apply new timeout immediately
	_ = c.resetTimeout()
}

// SetMaxWriteDelay will set the maximum amount of time allowed to pass until
// an asynchronous write is flushed.
func (c *BaseConn) SetMaxWriteDelay(delay time.Duration) {
	c.stream.SetMaxWriteDelay(delay)
}

func (c *BaseConn) resetTimeout() error {
	// check timeout
	if c.readTimeout > 0 {
		return c.carrier.SetReadDeadline(time.Now().Add(c.readTimeout))
	}

	return c.carrier.SetReadDeadline(time.Time{})
}


<|start_filename|>packet/naked.go<|end_filename|>
package packet

func nakedLen() int {
	return headerLen(0)
}

func nakedDecode(src []byte, t Type) (int, error) {
	// decode header
	hl, _, rl, err := decodeHeader(src, t)
	if rl != 0 {
		return hl, makeError(t, "expected zero remaining length")
	}

	return hl, err
}

func nakedEncode(dst []byte, t Type) (int, error) {
	return encodeHeader(dst, 0, 0, nakedLen(), t)
}

// A Disconnect packet is sent from the client to the server.
// It indicates that the client is disconnecting cleanly.
type Disconnect struct{}

// NewDisconnect creates a new Disconnect packet.
func NewDisconnect() *Disconnect {
	return &Disconnect{}
}

// Type returns the packets type.
func (d *Disconnect) Type() Type {
	return DISCONNECT
}

// Len returns the byte length of the encoded packet.
func (d *Disconnect) Len() int {
	return nakedLen()
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (d *Disconnect) Decode(src []byte) (int, error) {
	return nakedDecode(src, DISCONNECT)
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (d *Disconnect) Encode(dst []byte) (int, error) {
	return nakedEncode(dst, DISCONNECT)
}

// String returns a string representation of the packet.
func (d *Disconnect) String() string {
	return "<Disconnect>"
}

// A Pingreq packet is sent from a client to the server.
type Pingreq struct{}

// NewPingreq creates a new Pingreq packet.
func NewPingreq() *Pingreq {
	return &Pingreq{}
}

// Type returns the packets type.
func (p *Pingreq) Type() Type {
	return PINGREQ
}

// Len returns the byte length of the encoded packet.
func (p *Pingreq) Len() int {
	return nakedLen()
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (p *Pingreq) Decode(src []byte) (int, error) {
	return nakedDecode(src, PINGREQ)
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (p *Pingreq) Encode(dst []byte) (int, error) {
	return nakedEncode(dst, PINGREQ)
}

// String returns a string representation of the packet.
func (p *Pingreq) String() string {
	return "<Pingreq>"
}

// A Pingresp packet is sent by the server to the client in response to a
// Pingreq. It indicates that the server is alive.
type Pingresp struct{}

// NewPingresp creates a new Pingresp packet.
func NewPingresp() *Pingresp {
	return &Pingresp{}
}

// Type returns the packets type.
func (p *Pingresp) Type() Type {
	return PINGRESP
}

// Len returns the byte length of the encoded packet.
func (p *Pingresp) Len() int {
	return nakedLen()
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (p *Pingresp) Decode(src []byte) (int, error) {
	return nakedDecode(src, PINGRESP)
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (p *Pingresp) Encode(dst []byte) (int, error) {
	return nakedEncode(dst, PINGRESP)
}

// String returns a string representation of the packet.
func (p *Pingresp) String() string {
	return "<Pingresp>"
}


<|start_filename|>transport/launcher_test.go<|end_filename|>
package transport

import (
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestGlobalLaunch(t *testing.T) {
	server, err := Launch("tcp://localhost:0")
	require.NoError(t, err)

	err = server.Close()
	assert.NoError(t, err)
}

func TestLauncherBadURL(t *testing.T) {
	conn, err := Launch("foo")
	assert.Nil(t, conn)
	assert.Error(t, err)
}

func TestLauncherUnsupportedProtocol(t *testing.T) {
	conn, err := Launch("foo://localhost")
	assert.Nil(t, conn)
	assert.Equal(t, ErrUnsupportedProtocol, err)
}


<|start_filename|>packet/publish_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestPublishInterface(t *testing.T) {
	pkt := NewPublish()

	assert.Equal(t, pkt.Type(), PUBLISH)
	assert.Equal(t, "<Publish ID=0 Message=<Message Topic=\"\" QOS=0 Retain=false Payload=> Dup=false>", pkt.String())
}

func TestPublishDecode1(t *testing.T) {
	packet := []byte{
		byte(PUBLISH<<4) | 11,
		22,
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // packet ID MSB
		7, // packet ID LSB
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
	}

	pkt := NewPublish()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, ID(7), pkt.ID)
	assert.Equal(t, "gomqtt", pkt.Message.Topic)
	assert.Equal(t, []byte("send me home"), pkt.Message.Payload)
	assert.Equal(t, QOS(1), pkt.Message.QOS)
	assert.Equal(t, true, pkt.Message.Retain)
	assert.Equal(t, true, pkt.Dup)
}

func TestPublishDecode2(t *testing.T) {
	packet := []byte{
		byte(PUBLISH << 4),
		20,
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
	}

	pkt := NewPublish()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, ID(0), pkt.ID)
	assert.Equal(t, "gomqtt", pkt.Message.Topic)
	assert.Equal(t, []byte("send me home"), pkt.Message.Payload)
	assert.Equal(t, QOS(0), pkt.Message.QOS)
	assert.Equal(t, false, pkt.Message.Retain)
	assert.Equal(t, false, pkt.Dup)
}

func TestPublishDecodeError1(t *testing.T) {
	packet := []byte{
		byte(PUBLISH << 4),
		2, // < too much
	}

	pkt := NewPublish()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestPublishDecodeError2(t *testing.T) {
	packet := []byte{
		byte(PUBLISH<<4) | 6, // < wrong qos
		0,
	}

	pkt := NewPublish()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestPublishDecodeError3(t *testing.T) {
	packet := []byte{
		byte(PUBLISH << 4),
		0,
		// < missing topic stuff
	}

	pkt := NewPublish()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestPublishDecodeError4(t *testing.T) {
	packet := []byte{
		byte(PUBLISH << 4),
		2,
		0, // topic name MSB
		1, // topic name LSB
		// < missing topic string
	}

	pkt := NewPublish()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestPublishDecodeError5(t *testing.T) {
	packet := []byte{
		byte(PUBLISH<<4) | 2,
		2,
		0, // topic name MSB
		1, // topic name LSB
		't',
		// < missing packet id
	}

	pkt := NewPublish()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestPublishDecodeError6(t *testing.T) {
	packet := []byte{
		byte(PUBLISH<<4) | 2,
		2,
		0, // topic name MSB
		1, // topic name LSB
		't',
		0,
		0, // < zero packet id
	}

	pkt := NewPublish()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestPublishEncode1(t *testing.T) {
	packet := []byte{
		byte(PUBLISH<<4) | 11,
		22,
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // packet ID MSB
		7, // packet ID LSB
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
	}

	pkt := NewPublish()
	pkt.Message.Topic = "gomqtt"
	pkt.Message.QOS = QOSAtLeastOnce
	pkt.Message.Retain = true
	pkt.Dup = true
	pkt.ID = 7
	pkt.Message.Payload = []byte("send me home")

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, packet, dst[:n])
}

func TestPublishEncode2(t *testing.T) {
	packet := []byte{
		byte(PUBLISH << 4),
		20,
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
	}

	pkt := NewPublish()
	pkt.Message.Topic = "gomqtt"
	pkt.Message.Payload = []byte("send me home")

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, packet, dst[:n])
}

func TestPublishEncodeError1(t *testing.T) {
	pkt := NewPublish()
	pkt.Message.Topic = "" // < empty topic

	dst := make([]byte, pkt.Len())
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestPublishEncodeError2(t *testing.T) {
	pkt := NewPublish()
	pkt.Message.Topic = "t"
	pkt.Message.QOS = 3 // < wrong qos

	dst := make([]byte, pkt.Len())
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestPublishEncodeError3(t *testing.T) {
	pkt := NewPublish()
	pkt.Message.Topic = "t"

	dst := make([]byte, 1) // < too small
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestPublishEncodeError4(t *testing.T) {
	pkt := NewPublish()
	pkt.Message.Topic = string(make([]byte, 65536)) // < too big

	dst := make([]byte, pkt.Len())
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestPublishEncodeError5(t *testing.T) {
	pkt := NewPublish()
	pkt.Message.Topic = "test"
	pkt.Message.QOS = 1
	pkt.ID = 0 // < zero packet id

	dst := make([]byte, pkt.Len())
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestPublishEqualDecodeEncode(t *testing.T) {
	packet := []byte{
		byte(PUBLISH<<4) | 2,
		22,
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // packet ID MSB
		7, // packet ID LSB
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
	}

	pkt := NewPublish()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)

	dst := make([]byte, pkt.Len())
	n2, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n2)
	assert.Equal(t, packet, dst[:n2])

	n3, err := pkt.Decode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n3)
}

func BenchmarkPublishEncode(b *testing.B) {
	b.ReportAllocs()

	pkt := NewPublish()
	pkt.Message.Topic = "t"
	pkt.Message.QOS = QOSAtLeastOnce
	pkt.ID = 1
	pkt.Message.Payload = []byte("p")

	buf := make([]byte, pkt.Len())

	for i := 0; i < b.N; i++ {
		_, err := pkt.Encode(buf)
		if err != nil {
			panic(err)
		}
	}
}

func BenchmarkPublishDecode(b *testing.B) {
	b.ReportAllocs()

	packet := []byte{
		byte(PUBLISH<<4) | 2,
		6,
		0, // topic name MSB
		1, // topic name LSB
		't',
		0, // packet ID MSB
		1, // packet ID LSB
		'p',
	}

	pkt := NewPublish()

	for i := 0; i < b.N; i++ {
		_, err := pkt.Decode(packet)
		if err != nil {
			panic(err)
		}
	}
}


<|start_filename|>transport/net_server_test.go<|end_filename|>
package transport

import (
	"testing"
)

func TestTCPServer(t *testing.T) {
	abstractServerTest(t, "tcp")
}

func TestTLSServer(t *testing.T) {
	abstractServerTest(t, "tls")
}

func TestTCPServerLaunchError(t *testing.T) {
	abstractServerLaunchErrorTest(t, "tcp")
}

func TestTLSServerLaunchError(t *testing.T) {
	abstractServerLaunchErrorTest(t, "tls")
}

func TestNetServerAcceptAfterClose(t *testing.T) {
	abstractServerAcceptAfterCloseTest(t, "tcp")
}

func TestNetServerCloseAfterClose(t *testing.T) {
	abstractServerCloseAfterCloseTest(t, "tcp")
}

func TestNetServerAddr(t *testing.T) {
	abstractServerAddrTest(t, "tcp")
}


<|start_filename|>client/utils_test.go<|end_filename|>
package client

import (
	"net"
	"testing"
	"time"

	"github.com/256dpi/gomqtt/packet"
	"github.com/256dpi/gomqtt/transport"
	"github.com/256dpi/gomqtt/transport/flow"

	"github.com/stretchr/testify/assert"
)

func safeReceive(ch chan struct{}) {
	select {
	case <-time.After(1 * time.Minute):
		panic("nothing received")
	case <-ch:
	}
}

func errorCallback(t *testing.T) func(*packet.Message, error) error {
	return func(msg *packet.Message, err error) error {
		if err != nil {
			println(err.Error())
		}

		assert.Fail(t, "callback should not have been called")
		return nil
	}
}

func fakeBroker(t *testing.T, testFlows ...*flow.Flow) (chan struct{}, string) {
	done := make(chan struct{})

	server, err := transport.Launch("tcp://localhost:0")
	assert.NoError(t, err)

	go func() {
		for _, flow := range testFlows {
			conn, err := server.Accept()
			assert.NoError(t, err)

			err = flow.Test(conn)
			assert.NoError(t, err)
		}

		err = server.Close()
		assert.NoError(t, err)

		close(done)
	}()

	_, port, _ := net.SplitHostPort(server.Addr().String())

	return done, port
}

func connectPacket() *packet.Connect {
	pkt := packet.NewConnect()
	pkt.CleanSession = true
	pkt.KeepAlive = 30
	return pkt
}

func connackPacket() *packet.Connack {
	pkt := packet.NewConnack()
	pkt.ReturnCode = packet.ConnectionAccepted
	pkt.SessionPresent = false
	return pkt
}

func disconnectPacket() *packet.Disconnect {
	return packet.NewDisconnect()
}


<|start_filename|>packet/utils.go<|end_filename|>
package packet

import "unsafe"

func cast(s string) []byte {
	return *(*[]byte)(unsafe.Pointer(&s))
}


<|start_filename|>packet/header.go<|end_filename|>
package packet

func headerLen(rl int) int {
	// add packet type and flag byte + remaining length
	return 1 + varintLen(uint64(rl))
}

func encodeHeader(dst []byte, flags byte, rl int, tl int, t Type) (int, error) {
	// check buffer length
	if len(dst) < headerLen(rl) || len(dst) < tl {
		return 0, insufficientBufferSize(t)
	}

	// write type and flags
	typeAndFlags := byte(t)<<4 | (t.defaultFlags() & 0xf)
	typeAndFlags |= flags
	dst[0] = typeAndFlags

	// write remaining length
	n, err := writeVarint(dst[1:], uint64(rl), t)
	if err != nil {
		return 0, err
	}

	return 1 + n, nil
}

func decodeHeader(src []byte, t Type) (int, byte, int, error) {
	// check buffer size
	if len(src) < 2 {
		return 0, 0, 0, insufficientBufferSize(t)
	}

	// read type and flags
	decodedType := Type(src[0] >> 4)
	flags := src[0] & 0x0f
	total := 1

	// check against static type
	if decodedType != t {
		return total, 0, 0, makeError(t, "invalid type %d", decodedType)
	}

	// check flags except for publish packets
	if t != PUBLISH && flags != t.defaultFlags() {
		return total, 0, 0, makeError(t, "invalid flags, expected %d, got %d", t.defaultFlags(), flags)
	}

	// read remaining length
	_rl, n, err := readVarint(src[total:], t)
	total += n
	if err != nil {
		return total, 0, 0, err
	}

	// get remaining length
	rl := int(_rl)

	// check remaining buffer
	if rl > len(src[total:]) {
		return total, 0, 0, makeError(t, "remaining length (%d) is greater than remaining buffer (%d)", rl, len(src[total:]))
	}

	return total, flags, rl, nil
}


<|start_filename|>client/tools_test.go<|end_filename|>
package client

import (
	"testing"
	"time"

	"github.com/256dpi/gomqtt/packet"
	"github.com/256dpi/gomqtt/transport/flow"

	"github.com/stretchr/testify/assert"
)

func TestClearSession(t *testing.T) {
	connect := connectPacket()
	connect.ClientID = "test"

	broker := flow.New().
		Receive(connect).
		Send(connackPacket()).
		Receive(disconnectPacket()).
		End()

	done, port := fakeBroker(t, broker)

	err := ClearSession(NewConfigWithClientID("tcp://localhost:"+port, "test"), 1*time.Second)
	assert.NoError(t, err)

	safeReceive(done)
}

func TestClearRetainedMessage(t *testing.T) {
	publish := packet.NewPublish()
	publish.Message.Topic = "test"
	publish.Message.Payload = nil
	publish.Message.Retain = true

	broker := flow.New().
		Receive(connectPacket()).
		Send(connackPacket()).
		Receive(publish).
		Receive(disconnectPacket()).
		End()

	done, port := fakeBroker(t, broker)

	err := ClearRetainedMessage(NewConfig("tcp://localhost:"+port), "test", 1*time.Second)
	assert.NoError(t, err)

	safeReceive(done)
}

func TestPublishMessage(t *testing.T) {
	publish := packet.NewPublish()
	publish.Message = packet.Message{
		Topic:   "test",
		Payload: []byte("test"),
		Retain:  true,
	}

	broker := flow.New().
		Receive(connectPacket()).
		Send(connackPacket()).
		Receive(publish).
		Receive(disconnectPacket()).
		End()

	done, port := fakeBroker(t, broker)

	err := PublishMessage(NewConfig("tcp://localhost:"+port), &publish.Message, 1*time.Second)
	assert.NoError(t, err)

	safeReceive(done)
}

func TestReceiveMessage(t *testing.T) {
	subscribe := packet.NewSubscribe()
	subscribe.ID = 1
	subscribe.Subscriptions = []packet.Subscription{
		{Topic: "test"},
	}

	suback := packet.NewSuback()
	suback.ID = 1
	suback.ReturnCodes = []packet.QOS{0}

	publish := packet.NewPublish()
	publish.Message = packet.Message{
		Topic:   "test",
		Payload: []byte("test"),
		Retain:  true,
	}

	broker := flow.New().
		Receive(connectPacket()).
		Send(connackPacket()).
		Receive(subscribe).
		Send(suback).
		Send(publish).
		Receive(disconnectPacket()).
		End()

	done, port := fakeBroker(t, broker)

	msg, err := ReceiveMessage(NewConfig("tcp://localhost:"+port), "test", 0, 1*time.Second)
	assert.NoError(t, err)
	assert.Equal(t, publish.Message.String(), msg.String())

	safeReceive(done)
}


<|start_filename|>client/tracker_test.go<|end_filename|>
package client

import (
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
)

func TestTracker(t *testing.T) {
	tracker := NewTracker(10 * time.Millisecond)
	assert.False(t, tracker.Pending())
	assert.True(t, tracker.Window() > 0)

	time.Sleep(10 * time.Millisecond)
	assert.True(t, tracker.Window() <= 0)

	tracker.Reset()
	assert.True(t, tracker.Window() > 0)

	tracker.Ping()
	assert.True(t, tracker.Pending())

	tracker.Pong()
	assert.False(t, tracker.Pending())
}


<|start_filename|>session/packet_store_test.go<|end_filename|>
package session

import (
	"testing"

	"github.com/256dpi/gomqtt/packet"

	"github.com/stretchr/testify/assert"
)

func TestPacketStore(t *testing.T) {
	store := NewPacketStore()

	publish := packet.NewPublish()
	publish.ID = 1

	pkt := store.Lookup(1)
	assert.Nil(t, pkt)

	store.Save(publish)

	pkt = store.Lookup(1)
	assert.Equal(t, publish, pkt)

	pkts := store.All()
	assert.Equal(t, 1, len(pkts))

	store.Delete(1)

	pkt = store.Lookup(1)
	assert.Nil(t, pkt)

	pkts = store.All()
	assert.Equal(t, 0, len(pkts))

	store = NewPacketStoreWithPackets([]packet.Generic{&packet.Subscribe{ID: 7}})
	assert.Equal(t, []packet.Generic{&packet.Subscribe{ID: 7}}, store.All())
}


<|start_filename|>packet/packet_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestQOSCodes(t *testing.T) {
	assert.Equal(t, QOS(0), QOSAtMostOnce)
	assert.Equal(t, QOS(1), QOSAtLeastOnce)
	assert.Equal(t, QOS(2), QOSExactlyOnce)
}

func TestFixedHeaderFlags(t *testing.T) {
	type detail struct {
		name  string
		flags byte
	}

	details := map[Type]detail{
		CONNECT:     {"Connect", 0},
		CONNACK:     {"Connack", 0},
		PUBLISH:     {"Publish", 0},
		PUBACK:      {"Puback", 0},
		PUBREC:      {"Pubrec", 0},
		PUBREL:      {"Pubrel", 2},
		PUBCOMP:     {"Pubcomp", 0},
		SUBSCRIBE:   {"Subscribe", 2},
		SUBACK:      {"Suback", 0},
		UNSUBSCRIBE: {"Unsubscribe", 2},
		UNSUBACK:    {"Unsuback", 0},
		PINGREQ:     {"Pingreq", 0},
		PINGRESP:    {"Pingresp", 0},
		DISCONNECT:  {"Disconnect", 0},
	}

	for m, d := range details {
		assert.Equal(t, d.name, m.String())
		assert.Equal(t, d.flags, m.defaultFlags())
	}
}

func TestDetect1(t *testing.T) {
	buf := []byte{0x10, 0x0}

	l, tt := DetectPacket(buf)
	assert.Equal(t, 2, l)
	assert.Equal(t, 1, int(tt))
}

func TestDetect2(t *testing.T) {
	// not enough bytes
	buf := []byte{0x10, 0xff}

	l, tt := DetectPacket(buf)
	assert.Equal(t, 0, l)
	assert.Equal(t, 0, int(tt))
}

func TestDetect3(t *testing.T) {
	buf := []byte{0x10, 0xff, 0x0}

	l, tt := DetectPacket(buf)
	assert.Equal(t, 130, l)
	assert.Equal(t, 1, int(tt))
}

func TestDetect4(t *testing.T) {
	// not enough bytes
	buf := []byte{0x10, 0xff, 0xff}

	l, tt := DetectPacket(buf)
	assert.Equal(t, 0, l)
	assert.Equal(t, 0, int(tt))
}

func TestDetect5(t *testing.T) {
	buf := []byte{0x10, 0xff, 0xff, 0xff, 0x7f}

	l, tt := DetectPacket(buf)
	assert.Equal(t, 268435460, l)
	assert.Equal(t, 1, int(tt))
}

func TestDetect6(t *testing.T) {
	buf := []byte{0x10}

	l, tt := DetectPacket(buf)
	assert.Equal(t, 0, l)
	assert.Equal(t, 0, int(tt))
}

func TestDetect7(t *testing.T) {
	buf := []byte{0x10, 0xff, 0xff, 0xff, 0x80}

	l, tt := DetectPacket(buf)
	assert.Equal(t, 0, l)
	assert.Equal(t, 0, int(tt))
}

func TestGetID(t *testing.T) {
	type detail struct {
		packet Generic
		ok     bool
	}

	details := map[Type]detail{
		CONNECT:     {NewConnect(), false},
		CONNACK:     {NewConnack(), false},
		PUBLISH:     {NewPublish(), true},
		PUBACK:      {NewPuback(), true},
		PUBREC:      {NewPubrec(), true},
		PUBREL:      {NewPubrel(), true},
		PUBCOMP:     {NewPubcomp(), true},
		SUBSCRIBE:   {NewSubscribe(), true},
		SUBACK:      {NewSuback(), true},
		UNSUBSCRIBE: {NewUnsubscribe(), true},
		UNSUBACK:    {NewUnsuback(), true},
		PINGREQ:     {NewPingreq(), false},
		PINGRESP:    {NewPingresp(), false},
		DISCONNECT:  {NewDisconnect(), false},
	}

	for _, d := range details {
		_, ok := GetID(d.packet)
		assert.Equal(t, d.ok, ok)
	}
}

func TestFuzz(t *testing.T) {
	// too small buffer
	assert.Equal(t, 1, Fuzz([]byte{}))

	// wrong packet type
	b1 := []byte{0 << 4, 0x00}
	assert.Equal(t, 0, Fuzz(b1))

	// wrong packet format
	b2 := []byte{2 << 4, 0x02, 0x00, 0x06}
	assert.Equal(t, 0, Fuzz(b2))

	// right packet format
	b3 := []byte{2 << 4, 0x02, 0x00, 0x01}
	assert.Equal(t, 1, Fuzz(b3))
}


<|start_filename|>packet/stream_test.go<|end_filename|>
package packet

import (
	"bytes"
	"errors"
	"io"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
)

func TestEncoder(t *testing.T) {
	buf := new(bytes.Buffer)
	enc := NewEncoder(buf)
	enc.SetMaxWriteDelay(time.Millisecond)

	err := enc.Write(NewConnect(), false)
	assert.NoError(t, err)

	err = enc.Flush()
	assert.NoError(t, err)

	assert.Len(t, buf.Bytes(), 14)
}

func TestEncoderEncodeError(t *testing.T) {
	buf := new(bytes.Buffer)
	enc := NewEncoder(buf)
	enc.SetMaxWriteDelay(time.Millisecond)

	pkt := NewConnack()
	pkt.ReturnCode = 11 // < invalid return code

	err := enc.Write(pkt, false)
	assert.Error(t, err)
}

func TestEncoderWriterError(t *testing.T) {
	enc := NewEncoder(&errorWriter{
		err: errors.New("foo"),
	})
	enc.SetMaxWriteDelay(time.Millisecond)

	pkt := NewPublish()
	pkt.Message.Topic = "foo"
	pkt.Message.Payload = make([]byte, 4096)

	err := enc.Write(pkt, false)
	assert.Error(t, err)
}

func TestDecoder(t *testing.T) {
	buf := new(bytes.Buffer)
	dec := NewDecoder(buf)

	var pkt Generic = NewConnect()
	b := make([]byte, pkt.Len())
	_, err := pkt.Encode(b)
	assert.NoError(t, err)
	buf.Write(b)

	pkt, err = dec.Read()
	assert.NoError(t, err)
	assert.NotNil(t, pkt)
}

func TestDecoderDetectionOverflowError(t *testing.T) {
	buf := new(bytes.Buffer)
	dec := NewDecoder(buf)

	buf.Write([]byte{0x10, 0xff, 0xff, 0xff, 0x80})

	pkt, err := dec.Read()
	assert.Equal(t, ErrDetectionOverflow, err)
	assert.Nil(t, pkt)
}

func TestDecoderPeekUnexpectedEOFError(t *testing.T) {
	buf := new(bytes.Buffer)
	dec := NewDecoder(buf)

	buf.Write([]byte{0x10, 0xff, 0xff})

	pkt, err := dec.Read()
	assert.Equal(t, io.ErrUnexpectedEOF, err)
	assert.Nil(t, pkt)
}

func TestDecoderPeekError(t *testing.T) {
	dec := NewDecoder(&errorReader{
		err: errors.New("foo"),
	})

	pkt, err := dec.Read()
	assert.Error(t, err)
	assert.Nil(t, pkt)
}

func TestDecoderInvalidTypeError(t *testing.T) {
	buf := new(bytes.Buffer)
	dec := NewDecoder(buf)

	buf.Write([]byte{0x00, 0x00})

	pkt, err := dec.Read()
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "invalid packet type")
	assert.Nil(t, pkt)
}

func TestDecoderReadLimitError(t *testing.T) {
	buf := new(bytes.Buffer)
	dec := NewDecoder(buf)
	dec.limit = 1

	buf.Write([]byte{0x00, 0x00})

	pkt, err := dec.Read()
	assert.Equal(t, ErrReadLimitExceeded, err)
	assert.Nil(t, pkt)
}

func TestDecoderReadError(t *testing.T) {
	dec := NewDecoder(&errorReader{
		reader: bytes.NewBuffer([]byte{0x10, 0xc, 0x0, 0x4}),
		after:  1,
		err:    errors.New("foo"),
	})

	pkt, err := dec.Read()
	assert.Error(t, err)
	assert.Nil(t, pkt)
}

func TestDecoderDecodeError(t *testing.T) {
	buf := new(bytes.Buffer)
	dec := NewDecoder(buf)

	buf.Write([]byte{0x20, 0x02, 0x40, 0x00})

	pkt, err := dec.Read()
	assert.Error(t, err)
	assert.Nil(t, pkt)
}

func TestStream(t *testing.T) {
	in := new(bytes.Buffer)
	out := new(bytes.Buffer)

	s := NewStream(in, out)
	s.SetMaxWriteDelay(time.Millisecond)

	err := s.Write(NewConnect(), false)
	assert.NoError(t, err)

	err = s.Flush()
	assert.NoError(t, err)

	_, err = io.Copy(in, out)
	assert.NoError(t, err)

	pkt, err := s.Read()
	assert.NotNil(t, pkt)
	assert.NoError(t, err)
}


<|start_filename|>session/id_counter_test.go<|end_filename|>
package session

import (
	"math"
	"testing"

	"github.com/256dpi/gomqtt/packet"

	"github.com/stretchr/testify/assert"
)

func TestIDCounter(t *testing.T) {
	counter := NewIDCounter()

	assert.Equal(t, packet.ID(1), counter.NextID())
	assert.Equal(t, packet.ID(2), counter.NextID())

	for i := 0; i < math.MaxUint16-3; i++ {
		counter.NextID()
	}

	assert.Equal(t, packet.ID(math.MaxUint16), counter.NextID())
	assert.Equal(t, packet.ID(1), counter.NextID())

	counter.Reset()

	assert.Equal(t, packet.ID(1), counter.NextID())

	counter = NewIDCounterWithNext(10)

	assert.Equal(t, packet.ID(10), counter.NextID())
}


<|start_filename|>packet/publish.go<|end_filename|>
package packet

import "fmt"

// A Publish packet is sent from a client to a server or from server to a client
// to transport an application message.
type Publish struct {
	// The message to publish.
	Message Message

	// If the Dup flag is set to false, it indicates that this is the first
	// occasion that the client or server has attempted to send this
	// Publish packet. If the dup flag is set to true, it indicates that this
	// might be re-delivery of an earlier attempt to send the packet.
	Dup bool

	// The packet identifier.
	ID ID
}

// NewPublish creates a new Publish packet.
func NewPublish() *Publish {
	return &Publish{}
}

// Type returns the packets type.
func (p *Publish) Type() Type {
	return PUBLISH
}

// String returns a string representation of the packet.
func (p *Publish) String() string {
	return fmt.Sprintf("<Publish ID=%d Message=%s Dup=%t>",
		p.ID, p.Message.String(), p.Dup)
}

// Len returns the byte length of the encoded packet.
func (p *Publish) Len() int {
	ml := p.len()
	return headerLen(ml) + ml
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (p *Publish) Decode(src []byte) (int, error) {
	// decode header
	hl, flags, rl, err := decodeHeader(src, PUBLISH)
	total := hl
	if err != nil {
		return total, err
	}

	// read flags
	p.Dup = ((flags >> 3) & 0x1) == 1
	p.Message.Retain = (flags & 0x1) == 1
	p.Message.QOS = QOS((flags >> 1) & 0x3)

	// check qos
	if !p.Message.QOS.Successful() {
		return total, makeError(PUBLISH, "invalid QOS level (%d)", p.Message.QOS)
	}

	// read topic
	topic, n, err := readLPString(src[total:], PUBLISH)
	total += n
	if err != nil {
		return total, err
	}

	// set topic
	p.Message.Topic = topic

	// check quality of service
	if p.Message.QOS != 0 {
		// check buffer length
		if len(src) < total+2 {
			return total, insufficientBufferSize(PUBLISH)
		}

		// read packet id
		pid, n, err := readUint(src[total:], 2, PUBLISH)
		total += n
		if err != nil {
			return total, err
		}

		// set packet id
		p.ID = ID(pid)
		if !p.ID.Valid() {
			return total, makeError(PUBLISH, "packet id must be grater than zero")
		}
	}

	// calculate payload length
	l := rl - (total - hl)

	// read payload
	if l > 0 {
		p.Message.Payload = make([]byte, l)
		copy(p.Message.Payload, src[total:total+l])
		total += len(p.Message.Payload)
	}

	return total, nil
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (p *Publish) Encode(dst []byte) (int, error) {
	// check topic length
	if len(p.Message.Topic) == 0 {
		return 0, makeError(PUBLISH, "topic name is empty")
	}

	// prepare flags
	var flags byte

	// set dup flag
	if p.Dup {
		flags |= 0x8 // 00001000
	}

	// set retain flag
	if p.Message.Retain {
		flags |= 0x1 // 00000001
	}

	// check qos
	if !p.Message.QOS.Successful() {
		return 0, makeError(PUBLISH, "invalid QOS level %d", p.Message.QOS)
	}

	// check packet id
	if p.Message.QOS > 0 && !p.ID.Valid() {
		return 0, makeError(PUBLISH, "packet id must be grater than zero")
	}

	// set qos
	flags = (flags & 249) | (byte(p.Message.QOS) << 1) // 249 = 11111001

	// encode header
	total, err := encodeHeader(dst, flags, p.len(), p.Len(), PUBLISH)
	if err != nil {
		return total, err
	}

	// write topic
	n, err := writeLPString(dst[total:], p.Message.Topic, PUBLISH)
	total += n
	if err != nil {
		return total, err
	}

	// write packet id
	if p.Message.QOS != 0 {
		n, err := writeUint(dst[total:], uint64(p.ID), 2, PUBLISH)
		total += n
		if err != nil {
			return total, err
		}
	}

	// write payload
	copy(dst[total:], p.Message.Payload)
	total += len(p.Message.Payload)

	return total, nil
}

func (p *Publish) len() int {
	// topic + payload
	total := 2 + len(p.Message.Topic) + len(p.Message.Payload)

	// packet iD
	if p.Message.QOS != 0 {
		total += 2
	}

	return total
}


<|start_filename|>transport/websocket_conn_test.go<|end_filename|>
package transport

import (
	"io"
	"testing"

	"github.com/256dpi/gomqtt/packet"

	"github.com/gorilla/websocket"
	"github.com/stretchr/testify/assert"
)

func TestWebSocketConnConnection(t *testing.T) {
	abstractConnConnectTest(t, "ws")
}

func TestWebSocketConnClose(t *testing.T) {
	abstractConnCloseTest(t, "ws")
}

func TestWebSocketConnEncodeError(t *testing.T) {
	abstractConnEncodeErrorTest(t, "ws")
}

func TestWebSocketConnDecodeError(t *testing.T) {
	abstractConnDecodeErrorTest(t, "ws")
}

func TestWebSocketConnSendAfterClose(t *testing.T) {
	abstractConnSendAfterCloseTest(t, "ws")
}

func TestWebSocketConnCloseWhileSend(t *testing.T) {
	abstractConnCloseWhileSendTest(t, "ws")
}

func TestWebSocketSendAndCloseTest(t *testing.T) {
	abstractConnSendAndCloseTest(t, "ws")
}

func TestWebSocketConnReadLimit(t *testing.T) {
	abstractConnReadLimitTest(t, "ws")
}

func TestWebSocketConnReadTimeout(t *testing.T) {
	abstractConnReadTimeoutTest(t, "ws")
}

func TestWebSocketConnCloseAfterClose(t *testing.T) {
	abstractConnCloseAfterCloseTest(t, "ws")
}

func TestWebSocketConnAddr(t *testing.T) {
	abstractConnAddrTest(t, "ws")
}

func TestWebSocketConnAsyncSend(t *testing.T) {
	abstractConnAsyncSendTest(t, "ws")
}

func TestWebSocketConnSendAfterAsyncSend(t *testing.T) {
	abstractConnSendAfterAsyncSendTest(t, "ws")
}

func TestWebSocketConnAsyncSendAfterClose(t *testing.T) {
	abstractConnAsyncSendAfterCloseTest(t, "ws")
}

func TestWebSocketConnCloseAfterAsyncSend(t *testing.T) {
	abstractConnCloseAfterAsyncSendTest(t, "ws")
}

func TestWebSocketConnBigAsyncSendAfterClose(t *testing.T) {
	abstractConnBigAsyncSendAfterCloseTest(t, "ws")
}

func TestWebSocketBadFrameError(t *testing.T) {
	conn2, done := connectionPair("ws", func(conn1 Conn) {
		buf := []byte{0x07, 0x00, 0x00, 0x00, 0x00} // < bad frame

		_, err := conn1.(*WebSocketConn).UnderlyingConn().UnderlyingConn().Write(buf)
		assert.NoError(t, err)

		pkt, err := conn1.Receive()
		assert.Nil(t, pkt)
		assert.Equal(t, io.EOF, err)
	})

	pkt, err := conn2.Receive()
	assert.Nil(t, pkt)
	assert.Error(t, err)

	safeReceive(done)
}

func TestWebSocketChunkedMessage(t *testing.T) {
	pkt := packet.NewPublish()
	pkt.Message.Topic = "hello"
	pkt.Message.Payload = []byte("world")

	conn2, done := connectionPair("ws", func(conn1 Conn) {
		buf := make([]byte, pkt.Len())
		_, err := pkt.Encode(buf)
		assert.NoError(t, err)

		err = conn1.(*WebSocketConn).UnderlyingConn().WriteMessage(websocket.BinaryMessage, buf[:7])
		assert.NoError(t, err)

		err = conn1.(*WebSocketConn).UnderlyingConn().WriteMessage(websocket.BinaryMessage, buf[7:])
		assert.NoError(t, err)

		in, err := conn1.Receive()
		assert.Nil(t, in)
		assert.Equal(t, io.EOF, err)
	})

	in, err := conn2.Receive()
	assert.Nil(t, err)
	assert.Equal(t, pkt.String(), in.String())

	err = conn2.Close()
	assert.NoError(t, err)

	safeReceive(done)
}

func TestWebSocketCoalescedMessage(t *testing.T) {
	pkt := packet.NewPublish()
	pkt.Message.Topic = "hello"
	pkt.Message.Payload = []byte("world")

	conn2, done := connectionPair("ws", func(conn1 Conn) {
		buf := make([]byte, pkt.Len()*2)

		_, err := pkt.Encode(buf)
		assert.NoError(t, err)

		_, err = pkt.Encode(buf[pkt.Len():])
		assert.NoError(t, err)

		err = conn1.(*WebSocketConn).UnderlyingConn().WriteMessage(websocket.BinaryMessage, buf)
		assert.NoError(t, err)

		in, err := conn1.Receive()
		assert.Nil(t, in)
		assert.Equal(t, io.EOF, err)
	})

	in, err := conn2.Receive()
	assert.Nil(t, err)
	assert.Equal(t, pkt.String(), in.String())

	in, err = conn2.Receive()
	assert.Nil(t, err)
	assert.Equal(t, pkt.String(), in.String())

	err = conn2.Close()
	assert.NoError(t, err)

	safeReceive(done)
}

func TestWebSocketNotBinaryMessage(t *testing.T) {
	pkt := packet.NewPublish()
	pkt.Message.Topic = "hello"
	pkt.Message.Payload = []byte("world")

	conn2, done := connectionPair("ws", func(conn1 Conn) {
		err := conn1.(*WebSocketConn).UnderlyingConn().WriteMessage(websocket.TextMessage, []byte("hello"))
		assert.NoError(t, err)
	})

	in, err := conn2.Receive()
	assert.Error(t, err)
	assert.Nil(t, in)

	safeReceive(done)
}

func BenchmarkWebSocketConn(b *testing.B) {
	pkt := packet.NewPublish()
	pkt.Message.Topic = "foo/bar/baz"

	conn2, done := connectionPair("ws", func(conn1 Conn) {
		for i := 0; i < b.N; i++ {
			err := conn1.Send(pkt, false)
			if err != nil {
				panic(err)
			}
		}
	})

	for i := 0; i < b.N; i++ {
		_, err := conn2.Receive()
		if err != nil {
			panic(err)
		}
	}

	b.SetBytes(int64(pkt.Len() * 2))

	safeReceive(done)
}

func BenchmarkWebSocketConnAsync(b *testing.B) {
	pkt := packet.NewPublish()
	pkt.Message.Topic = "foo/bar/baz"

	conn2, done := connectionPair("ws", func(conn1 Conn) {
		for i := 0; i < b.N; i++ {
			err := conn1.Send(pkt, true)
			if err != nil {
				panic(err)
			}
		}
	})

	for i := 0; i < b.N; i++ {
		_, err := conn2.Receive()
		if err != nil {
			panic(err)
		}
	}

	b.SetBytes(int64(pkt.Len() * 2))

	safeReceive(done)
}


<|start_filename|>packet/suback.go<|end_filename|>
package packet

import (
	"fmt"
	"strings"
)

// A Suback packet is sent by the server to the client to confirm receipt and
// processing of a Subscribe packet. The Suback packet contains a list of return
// codes, that specify the maximum QOS levels that have been granted.
type Suback struct {
	// The granted QOS levels for the requested subscriptions.
	ReturnCodes []QOS

	// The packet identifier.
	ID ID
}

// NewSuback creates a new Suback packet.
func NewSuback() *Suback {
	return &Suback{}
}

// Type returns the packets type.
func (s *Suback) Type() Type {
	return SUBACK
}

// String returns a string representation of the packet.
func (s *Suback) String() string {
	var codes []string

	for _, c := range s.ReturnCodes {
		codes = append(codes, fmt.Sprintf("%d", c))
	}

	return fmt.Sprintf("<Suback ID=%d ReturnCodes=[%s]>",
		s.ID, strings.Join(codes, ", "))
}

// Len returns the byte length of the encoded packet.
func (s *Suback) Len() int {
	ml := s.len()
	return headerLen(ml) + ml
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (s *Suback) Decode(src []byte) (int, error) {
	// decode header
	total, _, rl, err := decodeHeader(src, SUBACK)
	if err != nil {
		return total, err
	}

	// read packet id
	pid, n, err := readUint(src[total:], 2, SUBACK)
	total += n
	if err != nil {
		return total, err
	}

	// set packet id
	s.ID = ID(pid)
	if !s.ID.Valid() {
		return total, makeError(SUBACK, "packet id must be grater than zero")
	}

	// calculate number of return codes
	rcl := rl - 2
	if rcl < 1 {
		return total, makeError(SUBACK, "expected at least one return code")
	}

	// prepare return codes
	s.ReturnCodes = make([]QOS, 0, rcl)

	// read return codes
	for i := 0; i < rcl; i++ {
		// read return code
		rc, n, err := readUint8(src[total:], SUBACK)
		total += n
		if err != nil {
			return total, err
		}

		// get return code
		returnCode := QOS(rc)
		if !returnCode.Successful() && returnCode != QOSFailure {
			return total, makeError(SUBACK, "invalid return code %d", returnCode)
		}

		// add return code
		s.ReturnCodes = append(s.ReturnCodes, returnCode)
	}

	return total, nil
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (s *Suback) Encode(dst []byte) (int, error) {
	// encode header
	total, err := encodeHeader(dst, 0, s.len(), s.Len(), SUBACK)
	if err != nil {
		return total, err
	}

	// check packet id
	if !s.ID.Valid() {
		return 0, makeError(SUBACK, "packet id must be grater than zero")
	}

	// write packet id
	n, err := writeUint(dst[total:], uint64(s.ID), 2, SUBACK)
	total += n
	if err != nil {
		return total, err
	}

	// write return codes
	for _, rc := range s.ReturnCodes {
		// check return code
		if !rc.Successful() && rc != QOSFailure {
			return 0, makeError(SUBACK, "invalid return code %d", rc)
		}

		// write return code
		n, err := writeUint8(dst[total:], uint8(rc), SUBACK)
		total += n
		if err != nil {
			return total, err
		}
	}

	return total, nil
}

func (s *Suback) len() int {
	return 2 + len(s.ReturnCodes)
}


<|start_filename|>packet/header_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestDecodeHeaderError1(t *testing.T) {
	buf := []byte{0x6f, 193, 2} // < not enough bytes

	_, _, _, err := decodeHeader(buf, 0)
	assert.Error(t, err)
}

func TestDecodeHeaderError2(t *testing.T) {
	// source to small
	buf := []byte{0x62}

	_, _, _, err := decodeHeader(buf, 0)
	assert.Error(t, err)
}

func TestDecodeHeaderError3(t *testing.T) {
	buf := []byte{0x62, 0xff} // < invalid packet type

	_, _, _, err := decodeHeader(buf, 0)
	assert.Error(t, err)
}

func TestDecodeHeaderError4(t *testing.T) {
	// invalid remaining length
	buf := []byte{0x62, 0xff, 0xff, 0xff, 0xff}

	n, _, _, err := decodeHeader(buf, 6)

	assert.Error(t, err)
	assert.Equal(t, 1, n)
}

func TestDecodeHeaderError5(t *testing.T) {
	// remaining length to big
	buf := []byte{0x62, 0xff, 0xff, 0xff, 0xff, 0x1}

	n, _, _, err := decodeHeader(buf, 6)

	assert.Error(t, err)
	assert.Equal(t, 1, n)
}

func TestDecodeHeaderError6(t *testing.T) {
	// source to small
	buf := []byte{0x62, 0xff, 0x1}

	n, _, _, err := decodeHeader(buf, 6)

	assert.Error(t, err)
	assert.Equal(t, 3, n)
}

func TestDecodeHeaderError7(t *testing.T) {
	buf := []byte{0x66, 0x00, 0x01} // < wrong flags

	n, _, _, err := decodeHeader(buf, 6)
	assert.Error(t, err)
	assert.Equal(t, 1, n)
}

func TestEncodeHeader1(t *testing.T) {
	buf := make([]byte, 3)
	n, err := encodeHeader(buf, 0, 321, 3, PUBREL)

	assert.NoError(t, err)
	assert.Equal(t, 3, n)
	assert.Equal(t, []byte{0x62, 193, 2}, buf)
}

func TestEncodeHeader2(t *testing.T) {
	buf := make([]byte, 5)
	n, err := encodeHeader(buf, 0, maxVarint, 5, PUBREL)

	assert.NoError(t, err)
	assert.Equal(t, 5, n)
	assert.Equal(t, []byte{0x62, 0xff, 0xff, 0xff, 0x7f}, buf)
}

func TestEncodeHeaderError1(t *testing.T) {
	buf := make([]byte, 1) // < wrong buffer size
	n, err := encodeHeader(buf, 0, 0, 2, PUBREL)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
	assert.Equal(t, []byte{0x00}, buf)
}

func TestEncodeHeaderError2(t *testing.T) {
	buf := make([]byte, 2)
	// overload max remaining length
	n, err := encodeHeader(buf, 0, maxVarint+1, 2, PUBREL)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
	assert.Equal(t, []byte{0x62, 0x00}, buf)
}

func TestEncodeHeaderError3(t *testing.T) {
	buf := make([]byte, 2)
	// too small buffer
	n, err := encodeHeader(buf, 0, 2097151, 2, PUBREL)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
	assert.Equal(t, []byte{0x00, 0x00}, buf)
}


<|start_filename|>transport/dialer_test.go<|end_filename|>
package transport

import (
	"io"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestGlobalDial(t *testing.T) {
	server, err := testLauncher.Launch("tcp://localhost:0")
	require.NoError(t, err)

	wait := make(chan struct{})

	go func() {
		conn, err := server.Accept()
		require.NoError(t, err)

		pkt, err := conn.Receive()
		assert.Nil(t, pkt)
		assert.Equal(t, io.EOF, err)

		close(wait)
	}()

	conn, err := Dial(getURL(server, "tcp"))
	require.NoError(t, err)

	err = conn.Close()
	assert.NoError(t, err)

	safeReceive(wait)

	err = server.Close()
	assert.NoError(t, err)
}

func TestDialerBadURL(t *testing.T) {
	conn, err := Dial("foo")
	assert.Nil(t, conn)
	assert.Error(t, err)
}

func TestDialerUnsupportedProtocol(t *testing.T) {
	conn, err := Dial("foo://localhost")
	assert.Nil(t, conn)
	assert.Equal(t, ErrUnsupportedProtocol, err)
}

func TestDialerTCPError(t *testing.T) {
	conn, err := Dial("tcp://localhost:1234567")
	assert.Nil(t, conn)
	assert.Error(t, err)
}

func TestDialerTLSError(t *testing.T) {
	conn, err := Dial("tls://localhost:1234567")
	assert.Nil(t, conn)
	assert.Error(t, err)
}

func TestDialerWSError(t *testing.T) {
	conn, err := Dial("ws://localhost:1234567")
	assert.Nil(t, conn)
	assert.Error(t, err)
}

func TestDialerWSSError(t *testing.T) {
	conn, err := Dial("wss://localhost:1234567")
	assert.Nil(t, conn)
	assert.Error(t, err)
}

func abstractDefaultPortTest(t *testing.T, protocol string) {
	server, err := testLauncher.Launch(protocol + "://localhost:0")
	require.NoError(t, err)

	wait := make(chan struct{})

	go func() {
		conn, err := server.Accept()
		require.NoError(t, err)

		pkt, err := conn.Receive()
		assert.Nil(t, pkt)
		assert.Equal(t, io.EOF, err)

		close(wait)
	}()

	dialer := NewDialer(DialConfig{
		DefaultTCPPort: getPort(server),
		DefaultTLSPort: getPort(server),
		DefaultWSPort:  getPort(server),
		DefaultWSSPort: getPort(server),
	})

	conn, err := dialer.Dial(protocol + "://localhost")
	require.NoError(t, err)

	err = conn.Close()
	assert.NoError(t, err)

	safeReceive(wait)

	err = server.Close()
	assert.NoError(t, err)
}

func TestTCPDefaultPort(t *testing.T) {
	abstractDefaultPortTest(t, "tcp")
}

func TestTLSDefaultPort(t *testing.T) {
	abstractDefaultPortTest(t, "tls")
}

func TestWSDefaultPort(t *testing.T) {
	abstractDefaultPortTest(t, "ws")
}

func TestWSSDefaultPort(t *testing.T) {
	abstractDefaultPortTest(t, "wss")
}


<|start_filename|>packet/subscribe_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestSubscribeInterface(t *testing.T) {
	pkt := NewSubscribe()
	pkt.Subscriptions = []Subscription{
		{Topic: "foo", QOS: QOSAtMostOnce},
		{Topic: "bar", QOS: QOSAtLeastOnce},
	}

	assert.Equal(t, pkt.Type(), SUBSCRIBE)
	assert.Equal(t, "<Subscribe ID=0 Subscriptions=[\"foo\"=>0, \"bar\"=>1]>", pkt.String())
}

func TestSubscribeDecode(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		35,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // QOS
		0, // topic name MSB
		8, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c',
		1,  // QOS
		0,  // topic name MSB
		10, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c', 'd', 'd',
		2, // QOS
	}

	pkt := NewSubscribe()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, 3, len(pkt.Subscriptions))
	assert.Equal(t, "gomqtt", pkt.Subscriptions[0].Topic)
	assert.Equal(t, QOS(0), pkt.Subscriptions[0].QOS)
	assert.Equal(t, "/a/b/#/c", pkt.Subscriptions[1].Topic)
	assert.Equal(t, QOS(1), pkt.Subscriptions[1].QOS)
	assert.Equal(t, "/a/b/#/cdd", pkt.Subscriptions[2].Topic)
	assert.Equal(t, QOS(2), pkt.Subscriptions[2].QOS)
}

func TestSubscribeDecodeError1(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		9, // < too much
	}

	pkt := NewSubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubscribeDecodeError2(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		0,
		// < missing packet id
	}

	pkt := NewSubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubscribeDecodeError3(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		2,
		0, // packet ID MSB
		7, // packet ID LSB
		// < missing subscription
	}

	pkt := NewSubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubscribeDecodeError4(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		5,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		2, // topic name LSB < wrong size
		's',
	}

	pkt := NewSubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubscribeDecodeError5(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		5,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		1, // topic name LSB
		's',
		// < missing qos
	}

	pkt := NewSubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubscribeDecodeError6(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		5,
		0, // packet ID MSB
		0, // packet ID LSB < zero packet id
		0, // topic name MSB
		1, // topic name LSB
		's',
		0,
	}

	pkt := NewSubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubscribeDecodeError7(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		5,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		1, // topic name LSB
		's',
		0x81, // < invalid qos
	}

	pkt := NewSubscribe()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestSubscribeEncode(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		35,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // QOS
		0, // topic name MSB
		8, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c',
		1,  // QOS
		0,  // topic name MSB
		10, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c', 'd', 'd',
		2, // QOS
	}

	pkt := NewSubscribe()
	pkt.ID = 7
	pkt.Subscriptions = []Subscription{
		{Topic: "gomqtt", QOS: 0},
		{Topic: "/a/b/#/c", QOS: 1},
		{Topic: "/a/b/#/cdd", QOS: 2},
	}

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, packet, dst)
}

func TestSubscribeEncodeError1(t *testing.T) {
	pkt := NewSubscribe()
	pkt.ID = 7

	dst := make([]byte, 1) // < too small
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestSubscribeEncodeError2(t *testing.T) {
	pkt := NewSubscribe()
	pkt.ID = 7
	pkt.Subscriptions = []Subscription{
		{Topic: string(make([]byte, 65536)), QOS: 0}, // too big
	}

	dst := make([]byte, pkt.Len())
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestSubscribeEncodeError3(t *testing.T) {
	pkt := NewSubscribe()
	pkt.ID = 0 // < zero packet id

	dst := make([]byte, pkt.Len())
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestSubscribeEncodeError4(t *testing.T) {
	pkt := NewSubscribe()
	pkt.ID = 7
	pkt.Subscriptions = []Subscription{
		{Topic: string(make([]byte, 10)), QOS: 0x81}, // invalid qos
	}

	dst := make([]byte, pkt.Len())
	_, err := pkt.Encode(dst)

	assert.Error(t, err)
}

func TestSubscribeEqualDecodeEncode(t *testing.T) {
	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		35,
		0, // packet ID MSB
		7, // packet ID LSB
		0, // topic name MSB
		6, // topic name LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // QOS
		0, // topic name MSB
		8, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c',
		1,  // QOS
		0,  // topic name MSB
		10, // topic name LSB
		'/', 'a', '/', 'b', '/', '#', '/', 'c', 'd', 'd',
		2, // QOS
	}

	pkt := NewSubscribe()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)

	dst := make([]byte, pkt.Len())
	n2, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n2)
	assert.Equal(t, packet, dst[:n2])

	n3, err := pkt.Decode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n3)
}

func BenchmarkSubscribeEncode(b *testing.B) {
	b.ReportAllocs()

	pkt := NewSubscribe()
	pkt.ID = 7
	pkt.Subscriptions = []Subscription{
		{Topic: "t", QOS: 0},
	}

	buf := make([]byte, pkt.Len())

	for i := 0; i < b.N; i++ {
		_, err := pkt.Encode(buf)
		if err != nil {
			panic(err)
		}
	}
}

func BenchmarkSubscribeDecode(b *testing.B) {
	b.ReportAllocs()

	packet := []byte{
		byte(SUBSCRIBE<<4) | 2,
		6,
		0, // packet ID MSB
		1, // packet ID LSB
		0, // topic name MSB
		1, // topic name LSB
		't',
		0, // QOS
	}

	pkt := NewSubscribe()

	for i := 0; i < b.N; i++ {
		_, err := pkt.Decode(packet)
		if err != nil {
			panic(err)
		}
	}
}


<|start_filename|>cmd/gomqtt-stat/main.go<|end_filename|>
package main

import (
	"flag"
	"fmt"
	"sync/atomic"
	"time"

	"github.com/256dpi/gomqtt/client"
	"github.com/256dpi/gomqtt/packet"
)

var broker = flag.String("broker", "tcp://0.0.0.0:1883", "the broker url")
var filter = flag.String("filter", "#", "the filter subscription")

func main() {
	// parse flags
	flag.Parse()

	// print info
	fmt.Printf("Starting analisys of %s with filter %s...\n", *broker, *filter)

	// prepare counter
	var received int32

	// create client
	c := client.New()

	// set callback
	c.Callback = func(msg *packet.Message, err error) error {
		// check error
		if err != nil {
			panic(err)
		}

		// increment
		atomic.AddInt32(&received, 1)

		return nil
	}

	// connect
	cf, err := c.Connect(client.NewConfig(*broker))
	if err != nil {
		panic(err)
	}

	// await connack
	err = cf.Wait(time.Second)
	if err != nil {
		panic(err)
	}

	// subscribe
	sf, err := c.Subscribe(*filter, 0)
	if err != nil {
		panic(err)
	}

	// await suback
	err = sf.Wait(time.Second)
	if err != nil {
		panic(err)
	}

	// prepare counters
	var iterations int32
	var total int32

	for {
		// wait a second
		time.Sleep(1 * time.Second)

		// get and reset counter
		curReceived := atomic.LoadInt32(&received)
		atomic.StoreInt32(&received, 0)

		// increment total and iterations
		total += curReceived
		iterations++

		// print statistics
		fmt.Printf("Received: %d msgs ", curReceived)
		fmt.Printf("(Average Throughput: %d msg/s)\n", total/iterations)
	}
}


<|start_filename|>topic/topic_test.go<|end_filename|>
package topic

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestParse(t *testing.T) {
	tests := map[string]string{
		"topic/hello":         "topic/hello",
		"topic//hello":        "topic/hello",
		"topic///hello":       "topic/hello",
		"/topic":              "/topic",
		"//topic":             "/topic",
		"///topic":            "/topic",
		"topic/":              "topic",
		"topic//":             "topic",
		"topic///":            "topic",
		"topic///cool//hello": "topic/cool/hello",
		"topic//cool///hello": "topic/cool/hello",
	}

	for str, result := range tests {
		str, err := Parse(str, true)
		assert.Equal(t, result, str)
		assert.NoError(t, err, str)
	}
}

func TestParseZeroLengthError(t *testing.T) {
	_, err := Parse("", true)
	assert.Equal(t, ErrZeroLength, err)

	_, err = Parse("/", true)
	assert.Equal(t, ErrZeroLength, err)

	_, err = Parse("//", true)
	assert.Equal(t, ErrZeroLength, err)
}

func TestParseDisallowWildcards(t *testing.T) {
	tests := map[string]bool{
		"topic":            true,
		"topic/hello":      true,
		"topic/cool/hello": true,
		"+":                false,
		"#":                false,
		"topic/+":          false,
		"topic/#":          false,
	}

	for str, result := range tests {
		_, err := Parse(str, false)

		if result {
			assert.NoError(t, err, str)
		} else {
			assert.Error(t, err, str)
		}
	}
}

func TestParseAllowWildcards(t *testing.T) {
	tests := map[string]bool{
		"topic":            true,
		"topic/hello":      true,
		"topic/cool/hello": true,
		"+":                true,
		"#":                true,
		"topic/+":          true,
		"topic/#":          true,
		"topic/+/hello":    true,
		"topic/cool/+":     true,
		"topic/cool/#":     true,
		"+/cool/#":         true,
		"+/+/#":            true,
		"":                 false,
		"++":               false,
		"##":               false,
		"#/+":              false,
		"#/#":              false,
	}

	for str, result := range tests {
		_, err := Parse(str, true)

		if result {
			assert.NoError(t, err, str)
		} else {
			assert.Error(t, err, str)
		}
	}
}

func TestContainsWildcards(t *testing.T) {
	assert.True(t, ContainsWildcards("topic/+"))
	assert.True(t, ContainsWildcards("topic/#"))
	assert.False(t, ContainsWildcards("topic/hello"))
}

func BenchmarkParse(b *testing.B) {
	for i := 0; i < b.N; i++ {
		_, err := Parse("foo", true)
		if err != nil {
			panic(err)
		}
	}
}


<|start_filename|>transport/example_test.go<|end_filename|>
package transport

import (
	"fmt"

	"github.com/256dpi/gomqtt/packet"
)

func Example() {
	// launch server
	server, err := Launch("tcp://localhost:1337")
	if err != nil {
		panic(err)
	}

	go func() {
		// accept next incoming connection
		conn, err := server.Accept()
		if err != nil {
			panic(err)
		}

		// receive next packet
		pkt, err := conn.Receive()
		if err != nil {
			panic(err)
		}

		// check packet type
		if _, ok := pkt.(*packet.Connect); ok {
			// send a connack packet
			err = conn.Send(packet.NewConnack(), false)
			if err != nil {
				panic(err)
			}
		} else {
			panic("unexpected packet")
		}
	}()

	// dial to server
	conn, err := Dial("tcp://localhost:1337")
	if err != nil {
		panic(err)
	}

	// send connect packet
	err = conn.Send(packet.NewConnect(), false)
	if err != nil {
		panic(err)
	}

	// receive next packet
	pkt, err := conn.Receive()
	if err != nil {
		panic(err)
	}

	// check packet type
	if connackPacket, ok := pkt.(*packet.Connack); ok {
		fmt.Println(connackPacket)

		// close connection
		err = conn.Close()
		if err != nil {
			panic(err)
		}
	} else {
		panic("unexpected packet")
	}

	// close server
	err = server.Close()
	if err != nil {
		panic(err)
	}

	// Output:
	// <Connack SessionPresent=false ReturnCode=0>
}


<|start_filename|>spec/advanced.go<|end_filename|>
package spec

import (
	"testing"
	"time"

	"github.com/256dpi/gomqtt/client"
	"github.com/256dpi/gomqtt/packet"

	"github.com/stretchr/testify/assert"
)

// AuthenticationTest tests the broker for valid and invalid authentication.
func AuthenticationTest(t *testing.T, config *Config) {
	deniedClient := client.New()
	deniedClient.Callback = func(msg *packet.Message, err error) error {
		assert.Equal(t, client.ErrClientConnectionDenied, err)
		return nil
	}

	cf, err := deniedClient.Connect(client.NewConfig(config.DenyURL))
	assert.NoError(t, err)
	assert.Error(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.NotAuthorized, cf.ReturnCode())
	assert.False(t, cf.SessionPresent())

	allowedClient := client.New()

	cf, err = allowedClient.Connect(client.NewConfig(config.URL))
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.ConnectionAccepted, cf.ReturnCode())
	assert.False(t, cf.SessionPresent())

	err = allowedClient.Disconnect()
	assert.NoError(t, err)
}

// UniqueClientIDUncleanTest tests the broker for enforcing unique client ids.
func UniqueClientIDUncleanTest(t *testing.T, config *Config) {
	id := config.clientID()

	options := client.NewConfigWithClientID(config.URL, id)
	options.CleanSession = false

	assert.NoError(t, client.ClearSession(options, 10*time.Second))

	wait := make(chan struct{})

	firstClient := client.New()
	firstClient.Callback = func(msg *packet.Message, err error) error {
		close(wait)
		return nil
	}

	cf, err := firstClient.Connect(options)
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.ConnectionAccepted, cf.ReturnCode())
	assert.False(t, cf.SessionPresent())

	secondClient := client.New()

	cf, err = secondClient.Connect(options)
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.ConnectionAccepted, cf.ReturnCode())
	assert.True(t, cf.SessionPresent())

	safeReceive(wait)

	err = secondClient.Disconnect()
	assert.NoError(t, err)
}

// UniqueClientIDCleanTest tests the broker for enforcing unique client ids.
func UniqueClientIDCleanTest(t *testing.T, config *Config) {
	id := config.clientID()

	options := client.NewConfigWithClientID(config.URL, id)
	options.CleanSession = true

	wait := make(chan struct{})

	firstClient := client.New()
	firstClient.Callback = func(msg *packet.Message, err error) error {
		close(wait)
		return nil
	}

	cf, err := firstClient.Connect(options)
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.ConnectionAccepted, cf.ReturnCode())
	assert.False(t, cf.SessionPresent())

	secondClient := client.New()

	cf, err = secondClient.Connect(options)
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.ConnectionAccepted, cf.ReturnCode())
	assert.False(t, cf.SessionPresent())

	safeReceive(wait)

	err = secondClient.Disconnect()
	assert.NoError(t, err)
}

// RootSlashDistinctionTest tests the broker for supporting the root slash
// distinction.
func RootSlashDistinctionTest(t *testing.T, config *Config, topic string) {
	c := client.New()
	wait := make(chan struct{})

	c.Callback = func(msg *packet.Message, err error) error {
		assert.NoError(t, err)
		assert.Equal(t, topic, msg.Topic)
		assert.Equal(t, testPayload, msg.Payload)
		assert.Equal(t, packet.QOS(0), msg.QOS)
		assert.False(t, msg.Retain)

		close(wait)
		return nil
	}

	cf, err := c.Connect(client.NewConfig(config.URL))
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))
	assert.Equal(t, packet.ConnectionAccepted, cf.ReturnCode())
	assert.False(t, cf.SessionPresent())

	sf, err := c.Subscribe("/"+topic, 0)
	assert.NoError(t, err)
	assert.NoError(t, sf.Wait(10*time.Second))
	assert.Equal(t, []packet.QOS{0}, sf.ReturnCodes())

	sf, err = c.Subscribe(topic, 0)
	assert.NoError(t, err)
	assert.NoError(t, sf.Wait(10*time.Second))
	assert.Equal(t, []packet.QOS{0}, sf.ReturnCodes())

	pf, err := c.Publish(topic, testPayload, 0, false)
	assert.NoError(t, err)
	assert.NoError(t, pf.Wait(10*time.Second))

	safeReceive(wait)

	time.Sleep(config.NoMessageWait)

	err = c.Disconnect()
	assert.NoError(t, err)
}


<|start_filename|>spec/spec_test.go<|end_filename|>
package spec

import (
	"testing"
	"time"
)

func TestSpec(t *testing.T) {
	config := AllFeatures()
	config.URL = "tcp://localhost:1883"

	// mosquitto specific config
	config.Authentication = false
	config.ProcessWait = 10 * time.Millisecond
	config.MessageRetainWait = 100 * time.Millisecond
	config.NoMessageWait = 50 * time.Millisecond

	Run(t, config)
}


<|start_filename|>client/tracker.go<|end_filename|>
package client

import (
	"sync"
	"time"
)

// A Tracker keeps track of keep alive intervals.
type Tracker struct {
	last    time.Time
	pings   uint8
	timeout time.Duration
	mutex   sync.RWMutex
}

// NewTracker returns a new tracker.
func NewTracker(timeout time.Duration) *Tracker {
	return &Tracker{
		last:    time.Now(),
		timeout: timeout,
	}
}

// Reset will reset the tracker.
func (t *Tracker) Reset() {
	// acquire mutex
	t.mutex.Lock()
	defer t.mutex.Unlock()

	// reset timestamp
	t.last = time.Now()
}

// Window returns the time until a new ping should be sent.
func (t *Tracker) Window() time.Duration {
	// acquire mutex
	t.mutex.RLock()
	defer t.mutex.RUnlock()

	return t.timeout - time.Since(t.last)
}

// Ping marks a ping.
func (t *Tracker) Ping() {
	// acquire mutex
	t.mutex.Lock()
	defer t.mutex.Unlock()

	// increment
	t.pings++
}

// Pong marks a pong.
func (t *Tracker) Pong() {
	// acquire mutex
	t.mutex.Lock()
	defer t.mutex.Unlock()

	// decrement
	t.pings--
}

// Pending returns if pings are pending.
func (t *Tracker) Pending() bool {
	// acquire mutex
	t.mutex.RLock()
	defer t.mutex.RUnlock()

	return t.pings > 0
}


<|start_filename|>cmd/gomqtt-membroker/main.go<|end_filename|>
package main

import (
	"flag"
	"fmt"
	"net/http"
	_ "net/http/pprof"
	"os"
	"os/signal"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/256dpi/gomqtt/broker"
	"github.com/256dpi/gomqtt/packet"
	"github.com/256dpi/gomqtt/transport"
)

var url = flag.String("url", "tcp://0.0.0.0:1883", "broker url")
var queue = flag.Int("queue", 100, "queue size")
var publish = flag.Int("publish", 100, "parallel publishes")
var inflight = flag.Int("inflight", 100, "inflight messages")
var timeout = flag.Int("timeout", 1, "token timeout")

func main() {
	// parse flags
	flag.Parse()

	// run pprof interface
	go func() {
		panic(http.ListenAndServe("localhost:6060", nil))
	}()

	// print info
	fmt.Printf("Starting broker on URL %s...\n", *url)

	// launch server
	server, err := transport.Launch(*url)
	if err != nil {
		panic(err)
	}

	// prepare backend
	backend := broker.NewMemoryBackend()
	backend.SessionQueueSize = *queue
	backend.ClientParallelPublishes = *publish
	backend.ClientParallelSubscribes = *inflight
	backend.ClientInflightMessages = *inflight
	backend.ClientTokenTimeout = time.Duration(*timeout) * time.Second

	// prepare counters
	var published int32
	var forwarded int32
	var clients int32

	// configure logger
	backend.Logger = func(event broker.LogEvent, client *broker.Client, pkt packet.Generic, msg *packet.Message, err error) {
		if event == broker.NewConnection {
			atomic.AddInt32(&clients, 1)
		} else if event == broker.MessagePublished {
			atomic.AddInt32(&published, 1)
		} else if event == broker.MessageForwarded {
			atomic.AddInt32(&forwarded, 1)
		} else if event == broker.LostConnection {
			atomic.AddInt32(&clients, -1)
		}
	}

	// prepare engine
	engine := broker.NewEngine(backend)

	// set error callback
	engine.OnError = func(err error) {
		println(err.Error())
	}

	// accept from server
	engine.Accept(server)

	// run reporter
	go func() {
		for {
			// wait a second
			time.Sleep(time.Second)

			// read counters
			pub := atomic.LoadInt32(&published)
			fwd := atomic.LoadInt32(&forwarded)

			// print statistics
			fmt.Printf("Publish Rate: %d msg/s, Forward Rate: %d msg/s, Clients: %d\n", pub, fwd, clients)

			// reset counters
			atomic.StoreInt32(&published, 0)
			atomic.StoreInt32(&forwarded, 0)
		}
	}()

	// await finish signal
	finish := make(chan os.Signal, 1)
	signal.Notify(finish, syscall.SIGINT, syscall.SIGTERM)
	<-finish

	// close backend
	backend.Close(5 * time.Second)

	// close server
	_ = server.Close()

	// close engine
	engine.Close()
}


<|start_filename|>packet/messages_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestMessageString(t *testing.T) {
	msg := &Message{
		Topic:   "w",
		Payload: []byte("m"),
		QOS:     QOSAtLeastOnce,
	}

	assert.Equal(t, "<Message Topic=\"w\" QOS=1 Retain=false Payload=6d>", msg.String())
}

func TestMessageCopy(t *testing.T) {
	msg1 := &Message{
		Topic:   "w",
		Payload: []byte("m"),
		QOS:     QOSAtLeastOnce,
	}

	msg2 := msg1.Copy()
	assert.Equal(t, msg1.String(), msg2.String())

	msg1.Retain = true
	assert.False(t, msg2.Retain)
}


<|start_filename|>session/memory_session_test.go<|end_filename|>
package session

import (
	"math"
	"testing"

	"github.com/256dpi/gomqtt/packet"

	"github.com/stretchr/testify/assert"
)

func TestMemorySessionNextID(t *testing.T) {
	session := NewMemorySession()

	assert.Equal(t, packet.ID(1), session.NextID())
	assert.Equal(t, packet.ID(2), session.NextID())

	for i := 0; i < math.MaxUint16-3; i++ {
		session.NextID()
	}

	assert.Equal(t, packet.ID(math.MaxUint16), session.NextID())
	assert.Equal(t, packet.ID(1), session.NextID())

	err := session.Reset()
	assert.NoError(t, err)

	assert.Equal(t, packet.ID(1), session.NextID())
}

func TestMemorySessionPacketStore(t *testing.T) {
	session := NewMemorySession()

	publish := packet.NewPublish()
	publish.ID = 1

	pkt, err := session.LookupPacket(Incoming, 1)
	assert.NoError(t, err)
	assert.Nil(t, pkt)

	err = session.SavePacket(Incoming, publish)
	assert.NoError(t, err)

	pkt, err = session.LookupPacket(Incoming, 1)
	assert.NoError(t, err)
	assert.Equal(t, publish, pkt)

	list, err := session.AllPackets(Incoming)
	assert.NoError(t, err)
	assert.Equal(t, 1, len(list))

	err = session.DeletePacket(Incoming, 1)
	assert.NoError(t, err)

	pkt, err = session.LookupPacket(Incoming, 1)
	assert.NoError(t, err)
	assert.Nil(t, pkt)

	list, err = session.AllPackets(Incoming)
	assert.NoError(t, err)
	assert.Equal(t, 0, len(list))

	err = session.SavePacket(Outgoing, publish)
	assert.NoError(t, err)

	list, err = session.AllPackets(Outgoing)
	assert.NoError(t, err)
	assert.Equal(t, 1, len(list))

	err = session.Reset()
	assert.NoError(t, err)

	list, err = session.AllPackets(Outgoing)
	assert.NoError(t, err)
	assert.Equal(t, 0, len(list))
}


<|start_filename|>packet/unsubscribe.go<|end_filename|>
package packet

import (
	"fmt"
	"strings"
)

// An Unsubscribe packet is sent by the client to the server.
type Unsubscribe struct {
	// The topics to unsubscribe from.
	Topics []string

	// The packet identifier.
	ID ID
}

// NewUnsubscribe creates a new Unsubscribe packet.
func NewUnsubscribe() *Unsubscribe {
	return &Unsubscribe{}
}

// Type returns the packets type.
func (u *Unsubscribe) Type() Type {
	return UNSUBSCRIBE
}

// String returns a string representation of the packet.
func (u *Unsubscribe) String() string {
	// collect topics
	var topics []string
	for _, t := range u.Topics {
		topics = append(topics, fmt.Sprintf("%q", t))
	}

	return fmt.Sprintf("<Unsubscribe Topics=[%s]>", strings.Join(topics, ", "))
}

// Len returns the byte length of the encoded packet.
func (u *Unsubscribe) Len() int {
	ml := u.len()
	return headerLen(ml) + ml
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (u *Unsubscribe) Decode(src []byte) (int, error) {
	// decode header
	total, _, rl, err := decodeHeader(src, UNSUBSCRIBE)
	if err != nil {
		return total, err
	}

	// read packet id
	pid, n, err := readUint(src[total:], 2, UNSUBSCRIBE)
	total += n
	if err != nil {
		return total, err
	}

	// set packet id
	u.ID = ID(pid)
	if !u.ID.Valid() {
		return total, makeError(UNSUBSCRIBE, "packet id must be grater than zero")
	}

	// reset topics
	u.Topics = u.Topics[:0]

	// read topics
	tl := rl - 2
	for tl > 0 {
		// read topic
		topic, n, err := readLPString(src[total:], UNSUBSCRIBE)
		total += n
		if err != nil {
			return total, err
		}

		// append to list
		u.Topics = append(u.Topics, topic)

		// decrement counter
		tl -= n
	}

	// check for empty list
	if len(u.Topics) == 0 {
		return total, makeError(UNSUBSCRIBE, "empty topic list")
	}

	return total, nil
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (u *Unsubscribe) Encode(dst []byte) (int, error) {
	// encode header
	total, err := encodeHeader(dst, 0, u.len(), u.Len(), UNSUBSCRIBE)
	if err != nil {
		return total, err
	}

	// check packet id
	if !u.ID.Valid() {
		return 0, makeError(UNSUBSCRIBE, "packet id must be grater than zero")
	}

	// write packet id
	n, err := writeUint(dst[total:], uint64(u.ID), 2, UNSUBSCRIBE)
	total += n
	if err != nil {
		return total, err
	}

	// write topics
	for _, topic := range u.Topics {
		// write topic
		n, err := writeLPString(dst[total:], topic, UNSUBSCRIBE)
		total += n
		if err != nil {
			return total, err
		}
	}

	return total, nil
}

func (u *Unsubscribe) len() int {
	// packet ID
	total := 2

	// add topics
	for _, t := range u.Topics {
		total += 2 + len(t)
	}

	return total
}


<|start_filename|>packet/identified_test.go<|end_filename|>
package packet

import (
	"fmt"
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestIdentifiedDecode(t *testing.T) {
	packet := []byte{
		byte(PUBACK << 4),
		2,
		0, // packet ID MSB
		7, // packet ID LSB
	}

	var pid ID
	n, err := identifiedDecode(packet, &pid, PUBACK)
	assert.NoError(t, err)
	assert.Equal(t, 4, n)
	assert.Equal(t, ID(7), pid)
}

func TestIdentifiedDecodeError1(t *testing.T) {
	packet := []byte{
		byte(PUBACK << 4),
		1, // < wrong remaining length
		0, // packet ID MSB
		7, // packet ID LSB
	}

	var pid ID
	n, err := identifiedDecode(packet, &pid, PUBACK)
	assert.Error(t, err)
	assert.Equal(t, 2, n)
	assert.Equal(t, ID(0), pid)
}

func TestIdentifiedDecodeError2(t *testing.T) {
	packet := []byte{
		byte(PUBACK << 4),
		2,
		7, // packet ID LSB
		// < insufficient bytes
	}

	var pid ID
	n, err := identifiedDecode(packet, &pid, PUBACK)
	assert.Error(t, err)
	assert.Equal(t, 2, n)
	assert.Equal(t, ID(0), pid)
}

func TestIdentifiedDecodeError3(t *testing.T) {
	packet := []byte{
		byte(PUBACK << 4),
		2,
		0, // packet ID LSB
		0, // packet ID MSB < zero id
	}

	var pid ID
	n, err := identifiedDecode(packet, &pid, PUBACK)
	assert.Error(t, err)
	assert.Equal(t, 4, n)
	assert.Equal(t, ID(0), pid)
}

func TestIdentifiedEncode(t *testing.T) {
	packet := []byte{
		byte(PUBACK << 4),
		2,
		0, // packet ID MSB
		7, // packet ID LSB
	}

	dst := make([]byte, identifiedLen())
	n, err := identifiedEncode(dst, 7, PUBACK)

	assert.NoError(t, err)
	assert.Equal(t, 4, n)
	assert.Equal(t, packet, dst[:n])
}

func TestIdentifiedEncodeError1(t *testing.T) {
	dst := make([]byte, 3) // < insufficient buffer
	n, err := identifiedEncode(dst, 7, PUBACK)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
}

func TestIdentifiedEncodeError2(t *testing.T) {
	dst := make([]byte, identifiedLen())
	n, err := identifiedEncode(dst, 0, PUBACK) // < zero id

	assert.Error(t, err)
	assert.Equal(t, 0, n)
}

func TestIdentifiedEqualDecodeEncode(t *testing.T) {
	packet := []byte{
		byte(PUBACK << 4),
		2,
		0, // packet ID MSB
		7, // packet ID LSB
	}

	pkt := &Puback{}
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, 4, n)

	dst := make([]byte, 100)
	n2, err := identifiedEncode(dst, 7, PUBACK)

	assert.NoError(t, err)
	assert.Equal(t, 4, n2)
	assert.Equal(t, packet, dst[:n2])

	var pid ID
	n3, err := identifiedDecode(packet, &pid, PUBACK)
	assert.NoError(t, err)
	assert.Equal(t, 4, n3)
	assert.Equal(t, ID(7), pid)
}

func BenchmarkIdentifiedEncode(b *testing.B) {
	b.ReportAllocs()

	pkt := &Puback{}
	pkt.ID = 1

	buf := make([]byte, pkt.Len())

	for i := 0; i < b.N; i++ {
		_, err := pkt.Encode(buf)
		if err != nil {
			panic(err)
		}
	}
}

func BenchmarkIdentifiedDecode(b *testing.B) {
	b.ReportAllocs()

	packet := []byte{
		byte(PUBACK << 4),
		2,
		0, // packet ID MSB
		1, // packet ID LSB
	}

	pkt := &Puback{}

	for i := 0; i < b.N; i++ {
		_, err := pkt.Decode(packet)
		if err != nil {
			panic(err)
		}
	}
}

func testIdentifiedImplementation(t *testing.T, pkt Generic) {
	assert.Equal(t, fmt.Sprintf("<%s ID=1>", pkt.Type().String()), pkt.String())

	buf := make([]byte, pkt.Len())
	n, err := pkt.Encode(buf)
	assert.NoError(t, err)
	assert.Equal(t, 4, n)

	n, err = pkt.Decode(buf)
	assert.NoError(t, err)
	assert.Equal(t, 4, n)
}

func TestPubackImplementation(t *testing.T) {
	pkt := NewPuback()
	pkt.ID = 1

	testIdentifiedImplementation(t, pkt)
}

func TestPubcompImplementation(t *testing.T) {
	pkt := NewPubcomp()
	pkt.ID = 1

	testIdentifiedImplementation(t, pkt)
}

func TestPubrecImplementation(t *testing.T) {
	pkt := NewPubrec()
	pkt.ID = 1

	testIdentifiedImplementation(t, pkt)
}

func TestPubrelImplementation(t *testing.T) {
	pkt := NewPubrel()
	pkt.ID = 1

	testIdentifiedImplementation(t, pkt)
}

func TestUnsubackImplementation(t *testing.T) {
	pkt := NewUnsuback()
	pkt.ID = 1

	testIdentifiedImplementation(t, pkt)
}


<|start_filename|>broker/client_test.go<|end_filename|>
package broker

import (
	"testing"
	"time"

	"github.com/256dpi/gomqtt/client"
	"github.com/256dpi/gomqtt/packet"
	"github.com/256dpi/gomqtt/transport"
	"github.com/256dpi/gomqtt/transport/flow"

	"github.com/stretchr/testify/assert"
)

type testMemoryBackend struct {
	MemoryBackend

	packets []packet.Generic
}

func (b *testMemoryBackend) Setup(client *Client, id string, clean bool) (Session, bool, error) {
	client.PacketCallback = func(pkt packet.Generic) error {
		b.packets = append(b.packets, pkt)
		return nil
	}

	return b.MemoryBackend.Setup(client, id, clean)
}

func TestClientMaximumKeepAlive(t *testing.T) {
	backend := &testMemoryBackend{
		MemoryBackend: *NewMemoryBackend(),
	}

	backend.MemoryBackend.ClientMaximumKeepAlive = 10 * time.Millisecond

	port, quit, done := Run(NewEngine(backend), "tcp")

	conn, err := transport.Dial("tcp://localhost:" + port)
	assert.NoError(t, err)

	f := flow.New().
		Send(packet.NewConnect()).
		Receive(packet.NewConnack()).
		End()

	err = f.Test(conn)
	assert.NoError(t, err)

	ret := backend.Close(5 * time.Second)
	assert.True(t, ret)

	close(quit)

	safeReceive(done)
}

func TestClientPacketCallback(t *testing.T) {
	backend := &testMemoryBackend{
		MemoryBackend: *NewMemoryBackend(),
	}

	port, quit, done := Run(NewEngine(backend), "tcp")

	options := client.NewConfig("tcp://localhost:" + port)

	client1 := client.New()

	cf, err := client1.Connect(options)
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))

	sf, err := client1.Subscribe("pc", 0)
	assert.NoError(t, err)
	assert.NoError(t, sf.Wait(10*time.Second))

	pf, err := client1.Publish("pc", nil, 1, false)
	assert.NoError(t, err)
	assert.NoError(t, pf.Wait(10*time.Second))

	err = client1.Disconnect()
	assert.NoError(t, err)

	ret := backend.Close(5 * time.Second)
	assert.True(t, ret)

	close(quit)

	safeReceive(done)

	assert.Len(t, backend.packets, 2)
	assert.Equal(t, packet.SUBSCRIBE, backend.packets[0].Type())
	assert.Equal(t, packet.PUBLISH, backend.packets[1].Type())
}

func TestClientTokenTimeoutPublish(t *testing.T) {
	backend := &testMemoryBackend{
		MemoryBackend: *NewMemoryBackend(),
	}

	backend.MemoryBackend.ClientParallelPublishes = 1
	backend.MemoryBackend.ClientTokenTimeout = 10 * time.Millisecond

	port, quit, done := Run(NewEngine(backend), "tcp")

	conn, err := transport.Dial("tcp://localhost:" + port)
	assert.NoError(t, err)

	f := flow.New().
		Send(packet.NewConnect()).
		Receive(packet.NewConnack()).
		Send(&packet.Publish{Message: packet.Message{Topic: "ttp", QOS: 2}, ID: 1}).
		Receive(&packet.Pubrec{ID: 1}).
		Send(&packet.Publish{Message: packet.Message{Topic: "ttp", QOS: 2}, ID: 2}).
		End()

	err = f.Test(conn)
	assert.NoError(t, err)

	ret := backend.Close(5 * time.Second)
	assert.True(t, ret)

	close(quit)

	safeReceive(done)
}

func TestClientTokenTimeoutDequeue(t *testing.T) {
	backend := &testMemoryBackend{
		MemoryBackend: *NewMemoryBackend(),
	}

	backend.MemoryBackend.ClientInflightMessages = 1
	backend.MemoryBackend.ClientTokenTimeout = 10 * time.Millisecond

	port, quit, done := Run(NewEngine(backend), "tcp")

	options := client.NewConfig("tcp://localhost:" + port)

	client1 := client.New()

	cf, err := client1.Connect(options)
	assert.NoError(t, err)
	assert.NoError(t, cf.Wait(10*time.Second))

	conn, err := transport.Dial("tcp://localhost:" + port)
	assert.NoError(t, err)

	f := flow.New().
		Send(packet.NewConnect()).
		Receive(packet.NewConnack()).
		Send(&packet.Subscribe{Subscriptions: []packet.Subscription{{Topic: "ttd", QOS: 2}}, ID: 1}).
		Receive(&packet.Suback{ID: 1, ReturnCodes: []packet.QOS{2}}).
		Run(func() {
			pf, err := client1.Publish("ttd", nil, 2, false)
			assert.NoError(t, err)
			assert.NoError(t, pf.Wait(10*time.Second))

			pf, err = client1.Publish("ttd", nil, 2, false)
			assert.NoError(t, err)
			assert.NoError(t, pf.Wait(10*time.Second))
		}).
		Receive(&packet.Publish{Message: packet.Message{Topic: "ttd", QOS: 2}, ID: 1}).
		End()

	err = f.Test(conn)
	assert.NoError(t, err)

	err = client1.Disconnect()
	assert.NoError(t, err)

	ret := backend.Close(5 * time.Second)
	assert.True(t, ret)

	close(quit)

	safeReceive(done)
}


<|start_filename|>packet/connect_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestConnectInterface(t *testing.T) {
	pkt := NewConnect()
	pkt.Will = &Message{
		Topic:   "w",
		Payload: []byte("m"),
		QOS:     QOSAtLeastOnce,
	}

	assert.Equal(t, pkt.Type(), CONNECT)
	assert.Equal(t, "<Connect ClientID=\"\" KeepAlive=0 Username=\"\" Password=\"\" CleanSession=true Will=<Message Topic=\"w\" QOS=1 Retain=false Payload=6d> Version=4>", pkt.String())
}

func TestConnectDecode1(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		58,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,   // Protocol Level
		206, // Connect Flags
		0,   // Keep Alive MSB
		10,  // Keep Alive LSB
		0,   // Client ID MSB
		6,   // Client ID LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // Will Topic MSB
		4, // Will Topic LSB
		'w', 'i', 'l', 'l',
		0,  // Will Message MSB
		12, // Will Message LSB
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
		0, // Username ID MSB
		6, // Username ID LSB
		'g', 'o', 'm', 'q', 't', 't',
		0,  // Password ID MSB
		10, // Password ID LSB
		'v', 'e', 'r', 'y', 's', 'e', 'c', 'r', 'e', 't',
	}

	pkt := NewConnect()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, uint16(10), pkt.KeepAlive)
	assert.Equal(t, "gomqtt", pkt.ClientID)
	assert.Equal(t, "will", pkt.Will.Topic)
	assert.Equal(t, []byte("send me home"), pkt.Will.Payload)
	assert.Equal(t, "gomqtt", pkt.Username)
	assert.Equal(t, "<PASSWORD>", pkt.Password)
	assert.Equal(t, Version311, pkt.Version)
}

func TestConnectDecode2(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		58,
		0, // Protocol String MSB
		6, // Protocol String LSB
		'M', 'Q', 'I', 's', 'd', 'p',
		3,   // Protocol Level
		206, // Connect Flags
		0,   // Keep Alive MSB
		10,  // Keep Alive LSB
		0,   // Client ID MSB
		6,   // Client ID LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // Will Topic MSB
		4, // Will Topic LSB
		'w', 'i', 'l', 'l',
		0,  // Will Message MSB
		12, // Will Message LSB
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
		0, // Username ID MSB
		6, // Username ID LSB
		'g', 'o', 'm', 'q', 't', 't',
		0,  // Password ID MSB
		10, // Password ID LSB
		'v', 'e', 'r', 'y', 's', 'e', 'c', 'r', 'e', 't',
	}

	pkt := NewConnect()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, uint16(10), pkt.KeepAlive)
	assert.Equal(t, "gomqtt", pkt.ClientID)
	assert.Equal(t, "will", pkt.Will.Topic)
	assert.Equal(t, []byte("send me home"), pkt.Will.Payload)
	assert.Equal(t, "gomqtt", pkt.Username)
	assert.Equal(t, "<PASSWORD>", pkt.Password)
	assert.Equal(t, Version31, pkt.Version)
}

func TestConnectDecodeError1(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		60, // < wrong remaining length
		0,  // Protocol String MSB
		5,  // Protocol String LSB
		'M', 'Q', 'T', 'T',
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError2(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		6,
		0, // Protocol String MSB
		5, // Protocol String LSB < invalid size
		'M', 'Q', 'T', 'T',
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError3(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		6,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		// Protocol Level < missing
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError4(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		8,
		0,                       // Protocol String MSB
		5,                       // Protocol String LSB
		'M', 'Q', 'T', 'T', 'X', // < wrong protocol string
		4,
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError5(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		7,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		5, // Protocol Level < wrong id
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError6(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		7,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4, // Protocol Level
		// Connect Flags < missing
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError7(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		7,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4, // Protocol Level
		1, // Connect Flags < reserved bit set to one
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError8(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		7,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,  // Protocol Level
		24, // Connect Flags < invalid qos
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError9(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		7,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4, // Protocol Level
		8, // Connect Flags < will flag set to zero but others not
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError10(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		7,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,  // Protocol Level
		64, // Connect Flags < password flag set but not username
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError11(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		7,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4, // Protocol Level
		0, // Connect Flags
		0, // Keep Alive MSB < missing
		// Keep Alive LSB < missing
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError12(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		7,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4, // Protocol Level
		0, // Connect Flags
		0, // Keep Alive MSB
		1, // Keep Alive LSB
		0, // Client ID MSB
		2, // Client ID LSB < wrong size
		'x',
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError13(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		6,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4, // Protocol Level
		0, // Connect Flags < clean session false
		0, // Keep Alive MSB
		1, // Keep Alive LSB
		0, // Client ID MSB
		0, // Client ID LSB
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError14(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		6,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4, // Protocol Level
		6, // Connect Flags
		0, // Keep Alive MSB
		1, // Keep Alive LSB
		0, // Client ID MSB
		0, // Client ID LSB
		0, // Will Topic MSB
		1, // Will Topic LSB < wrong size
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError15(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		6,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4, // Protocol Level
		6, // Connect Flags
		0, // Keep Alive MSB
		1, // Keep Alive LSB
		0, // Client ID MSB
		0, // Client ID LSB
		0, // Will Topic MSB
		0, // Will Topic LSB
		0, // Will Payload MSB
		1, // Will Payload LSB < wrong size
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError16(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		6,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,   // Protocol Level
		194, // Connect Flags
		0,   // Keep Alive MSB
		1,   // Keep Alive LSB
		0,   // Client ID MSB
		0,   // Client ID LSB
		0,   // Username MSB
		1,   // Username LSB < wrong size
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectDecodeError17(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		6,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,   // Protocol Level
		194, // Connect Flags
		0,   // Keep Alive MSB
		1,   // Keep Alive LSB
		0,   // Client ID MSB
		0,   // Client ID LSB
		0,   // Username MSB
		0,   // Username LSB
		0,   // Password MSB
		1,   // Password LSB < wrong size
	}

	pkt := NewConnect()
	_, err := pkt.Decode(packet)

	assert.Error(t, err)
}

func TestConnectEncode1(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		58,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,   // Protocol level 4
		204, // Connect Flags
		0,   // Keep Alive MSB
		10,  // Keep Alive LSB
		0,   // Client ID MSB
		6,   // Client ID LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // Will Topic MSB
		4, // Will Topic LSB
		'w', 'i', 'l', 'l',
		0,  // Will Message MSB
		12, // Will Message LSB
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
		0, // Username ID MSB
		6, // Username ID LSB
		'g', 'o', 'm', 'q', 't', 't',
		0,  // Password ID MSB
		10, // Password ID LSB
		'v', 'e', 'r', 'y', 's', 'e', 'c', 'r', 'e', 't',
	}

	pkt := NewConnect()
	pkt.Will = &Message{
		QOS:     QOSAtLeastOnce,
		Topic:   "will",
		Payload: []byte("send me home"),
	}
	pkt.CleanSession = false
	pkt.ClientID = "gomqtt"
	pkt.KeepAlive = 10
	pkt.Username = "gomqtt"
	pkt.Password = "<PASSWORD>"

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, packet, dst[:n])
}

func TestConnectEncode2(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		14,
		0, // Protocol String MSB
		6, // Protocol String LSB
		'M', 'Q', 'I', 's', 'd', 'p',
		3,  // Protocol level 4
		2,  // Connect Flags
		0,  // Keep Alive MSB
		10, // Keep Alive LSB
		0,  // Client ID MSB
		0,  // Client ID LSB
	}

	pkt := NewConnect()
	pkt.CleanSession = true
	pkt.KeepAlive = 10
	pkt.Version = Version31

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, packet, dst[:n])
}

func TestConnectEncode3(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		12,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,  // Protocol level 4
		2,  // Connect Flags
		0,  // Keep Alive MSB
		10, // Keep Alive LSB
		0,  // Client ID MSB
		0,  // Client ID LSB
	}

	pkt := NewConnect()
	pkt.CleanSession = true
	pkt.KeepAlive = 10
	pkt.Version = 0

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)
	assert.Equal(t, packet, dst[:n])
}

func TestConnectEncodeError1(t *testing.T) {
	pkt := NewConnect()

	dst := make([]byte, 4) // < too small buffer
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 0, n)
}

func TestConnectEncodeError2(t *testing.T) {
	pkt := NewConnect()
	pkt.Will = &Message{
		Topic: "t",
		QOS:   3, // < wrong qos
	}

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 9, n)
}

func TestConnectEncodeError3(t *testing.T) {
	pkt := NewConnect()
	pkt.ClientID = string(make([]byte, 65536)) // < too big

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 14, n)
}

func TestConnectEncodeError4(t *testing.T) {
	pkt := NewConnect()
	pkt.Will = &Message{
		Topic: string(make([]byte, 65536)), // < too big
	}

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 16, n)
}

func TestConnectEncodeError5(t *testing.T) {
	pkt := NewConnect()
	pkt.Will = &Message{
		Topic:   "t",
		Payload: make([]byte, 65536), // < too big
	}

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 19, n)
}

func TestConnectEncodeError6(t *testing.T) {
	pkt := NewConnect()
	pkt.Username = string(make([]byte, 65536)) // < too big

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 16, n)
}

func TestConnectEncodeError7(t *testing.T) {
	pkt := NewConnect()
	pkt.Password = "p" // < missing username

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 14, n)
}

func TestConnectEncodeError8(t *testing.T) {
	pkt := NewConnect()
	pkt.Username = "u"
	pkt.Password = string(make([]byte, 65536)) // < too big

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 19, n)
}

func TestConnectEncodeError9(t *testing.T) {
	pkt := NewConnect()
	pkt.Will = &Message{
		// < missing topic
	}

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 9, n)
}

func TestConnectEncodeError10(t *testing.T) {
	pkt := NewConnect()
	pkt.Version = 255

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 2, n)
}

func TestConnectEncodeError11(t *testing.T) {
	pkt := NewConnect()
	pkt.CleanSession = false // < client id is empty

	dst := make([]byte, pkt.Len())
	n, err := pkt.Encode(dst)

	assert.Error(t, err)
	assert.Equal(t, 9, n)
}

func TestConnectEqualDecodeEncode(t *testing.T) {
	packet := []byte{
		byte(CONNECT << 4),
		58,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,   // Protocol level 4
		238, // Connect Flags
		0,   // Keep Alive MSB
		10,  // Keep Alive LSB
		0,   // Client ID MSB
		6,   // Client ID LSB
		'g', 'o', 'm', 'q', 't', 't',
		0, // Will Topic MSB
		4, // Will Topic LSB
		'w', 'i', 'l', 'l',
		0,  // Will Message MSB
		12, // Will Message LSB
		's', 'e', 'n', 'd', ' ', 'm', 'e', ' ', 'h', 'o', 'm', 'e',
		0, // Username ID MSB
		6, // Username ID LSB
		'g', 'o', 'm', 'q', 't', 't',
		0,  // Password ID MSB
		10, // Password ID LSB
		'v', 'e', 'r', 'y', 's', 'e', 'c', 'r', 'e', 't',
	}

	pkt := NewConnect()
	n, err := pkt.Decode(packet)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n)

	dst := make([]byte, pkt.Len())
	n2, err := pkt.Encode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n2)
	assert.Equal(t, packet, dst[:n2])

	n3, err := pkt.Decode(dst)

	assert.NoError(t, err)
	assert.Equal(t, len(packet), n3)
}

func BenchmarkConnectEncode(b *testing.B) {
	b.ReportAllocs()

	pkt := NewConnect()
	pkt.Will = &Message{
		Topic:   "w",
		Payload: []byte("m"),
		QOS:     QOSAtLeastOnce,
	}
	pkt.CleanSession = true
	pkt.ClientID = "i"
	pkt.KeepAlive = 10
	pkt.Username = "u"
	pkt.Password = "p"

	buf := make([]byte, pkt.Len())

	for i := 0; i < b.N; i++ {
		_, err := pkt.Encode(buf)
		if err != nil {
			panic(err)
		}
	}
}

func BenchmarkConnectDecode(b *testing.B) {
	b.ReportAllocs()

	packet := []byte{
		byte(CONNECT << 4),
		25,
		0, // Protocol String MSB
		4, // Protocol String LSB
		'M', 'Q', 'T', 'T',
		4,   // Protocol level 4
		206, // Connect Flags
		0,   // Keep Alive MSB
		10,  // Keep Alive LSB
		0,   // Client ID MSB
		1,   // Client ID LSB
		'i',
		0, // Will Topic MSB
		1, // Will Topic LSB
		'w',
		0, // Will Message MSB
		1, // Will Message LSB
		'm',
		0, // Username ID MSB
		1, // Username ID LSB
		'u',
		0, // Password ID MSB
		1, // Password ID LSB
		'p',
	}

	pkt := NewConnect()

	for i := 0; i < b.N; i++ {
		_, err := pkt.Decode(packet)
		if err != nil {
			panic(err)
		}
	}
}


<|start_filename|>packet/utils_test.go<|end_filename|>
package packet

import "io"

type errorWriter struct {
	writer io.Writer
	after  int
	err    error
}

func (w *errorWriter) Write(p []byte) (int, error) {
	if w.after == 0 {
		return 0, w.err
	}

	return w.writer.Write(p)
}

type errorReader struct {
	reader io.Reader
	after  int
	err    error
}

func (r *errorReader) Read(p []byte) (int, error) {
	if r.after == 0 {
		return 0, r.err
	}

	r.after--
	return r.reader.Read(p)
}


<|start_filename|>packet/type_test.go<|end_filename|>
package packet

import (
	"testing"

	"github.com/stretchr/testify/assert"
)

func TestTypes(t *testing.T) {
	assert.Len(t, Types(), 14)
}

func TestTypeString(t *testing.T) {
	assert.Equal(t, "Unknown", Type(99).String())
}

func TestTypeValid(t *testing.T) {
	assert.True(t, CONNECT.Valid())
}

func TestTypeNew(t *testing.T) {
	for _, tt := range Types() {
		m, err := tt.New()
		assert.NotNil(t, m)
		assert.NoError(t, err)
	}
}


<|start_filename|>packet/packet.go<|end_filename|>
// Package packet implements functionality for encoding and decoding MQTT packets.
package packet

import "encoding/binary"

// QOS is the type used to store quality of service levels.
type QOS byte

const (
	// QOSAtMostOnce defines that the message is delivered at most once, or it
	// may not be delivered at all.
	QOSAtMostOnce QOS = iota

	// QOSAtLeastOnce defines that the message is always delivered at least once.
	QOSAtLeastOnce QOS = iota

	// QOSExactlyOnce defines that the message is always delivered exactly once.
	QOSExactlyOnce QOS = iota

	// QOSFailure indicates that there has been an error while subscribing
	// to a specific topic.
	QOSFailure QOS = 0x80
)

// Successful returns if the provided quality of service level represents a
// successful value.
func (qos QOS) Successful() bool {
	return qos == QOSAtMostOnce || qos == QOSAtLeastOnce || qos == QOSExactlyOnce
}

// ID is the type used to store packet ids.
type ID uint16

// Valid returns whether this packet id is valid.
func (id ID) Valid() bool {
	return id != 0
}

// Generic is an MQTT control packet that can be encoded to a buffer or decoded
// from a buffer.
type Generic interface {
	// Type returns the packets type.
	Type() Type

	// Len returns the byte length of the encoded packet.
	Len() int

	// Decode reads from the byte slice argument. It returns the total number of
	// bytes decoded, and whether there have been any errors during the process.
	Decode(src []byte) (int, error)

	// Encode writes the packet bytes into the byte slice from the argument. It
	// returns the number of bytes encoded and whether there's any errors along
	// the way. If there is an error, the byte slice should be considered invalid.
	Encode(dst []byte) (int, error)

	// String returns a string representation of the packet.
	String() string
}

// DetectPacket tries to detect the next packet in a buffer. It returns a length
// greater than zero if the packet has been detected as well as its Type.
func DetectPacket(src []byte) (int, Type) {
	// check for minimum size
	if len(src) < 2 {
		return 0, 0
	}

	// get type
	t := Type(src[0] >> 4)

	// get remaining length
	rl, n := binary.Uvarint(src[1:])
	if n <= 0 {
		return 0, 0
	}

	return 1 + n + int(rl), t
}

// GetID checks the packets type and returns its ID and true, or if it
// does not have a ID, zero and false.
func GetID(pkt Generic) (ID, bool) {
	switch pkt.Type() {
	case PUBLISH:
		return pkt.(*Publish).ID, true
	case PUBACK:
		return pkt.(*Puback).ID, true
	case PUBREC:
		return pkt.(*Pubrec).ID, true
	case PUBREL:
		return pkt.(*Pubrel).ID, true
	case PUBCOMP:
		return pkt.(*Pubcomp).ID, true
	case SUBSCRIBE:
		return pkt.(*Subscribe).ID, true
	case SUBACK:
		return pkt.(*Suback).ID, true
	case UNSUBSCRIBE:
		return pkt.(*Unsubscribe).ID, true
	case UNSUBACK:
		return pkt.(*Unsuback).ID, true
	}

	return 0, false
}

// Fuzz is a basic fuzzing test that works with https://github.com/dvyukov/go-fuzz:
//
//		$ go-fuzz-build github.com/gomqtt/packet
//		$ go-fuzz -bin=./packet-fuzz.zip -workdir=./fuzz
func Fuzz(data []byte) int {
	// check for zero length data
	if len(data) == 0 {
		return 1
	}

	// detect packet
	_, mt := DetectPacket(data)

	// for testing purposes we will not cancel
	// on incomplete buffers

	// create a new packet
	pkt, err := mt.New()
	if err != nil {
		return 0
	}

	// decode it from the buffer.
	_, err = pkt.Decode(data)
	if err != nil {
		return 0
	}

	// everything was ok
	return 1
}


<|start_filename|>packet/message.go<|end_filename|>
package packet

import "fmt"

// A Message bundles data that is published between brokers and clients.
type Message struct {
	// The Topic of the message.
	Topic string

	// The Payload of the message.
	Payload []byte

	// The QOS indicates the level of assurance for delivery.
	QOS QOS

	// If the Retain flag is set to true, the server must store the message,
	// so that it can be delivered to future subscribers whose subscriptions
	// match its topic name.
	Retain bool
}

// String returns a string representation of the message.
func (m *Message) String() string {
	return fmt.Sprintf("<Message Topic=%q QOS=%d Retain=%t Payload=%x>",
		m.Topic, m.QOS, m.Retain, m.Payload)
}

// Copy returns a copy of the message.
func (m Message) Copy() *Message {
	return &m
}


<|start_filename|>packet/identified.go<|end_filename|>
package packet

import (
	"fmt"
)

// returns the byte length of an identified packet
func identifiedLen() int {
	return headerLen(2) + 2
}

// decodes an identified packet
func identifiedDecode(src []byte, id *ID, t Type) (int, error) {
	// decode header
	total, _, rl, err := decodeHeader(src, t)
	if err != nil {
		return total, err
	}

	// check remaining length
	if rl != 2 {
		return total, makeError(t, "expected remaining length to be 2")
	}

	// read packet id
	pid, n, err := readUint(src[total:], 2, t)
	total += n
	if err != nil {
		return total, err
	}

	// get packet id
	packetID := ID(pid)
	if !packetID.Valid() {
		return total, makeError(t, "packet id must be grater than zero")
	}

	// set packet id
	*id = packetID

	return total, nil
}

// encodes an identified packet
func identifiedEncode(dst []byte, id ID, t Type) (int, error) {
	// check packet id
	if !id.Valid() {
		return 0, makeError(t, "packet id must be grater than zero")
	}

	// encode header
	total, err := encodeHeader(dst, 0, 2, identifiedLen(), t)
	if err != nil {
		return total, err
	}

	// write packet id
	n, err := writeUint(dst[total:], uint64(id), 2, t)
	total += n
	if err != nil {
		return total, err
	}

	return total, nil
}

// A Puback packet is the response to a Publish packet with QOS level 1.
type Puback struct {
	// The packet identifier.
	ID ID
}

// NewPuback creates a new Puback packet.
func NewPuback() *Puback {
	return &Puback{}
}

// Type returns the packets type.
func (p *Puback) Type() Type {
	return PUBACK
}

// Len returns the byte length of the encoded packet.
func (p *Puback) Len() int {
	return identifiedLen()
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (p *Puback) Decode(src []byte) (int, error) {
	return identifiedDecode(src, &p.ID, PUBACK)
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (p *Puback) Encode(dst []byte) (int, error) {
	return identifiedEncode(dst, p.ID, PUBACK)
}

// String returns a string representation of the packet.
func (p *Puback) String() string {
	return fmt.Sprintf("<Puback ID=%d>", p.ID)
}

// A Pubcomp packet is the response to a Pubrel. It is the fourth and
// final packet of the QOS 2 protocol exchange.
type Pubcomp struct {
	// The packet identifier.
	ID ID
}

// NewPubcomp creates a new Pubcomp packet.
func NewPubcomp() *Pubcomp {
	return &Pubcomp{}
}

// Type returns the packets type.
func (p *Pubcomp) Type() Type {
	return PUBCOMP
}

// Len returns the byte length of the encoded packet.
func (p *Pubcomp) Len() int {
	return identifiedLen()
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (p *Pubcomp) Decode(src []byte) (int, error) {
	return identifiedDecode(src, &p.ID, PUBCOMP)
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (p *Pubcomp) Encode(dst []byte) (int, error) {
	return identifiedEncode(dst, p.ID, PUBCOMP)
}

// String returns a string representation of the packet.
func (p *Pubcomp) String() string {
	return fmt.Sprintf("<Pubcomp ID=%d>", p.ID)
}

// A Pubrec packet is the response to a Publish packet with QOS 2. It is the
// second packet of the QOS 2 protocol exchange.
type Pubrec struct {
	// Shared packet identifier.
	ID ID
}

// NewPubrec creates a new Pubrec packet.
func NewPubrec() *Pubrec {
	return &Pubrec{}
}

// Type returns the packets type.
func (p *Pubrec) Type() Type {
	return PUBREC
}

// Len returns the byte length of the encoded packet.
func (p *Pubrec) Len() int {
	return identifiedLen()
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (p *Pubrec) Decode(src []byte) (int, error) {
	return identifiedDecode(src, &p.ID, PUBREC)
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (p *Pubrec) Encode(dst []byte) (int, error) {
	return identifiedEncode(dst, p.ID, PUBREC)
}

// String returns a string representation of the packet.
func (p *Pubrec) String() string {
	return fmt.Sprintf("<Pubrec ID=%d>", p.ID)
}

// A Pubrel packet is the response to a Pubrec packet. It is the third packet of
// the QOS 2 protocol exchange.
type Pubrel struct {
	// Shared packet identifier.
	ID ID
}

// NewPubrel creates a new Pubrel packet.
func NewPubrel() *Pubrel {
	return &Pubrel{}
}

// Type returns the packets type.
func (p *Pubrel) Type() Type {
	return PUBREL
}

// Len returns the byte length of the encoded packet.
func (p *Pubrel) Len() int {
	return identifiedLen()
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (p *Pubrel) Decode(src []byte) (int, error) {
	return identifiedDecode(src, &p.ID, PUBREL)
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (p *Pubrel) Encode(dst []byte) (int, error) {
	return identifiedEncode(dst, p.ID, PUBREL)
}

// String returns a string representation of the packet.
func (p *Pubrel) String() string {
	return fmt.Sprintf("<Pubrel ID=%d>", p.ID)
}

// An Unsuback packet is sent by the server to the client to confirm receipt of
// an Unsubscribe packet.
type Unsuback struct {
	// Shared packet identifier.
	ID ID
}

// NewUnsuback creates a new Unsuback packet.
func NewUnsuback() *Unsuback {
	return &Unsuback{}
}

// Type returns the packets type.
func (u *Unsuback) Type() Type {
	return UNSUBACK
}

// Len returns the byte length of the encoded packet.
func (u *Unsuback) Len() int {
	return identifiedLen()
}

// Decode reads from the byte slice argument. It returns the total number of
// bytes decoded, and whether there have been any errors during the process.
func (u *Unsuback) Decode(src []byte) (int, error) {
	return identifiedDecode(src, &u.ID, UNSUBACK)
}

// Encode writes the packet bytes into the byte slice from the argument. It
// returns the number of bytes encoded and whether there's any errors along
// the way. If there is an error, the byte slice should be considered invalid.
func (u *Unsuback) Encode(dst []byte) (int, error) {
	return identifiedEncode(dst, u.ID, UNSUBACK)
}

// String returns a string representation of the packet.
func (u *Unsuback) String() string {
	return fmt.Sprintf("<Unsuback ID=%d>", u.ID)
}
