use vendor

40 files changed, 5713 insertions, 0 deletions

diff --git a/vendor/github.com/256dpi/gomqtt/LICENSE b/vendor/github.com/256dpi/gomqtt/LICENSE
new file mode 100644
index 000000000..5c304d1a4
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/LICENSE
@@ -0,0 +1,201 @@
+Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "{}"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright {yyyy} {name of copyright owner}
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
diff --git a/vendor/github.com/256dpi/gomqtt/broker/backend.go b/vendor/github.com/256dpi/gomqtt/broker/backend.go
new file mode 100644
index 000000000..efdb7ecb8
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/broker/backend.go
@@ -0,0 +1,506 @@
+package broker
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/256dpi/gomqtt/packet"
+	"github.com/256dpi/gomqtt/session"
+	"github.com/256dpi/gomqtt/topic"
+)
+
+type memorySession struct {
+	*session.MemorySession
+
+	subscriptions *topic.Tree
+	stored        chan *packet.Message
+	temporary     chan *packet.Message
+
+	owner *Client
+}
+
+func newMemorySession(backlog int) *memorySession {
+	return &memorySession{
+		MemorySession: session.NewMemorySession(),
+		subscriptions: topic.NewTree(),
+		stored:        make(chan *packet.Message, backlog),
+		temporary:     make(chan *packet.Message, backlog),
+	}
+}
+
+func (s *memorySession) lookupSubscription(topic string) *packet.Subscription {
+	values := s.subscriptions.Match(topic)
+
+	if len(values) > 0 {
+		sub := values[0].(packet.Subscription)
+		return &sub
+	}
+
+	return nil
+}
+
+func (s *memorySession) applyQOS(msg *packet.Message) *packet.Message {
+	// get subscription
+	sub := s.lookupSubscription(msg.Topic)
+	if sub != nil {
+		// respect maximum qos
+		if msg.QOS > sub.QOS {
+			msg = msg.Copy()
+			msg.QOS = sub.QOS
+		}
+	}
+
+	return msg
+}
+
+func (s *memorySession) reuse() {
+	s.temporary = make(chan *packet.Message, cap(s.temporary))
+}
+
+// ErrQueueFull is returned to a client that attempts two write to its own full
+// queue, which would result in a deadlock.
+var ErrQueueFull = errors.New("queue full")
+
+// ErrClosing is returned to a client if the backend is closing.
+var ErrClosing = errors.New("closing")
+
+// ErrKillTimeout is returned to a client if the existing client does not close
+// in time.
+var ErrKillTimeout = errors.New("kill timeout")
+
+// A MemoryBackend stores everything in memory.
+type MemoryBackend struct {
+	// The maximal size of the session queue.
+	//
+	// Will default to 100.
+	SessionQueueSize int
+
+	// The time after an error is returned while waiting on an killed existing
+	// client to exit.
+	//
+	// Will default to 5 seconds.
+	KillTimeout time.Duration
+
+	// Client configuration options. See broker.Client for details.
+	ClientParallelPublishes  int
+	ClientParallelSubscribes int
+	ClientInflightMessages   int
+	ClientTokenTimeout       time.Duration
+
+	// A map of username and passwords that grant read and write access.
+	Credentials map[string]string
+
+	// The Logger callback handles incoming log events.
+	Logger func(LogEvent, *Client, packet.Generic, *packet.Message, error)
+
+	activeClients     map[string]*Client
+	storedSessions    map[string]*memorySession
+	temporarySessions map[*Client]*memorySession
+	retainedMessages  *topic.Tree
+
+	globalMutex sync.Mutex
+	setupMutex  sync.Mutex
+	closing     bool
+}
+
+// NewMemoryBackend returns a new MemoryBackend.
+func NewMemoryBackend() *MemoryBackend {
+	return &MemoryBackend{
+		SessionQueueSize:  100,
+		KillTimeout:       5 * time.Second,
+		activeClients:     make(map[string]*Client),
+		storedSessions:    make(map[string]*memorySession),
+		temporarySessions: make(map[*Client]*memorySession),
+		retainedMessages:  topic.NewTree(),
+	}
+}
+
+// Authenticate will authenticates a clients credentials.
+func (m *MemoryBackend) Authenticate(client *Client, user, password string) (bool, error) {
+	// acquire global mutex
+	m.globalMutex.Lock()
+	defer m.globalMutex.Unlock()
+
+	// return error if closing
+	if m.closing {
+		return false, ErrClosing
+	}
+
+	// allow all if there are no credentials
+	if m.Credentials == nil {
+		return true, nil
+	}
+
+	// check login
+	if pw, ok := m.Credentials[user]; ok && pw == password {
+		return true, nil
+	}
+
+	return false, nil
+}
+
+// Setup will close existing clients and return an appropriate session.
+func (m *MemoryBackend) Setup(client *Client, id string, clean bool) (Session, bool, error) {
+	// acquire setup mutex
+	m.setupMutex.Lock()
+	defer m.setupMutex.Unlock()
+
+	// acquire global mutex
+	m.globalMutex.Lock()
+	defer m.globalMutex.Unlock()
+
+	// return error if closing
+	if m.closing {
+		return nil, false, ErrClosing
+	}
+
+	// apply client settings
+	client.ParallelPublishes = m.ClientParallelPublishes
+	client.ParallelSubscribes = m.ClientParallelSubscribes
+	client.InflightMessages = m.ClientInflightMessages
+	client.TokenTimeout = m.ClientTokenTimeout
+
+	// return a new temporary session if id is zero
+	if len(id) == 0 {
+		// create session
+		sess := newMemorySession(m.SessionQueueSize)
+		sess.owner = client
+
+		// save session
+		m.temporarySessions[client] = sess
+
+		return sess, false, nil
+	}
+
+	// client id is available
+
+	// retrieve existing client
+	existingSession, ok := m.storedSessions[id]
+	if !ok {
+		if existingClient, ok2 := m.activeClients[id]; ok2 {
+			existingSession, ok = m.temporarySessions[existingClient]
+		}
+	}
+
+	// kill existing client if session is taken
+	if ok && existingSession.owner != nil {
+		// close client
+		existingSession.owner.Close()
+
+		// release global mutex to allow publish and termination, but leave the
+		// setup mutex to prevent setups
+		m.globalMutex.Unlock()
+
+		// wait for client to close
+		var err error
+		select {
+		case <-existingSession.owner.Closed():
+			// continue
+		case <-time.After(m.KillTimeout):
+			err = ErrKillTimeout
+		}
+
+		// acquire mutex again
+		m.globalMutex.Lock()
+
+		// return err if set
+		if err != nil {
+			return nil, false, err
+		}
+	}
+
+	// delete any stored session and return a temporary session if a clean
+	// session is requested
+	if clean {
+		// delete any stored session
+		delete(m.storedSessions, id)
+
+		// create new session
+		sess := newMemorySession(m.SessionQueueSize)
+		sess.owner = client
+
+		// save session
+		m.temporarySessions[client] = sess
+
+		// save client
+		m.activeClients[id] = client
+
+		return sess, false, nil
+	}
+
+	// attempt to reuse a stored session
+	storedSession, ok := m.storedSessions[id]
+	if ok {
+		// reuse session
+		storedSession.reuse()
+		storedSession.owner = client
+
+		// save client
+		m.activeClients[id] = client
+
+		return storedSession, true, nil
+	}
+
+	// otherwise create fresh session
+	storedSession = newMemorySession(m.SessionQueueSize)
+	storedSession.owner = client
+
+	// save session
+	m.storedSessions[id] = storedSession
+
+	// save client
+	m.activeClients[id] = client
+
+	return storedSession, false, nil
+}
+
+// Restore is not needed at the moment.
+func (m *MemoryBackend) Restore(client *Client) error {
+	return nil
+}
+
+// Subscribe will store the subscription and queue retained messages.
+func (m *MemoryBackend) Subscribe(client *Client, subs []packet.Subscription, ack Ack) error {
+	// acquire global mutex
+	m.globalMutex.Lock()
+	defer m.globalMutex.Unlock()
+
+	// save subscription
+	for _, sub := range subs {
+		client.Session().(*memorySession).subscriptions.Set(sub.Topic, sub)
+	}
+
+	// call ack if provided
+	if ack != nil {
+		ack()
+	}
+
+	// get session
+	sess := client.Session().(*memorySession)
+
+	// handle all subscriptions
+	for _, sub := range subs {
+		// get retained messages
+		values := m.retainedMessages.Search(sub.Topic)
+
+		// publish messages
+		for _, value := range values {
+			// add to temporary queue or return error if queue is full
+			select {
+			case sess.temporary <- value.(*packet.Message):
+			default:
+				return ErrQueueFull
+			}
+		}
+	}
+
+	return nil
+}
+
+// Unsubscribe will delete the subscription.
+func (m *MemoryBackend) Unsubscribe(client *Client, topics []string, ack Ack) error {
+	// delete subscriptions
+	for _, t := range topics {
+		client.Session().(*memorySession).subscriptions.Empty(t)
+	}
+
+	// call ack if provided
+	if ack != nil {
+		ack()
+	}
+
+	return nil
+}
+
+// Publish will handle retained messages and add the message to the session queues.
+func (m *MemoryBackend) Publish(client *Client, msg *packet.Message, ack Ack) error {
+	// acquire global mutex
+	m.globalMutex.Lock()
+	defer m.globalMutex.Unlock()
+
+	// this implementation is very basic and will block the backend on every
+	// publish. clients that stay connected but won't drain their queue will
+	// eventually deadlock the broker
+
+	// check retain flag
+	if msg.Retain {
+		if len(msg.Payload) > 0 {
+			// retain message
+			m.retainedMessages.Set(msg.Topic, msg.Copy())
+		} else {
+			// clear already retained message
+			m.retainedMessages.Empty(msg.Topic)
+		}
+	}
+
+	// use temporary queue by default
+	queue := func(s *memorySession) chan *packet.Message {
+		return s.temporary
+	}
+
+	// use stored queue if qos > 0
+	if msg.QOS > 0 {
+		queue = func(s *memorySession) chan *packet.Message {
+			return s.stored
+		}
+	}
+
+	// reset retained flag
+	msg.Retain = false
+
+	// add message to temporary sessions
+	for _, sess := range m.temporarySessions {
+		if sub := sess.lookupSubscription(msg.Topic); sub != nil {
+			if sess.owner == client {
+				// detect deadlock when adding to own queue
+				select {
+				case queue(sess) <- msg:
+				default:
+					return ErrQueueFull
+				}
+			} else {
+				// wait for room since client is online
+				select {
+				case queue(sess) <- msg:
+				case <-sess.owner.Closed():
+				case <-client.Closed():
+				}
+			}
+		}
+	}
+
+	// add message to stored sessions
+	for _, sess := range m.storedSessions {
+		if sub := sess.lookupSubscription(msg.Topic); sub != nil {
+			if sess.owner == client {
+				// detect deadlock when adding to own queue
+				select {
+				case queue(sess) <- msg:
+				default:
+					return ErrQueueFull
+				}
+			} else if sess.owner != nil {
+				// wait for room if client is online
+				select {
+				case queue(sess) <- msg:
+				case <-sess.owner.Closed():
+				case <-client.Closed():
+				}
+			} else {
+				// ignore message if stored queue is full
+				select {
+				case queue(sess) <- msg:
+				default:
+				}
+			}
+		}
+	}
+
+	// call ack if available
+	if ack != nil {
+		ack()
+	}
+
+	return nil
+}
+
+// Dequeue will get the next message from the temporary or stored queue.
+func (m *MemoryBackend) Dequeue(client *Client) (*packet.Message, Ack, error) {
+	// mutex locking not needed
+
+	// get session
+	sess := client.Session().(*memorySession)
+
+	// this implementation is very basic and will dequeue messages immediately
+	// and not return no ack. messages are lost if the client fails to handle them
+
+	// get next message from queue
+	select {
+	case msg := <-sess.temporary:
+		return sess.applyQOS(msg), nil, nil
+	case msg := <-sess.stored:
+		return sess.applyQOS(msg), nil, nil
+	case <-client.Closing():
+		return nil, nil, nil
+	}
+}
+
+// Terminate will disassociate the session from the client.
+func (m *MemoryBackend) Terminate(client *Client) error {
+	// acquire global mutex
+	m.globalMutex.Lock()
+	defer m.globalMutex.Unlock()
+
+	// release session if available
+	sess, ok := client.Session().(*memorySession)
+	if ok && sess != nil {
+		sess.owner = nil
+	}
+
+	// remove any temporary session
+	delete(m.temporarySessions, client)
+
+	// remove any saved client
+	delete(m.activeClients, client.ID())
+
+	return nil
+}
+
+// Log will call the associated logger.
+func (m *MemoryBackend) Log(event LogEvent, client *Client, pkt packet.Generic, msg *packet.Message, err error) {
+	// call logger if available
+	if m.Logger != nil {
+		m.Logger(event, client, pkt, msg, err)
+	}
+}
+
+// Close will close all active clients and close the backend. The return value
+// denotes if the timeout has been reached.
+func (m *MemoryBackend) Close(timeout time.Duration) bool {
+	// acquire global mutex
+	m.globalMutex.Lock()
+
+	// set closing
+	m.closing = true
+
+	// prepare list
+	var clients []*Client
+
+	// close temporary sessions
+	for _, sess := range m.temporarySessions {
+		sess.owner.Close()
+		clients = append(clients, sess.owner)
+	}
+
+	// closed owned stored sessions
+	for _, sess := range m.storedSessions {
+		if sess.owner != nil {
+			sess.owner.Close()
+			clients = append(clients, sess.owner)
+		}
+	}
+
+	// release mutex to allow termination
+	m.globalMutex.Unlock()
+
+	// return early if empty
+	if len(clients) == 0 {
+		return true
+	}
+
+	// prepare timeout
+	tm := time.After(timeout)
+
+	// wait for clients to close
+	for _, client := range clients {
+		select {
+		case <-client.Closed():
+			continue
+		case <-tm:
+			return false
+		}
+	}
+
+	return true
+}
diff --git a/vendor/github.com/256dpi/gomqtt/broker/client.go b/vendor/github.com/256dpi/gomqtt/broker/client.go
new file mode 100644
index 000000000..e254254fa
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/broker/client.go
@@ -0,0 +1,946 @@
+package broker
+
+import (
+	"errors"
+	"sync/atomic"
+	"time"
+
+	"github.com/256dpi/gomqtt/packet"
+	"github.com/256dpi/gomqtt/session"
+	"github.com/256dpi/gomqtt/transport"
+
+	"gopkg.in/tomb.v2"
+)
+
+// LogEvent denotes the class of an event passed to the logger.
+type LogEvent string
+
+const (
+	// NewConnection is emitted when a client comes online.
+	NewConnection LogEvent = "new connection"
+
+	// PacketReceived is emitted when a packet has been received.
+	PacketReceived LogEvent = "packet received"
+
+	// MessagePublished is emitted after a message has been published.
+	MessagePublished LogEvent = "message published"
+
+	// MessageAcknowledged is emitted after a message has been acknowledged.
+	MessageAcknowledged LogEvent = "message acknowledged"
+
+	// MessageDequeued is emitted after a message has been dequeued.
+	MessageDequeued LogEvent = "message dequeued"
+
+	// MessageForwarded is emitted after a message has been forwarded.
+	MessageForwarded LogEvent = "message forwarded"
+
+	// PacketSent is emitted when a packet has been sent.
+	PacketSent LogEvent = "packet sent"
+
+	// ClientDisconnected is emitted when a client disconnects cleanly.
+	ClientDisconnected LogEvent = "client disconnected"
+
+	// TransportError is emitted when an underlying transport error occurs.
+	TransportError LogEvent = "transport error"
+
+	// SessionError is emitted when a call to the session fails.
+	SessionError LogEvent = "session error"
+
+	// BackendError is emitted when a call to the backend fails.
+	BackendError LogEvent = "backend error"
+
+	// ClientError is emitted when the client violates the protocol.
+	ClientError LogEvent = "client error"
+
+	// LostConnection is emitted when the connection has been terminated.
+	LostConnection LogEvent = "lost connection"
+)
+
+// A Session is used to get packet ids and persist incoming/outgoing packets.
+type Session interface {
+	// NextID should return the next id for outgoing packets.
+	NextID() packet.ID
+
+	// SavePacket should store a packet in the session. An eventual existing
+	// packet with the same id should be quietly overwritten.
+	SavePacket(session.Direction, packet.Generic) error
+
+	// LookupPacket should retrieve a packet from the session using the packet id.
+	LookupPacket(session.Direction, packet.ID) (packet.Generic, error)
+
+	// DeletePacket should remove a packet from the session. The method should
+	// not return an error if no packet with the specified id does exists.
+	DeletePacket(session.Direction, packet.ID) error
+
+	// AllPackets should return all packets currently saved in the session.
+	AllPackets(session.Direction) ([]packet.Generic, error)
+}
+
+// Ack is executed by the Backend or Client to signal either that a message will
+// be delivered under the selected qos level and is therefore safe to be deleted
+// from either queue or the successful handling of subscriptions.
+type Ack func()
+
+// A Backend provides the effective brokering functionality to its clients.
+type Backend interface {
+	// Authenticate should authenticate the client using the user and password
+	// values and return true if the client is eligible to continue or false
+	// when the broker should terminate the connection.
+	Authenticate(client *Client, user, password string) (ok bool, err error)
+
+	// Setup is called when a new client comes online and is successfully
+	// authenticated. Setup should return the already stored session for the
+	// supplied id or create and return a new one if it is missing or a clean
+	// session is requested. If the supplied id has a zero length, a new
+	// temporary session should be returned that is not stored further. The
+	// backend should also close any existing clients that use the same id.
+	//
+	// Note: In this call the Backend may also allocate other resources and
+	// setup the client for further usage as the broker will acknowledge the
+	// connection when the call returns. The Terminate function is called for
+	// every client that Setup has been called for.
+	Setup(client *Client, id string, clean bool) (a Session, resumed bool, err error)
+
+	// Restore is called after the client has restored packets from the session.
+	//
+	// The Backend should resubscribe stored subscriptions and begin with queueing
+	// missed offline messages. When all offline messages have been queued the
+	// client may receive online messages. Depending on the implementation, this
+	// may not be required as Dequeue will already pick up offline messages.
+	Restore(client *Client) error
+
+	// Subscribe should subscribe the passed client to the specified topics and
+	// store the subscription in the session. If an Ack is provided, the
+	// subscription will be acknowledged when called during or after the call to
+	// Subscribe.
+	//
+	// Incoming messages that match the supplied subscription should be added to
+	// a temporary or persistent queue that is drained when Dequeue is called.
+	//
+	// Retained messages that match the supplied subscription should be added to
+	// a temporary queue that is also drained when Dequeue is called. The messages
+	// must be delivered with the retained flag set to true.
+	Subscribe(client *Client, subs []packet.Subscription, ack Ack) error
+
+	// Unsubscribe should unsubscribe the passed client from the specified topics
+	// and remove the subscriptions from the session. If an Ack is provided, the
+	// unsubscription will be acknowledged when called during or after the call
+	// to Unsubscribe.
+	Unsubscribe(client *Client, topics []string, ack Ack) error
+
+	// Publish should forward the passed message to all other clients that hold
+	// a subscription that matches the messages topic. It should also add the
+	// message to all sessions that have a matching offline subscription. The
+	// later may only apply to messages with a QOS greater than 0. If an Ack is
+	// provided, the message will be acknowledged when called during or after
+	// the call to Publish.
+	//
+	// If the retained flag is set, messages with a payload should replace the
+	// currently retained message. Otherwise, the currently retained message
+	// should be removed. The flag should be cleared before publishing the
+	// message to other subscribed clients.
+	Publish(client *Client, msg *packet.Message, ack Ack) error
+
+	// Dequeue is called by the Client to obtain the next message from the queue
+	// and must return either a message or an error. The backend must only return
+	// no message and no error if the client's Closing channel has been closed.
+	//
+	// The Backend may return an Ack to receive a signal that the message is being
+	// delivered under the selected qos level and is therefore safe to be deleted
+	// from the queue. The Ack will be called before Dequeue is called again.
+	//
+	// The returned message must have a QOS set that respects the QOS set by
+	// the matching subscription.
+	Dequeue(client *Client) (*packet.Message, Ack, error)
+
+	// Terminate is called when the client goes offline. Terminate should
+	// unsubscribe the passed client from all previously subscribed topics. The
+	// backend may also convert a clients subscriptions to offline subscriptions.
+	//
+	// Note: The Backend may also cleanup previously allocated resources for
+	// that client as the broker will close the connection when the call
+	// returns.
+	Terminate(client *Client) error
+
+	// Log is called multiple times during the lifecycle of a client see LogEvent
+	// for a list of all events.
+	Log(event LogEvent, client *Client, pkt packet.Generic, msg *packet.Message, err error)
+}
+
+// ErrUnexpectedPacket is returned when an unexpected packet is received.
+var ErrUnexpectedPacket = errors.New("unexpected packet")
+
+// ErrNotAuthorized is returned when a client is not authorized.
+var ErrNotAuthorized = errors.New("not authorized")
+
+// ErrMissingSession is returned if the backend does not return a session.
+var ErrMissingSession = errors.New("missing session")
+
+// ErrTokenTimeout is returned if the client reaches the token timeout.
+var ErrTokenTimeout = errors.New("token timeout")
+
+// ErrClientDisconnected is returned if a client disconnects cleanly.
+var ErrClientDisconnected = errors.New("client disconnected")
+
+// ErrClientClosed is returned if a client is being closed by the broker.
+var ErrClientClosed = errors.New("client closed")
+
+const (
+	clientConnecting uint32 = iota
+	clientConnected
+	clientDisconnected
+)
+
+// A Client represents a remote client that is connected to the broker.
+type Client struct {
+	// ParallelPublishes may be set during Setup to control the number of
+	// parallel calls to Publish a client can perform. This setting also has an
+	// effect on how many incoming packets are stored in the clients session.
+	//
+	// Will default to 10.
+	ParallelPublishes int
+
+	// ParallelSubscribes may be set during Setup to control the number of
+	// parallel calls to Subscribe and Unsubscribe a client can perform.
+	//
+	// Will default to 10.
+	ParallelSubscribes int
+
+	// InflightMessages may be set during Setup to control the number of
+	// inflight messages from the broker to the client. This also defines how
+	// many outgoing packets are stored in the clients session.
+	//
+	// Will default to 10.
+	InflightMessages int
+
+	// TokenTimeout sets the timeout after which the client should fail when
+	// obtaining publish, subscribe and dequeue tokens in order to prevent
+	// potential deadlocks.
+	//
+	// Will default to 30 seconds.
+	TokenTimeout time.Duration
+
+	// PacketCallback can be set to inspect packets before processing and
+	// apply rate limits. To guarantee the connection lifecycle, Connect and
+	// Disconnect packets are not provided to the callback.
+	PacketCallback func(packet.Generic) error
+
+	state   uint32
+	backend Backend
+	conn    transport.Conn
+
+	id      string
+	will    *packet.Message
+	session Session
+
+	ackQueue chan packet.Generic
+
+	publishTokens   chan struct{}
+	subscribeTokens chan struct{}
+	dequeueTokens   chan struct{}
+
+	tomb tomb.Tomb
+	done chan struct{}
+}
+
+// NewClient takes over a connection and returns a Client.
+func NewClient(backend Backend, conn transport.Conn) *Client {
+	// create client
+	c := &Client{
+		state:   clientConnecting,
+		backend: backend,
+		conn:    conn,
+		done:    make(chan struct{}),
+	}
+
+	// start processor
+	c.tomb.Go(c.processor)
+
+	// run cleanup goroutine
+	go func() {
+		// wait for death and cleanup
+		c.tomb.Wait()
+		c.cleanup()
+
+		// close channel
+		close(c.done)
+	}()
+
+	return c
+}
+
+// Session returns the current Session used by the client.
+func (c *Client) Session() Session {
+	return c.session
+}
+
+// ID returns the clients id that has been supplied during connect.
+func (c *Client) ID() string {
+	return c.id
+}
+
+// Conn returns the client's underlying connection. Calls to SetReadLimit,
+// LocalAddr and RemoteAddr are safe.
+func (c *Client) Conn() transport.Conn {
+	return c.conn
+}
+
+// Close will immediately close the client.
+func (c *Client) Close() {
+	c.tomb.Kill(ErrClientClosed)
+	c.conn.Close()
+}
+
+// Closing returns a channel that is closed when the client is closing.
+func (c *Client) Closing() <-chan struct{} {
+	return c.tomb.Dying()
+}
+
+// Closed returns a channel that is closed when the client is closed.
+func (c *Client) Closed() <-chan struct{} {
+	return c.done
+}
+
+/* goroutines */
+
+// main processor
+func (c *Client) processor() error {
+	c.backend.Log(NewConnection, c, nil, nil, nil)
+
+	// get first packet from connection
+	pkt, err := c.conn.Receive()
+	if err != nil {
+		return c.die(TransportError, err)
+	}
+
+	c.backend.Log(PacketReceived, c, pkt, nil, nil)
+
+	// get connect
+	connect, ok := pkt.(*packet.Connect)
+	if !ok {
+		return c.die(ClientError, ErrUnexpectedPacket)
+	}
+
+	// process connect
+	err = c.processConnect(connect)
+	if err != nil {
+		return err // error has already been handled
+	}
+
+	// start dequeuer and acker
+	c.tomb.Go(c.dequeuer)
+	c.tomb.Go(c.acker)
+
+	for {
+		// check if still alive
+		if !c.tomb.Alive() {
+			return tomb.ErrDying
+		}
+
+		// receive next packet
+		pkt, err := c.conn.Receive()
+		if err != nil {
+			return c.die(TransportError, err)
+		}
+
+		c.backend.Log(PacketReceived, c, pkt, nil, nil)
+
+		// call callback
+		if c.PacketCallback != nil && pkt.Type() != packet.DISCONNECT {
+			err = c.PacketCallback(pkt)
+			if err != nil {
+				return c.die(ClientError, err)
+			}
+		}
+
+		// process packet
+		err = c.processPacket(pkt)
+		if err != nil {
+			return err // error has already been handled
+		}
+	}
+}
+
+// message dequeuer
+func (c *Client) dequeuer() error {
+	for {
+		// acquire dequeue token
+		select {
+		case <-c.dequeueTokens:
+			// continue
+		case <-time.After(c.TokenTimeout):
+			return c.die(ClientError, ErrTokenTimeout)
+		case <-c.tomb.Dying():
+			return tomb.ErrDying
+		}
+
+		// request next message
+		msg, ack, err := c.backend.Dequeue(c)
+		if err != nil {
+			return c.die(BackendError, err)
+		} else if msg == nil {
+			return tomb.ErrDying
+		}
+
+		c.backend.Log(MessageDequeued, c, nil, msg, nil)
+
+		// prepare publish packet
+		publish := packet.NewPublish()
+		publish.Message = *msg
+
+		// set packet id
+		if publish.Message.QOS > 0 {
+			publish.ID = c.session.NextID()
+		}
+
+		// store packet if at least qos 1
+		if publish.Message.QOS > 0 {
+			err := c.session.SavePacket(session.Outgoing, publish)
+			if err != nil {
+				return c.die(SessionError, err)
+			}
+		}
+
+		// acknowledge message
+		if ack != nil {
+			ack()
+
+			c.backend.Log(MessageAcknowledged, c, nil, msg, nil)
+		}
+
+		// send packet
+		err = c.send(publish, true)
+		if err != nil {
+			return c.die(TransportError, err)
+		}
+
+		// immediately put back dequeue token for qos 0 messages
+		if publish.Message.QOS == 0 {
+			select {
+			case c.dequeueTokens <- struct{}{}:
+			default:
+				// continue if full for some reason
+			}
+		}
+
+		c.backend.Log(MessageForwarded, c, nil, msg, nil)
+	}
+}
+
+// packet acker
+func (c *Client) acker() error {
+	for {
+		select {
+		case pkt := <-c.ackQueue:
+			// send packet
+			err := c.send(pkt, true)
+			if err != nil {
+				return err // error already handled
+			}
+
+			// remove publish from session if pubcomp
+			if pubcomp, ok := pkt.(*packet.Pubcomp); ok {
+				err = c.session.DeletePacket(session.Incoming, pubcomp.ID)
+				if err != nil {
+					return c.die(SessionError, err)
+				}
+			}
+
+			// put back tokens based on type
+			switch pkt.(type) {
+			case *packet.Suback, *packet.Unsuback:
+				select {
+				case c.subscribeTokens <- struct{}{}:
+				default:
+					// continue if full for some reason
+				}
+			case *packet.Puback, *packet.Pubcomp:
+				select {
+				case c.publishTokens <- struct{}{}:
+				default:
+					// continue if full for some reason
+				}
+			}
+		case <-c.tomb.Dying():
+			return tomb.ErrDying
+		}
+	}
+}
+
+/* packet handling */
+
+// handle an incoming Connect packet
+func (c *Client) processConnect(pkt *packet.Connect) error {
+	// save id
+	c.id = pkt.ClientID
+
+	// authenticate
+	ok, err := c.backend.Authenticate(c, pkt.Username, pkt.Password)
+	if err != nil {
+		return c.die(BackendError, err)
+	}
+
+	// prepare connack packet
+	connack := packet.NewConnack()
+	connack.ReturnCode = packet.ConnectionAccepted
+	connack.SessionPresent = false
+
+	// check authentication
+	if !ok {
+		// set return code
+		connack.ReturnCode = packet.NotAuthorized
+
+		// send connack
+		err = c.send(connack, false)
+		if err != nil {
+			return c.die(TransportError, err)
+		}
+
+		// close client
+		return c.die(ClientError, ErrNotAuthorized)
+	}
+
+	// set state
+	atomic.StoreUint32(&c.state, clientConnected)
+
+	// set keep alive
+	if pkt.KeepAlive > 0 {
+		c.conn.SetReadTimeout(time.Duration(pkt.KeepAlive) * 1500 * time.Millisecond)
+	} else {
+		c.conn.SetReadTimeout(0)
+	}
+
+	// retrieve session
+	s, resumed, err := c.backend.Setup(c, pkt.ClientID, pkt.CleanSession)
+	if err != nil {
+		return c.die(BackendError, err)
+	} else if s == nil {
+		return c.die(BackendError, ErrMissingSession)
+	}
+
+	// set session present
+	connack.SessionPresent = !pkt.CleanSession && resumed
+
+	// assign session
+	c.session = s
+
+	// set default parallel publishes
+	if c.ParallelPublishes <= 0 {
+		c.ParallelPublishes = 10
+	}
+
+	// set default parallel subscribes
+	if c.ParallelSubscribes <= 0 {
+		c.ParallelSubscribes = 10
+	}
+
+	// set default parallel dequeues
+	if c.InflightMessages <= 0 {
+		c.InflightMessages = 10
+	}
+
+	// set default token timeout
+	if c.TokenTimeout == 0 {
+		c.TokenTimeout = 30 * time.Second
+	}
+
+	// prepare publish tokens
+	c.publishTokens = make(chan struct{}, c.ParallelPublishes)
+	for i := 0; i < c.ParallelPublishes; i++ {
+		c.publishTokens <- struct{}{}
+	}
+
+	// prepare subscribe tokens
+	c.subscribeTokens = make(chan struct{}, c.ParallelSubscribes)
+	for i := 0; i < c.ParallelSubscribes; i++ {
+		c.subscribeTokens <- struct{}{}
+	}
+
+	// prepare dequeue tokens
+	c.dequeueTokens = make(chan struct{}, c.InflightMessages)
+	for i := 0; i < c.InflightMessages; i++ {
+		c.dequeueTokens <- struct{}{}
+	}
+
+	// create ack queue
+	c.ackQueue = make(chan packet.Generic, c.ParallelPublishes+c.ParallelSubscribes)
+
+	// save will if present
+	if pkt.Will != nil {
+		c.will = pkt.Will
+	}
+
+	// send connack
+	err = c.send(connack, false)
+	if err != nil {
+		return c.die(TransportError, err)
+	}
+
+	// retrieve stored packets
+	packets, err := c.session.AllPackets(session.Outgoing)
+	if err != nil {
+		return c.die(SessionError, err)
+	}
+
+	// resend stored packets
+	for _, pkt := range packets {
+		// consume a dequeue token (will be replaced once the flow is complete)
+		select {
+		case <-c.dequeueTokens:
+		default:
+			// continue if depleted
+		}
+
+		// set the dup flag on a publish packet
+		publish, ok := pkt.(*packet.Publish)
+		if ok {
+			publish.Dup = true
+		}
+
+		// send packet
+		err = c.send(pkt, true)
+		if err != nil {
+			return c.die(TransportError, err)
+		}
+	}
+
+	// restore client
+	err = c.backend.Restore(c)
+	if err != nil {
+		return c.die(BackendError, err)
+	}
+
+	return nil
+}
+
+// handle an incoming Generic
+func (c *Client) processPacket(pkt packet.Generic) error {
+	// prepare error
+	var err error
+
+	// handle individual packets
+	switch typedPkt := pkt.(type) {
+	case *packet.Subscribe:
+		err = c.processSubscribe(typedPkt)
+	case *packet.Unsubscribe:
+		err = c.processUnsubscribe(typedPkt)
+	case *packet.Publish:
+		err = c.processPublish(typedPkt)
+	case *packet.Puback:
+		err = c.processPubackAndPubcomp(typedPkt.ID)
+	case *packet.Pubcomp:
+		err = c.processPubackAndPubcomp(typedPkt.ID)
+	case *packet.Pubrec:
+		err = c.processPubrec(typedPkt.ID)
+	case *packet.Pubrel:
+		err = c.processPubrel(typedPkt.ID)
+	case *packet.Pingreq:
+		err = c.processPingreq()
+	case *packet.Disconnect:
+		err = c.processDisconnect()
+	default:
+		err = c.die(ClientError, ErrUnexpectedPacket)
+	}
+
+	// return eventual error
+	if err != nil {
+		return err // error has already been handled
+	}
+
+	return nil
+}
+
+// handle an incoming Pingreq packet
+func (c *Client) processPingreq() error {
+	// send a pingresp packet
+	err := c.send(packet.NewPingresp(), true)
+	if err != nil {
+		return c.die(TransportError, err)
+	}
+
+	return nil
+}
+
+// handle an incoming subscribe packet
+func (c *Client) processSubscribe(pkt *packet.Subscribe) error {
+	// acquire subscribe token
+	select {
+	case <-c.subscribeTokens:
+		// continue
+	case <-time.After(c.TokenTimeout):
+		return c.die(ClientError, ErrTokenTimeout)
+	case <-c.tomb.Dying():
+		return tomb.ErrDying
+	}
+
+	// prepare suback packet
+	suback := packet.NewSuback()
+	suback.ReturnCodes = make([]packet.QOS, len(pkt.Subscriptions))
+	suback.ID = pkt.ID
+
+	// set granted qos
+	for i, subscription := range pkt.Subscriptions {
+		suback.ReturnCodes[i] = subscription.QOS
+	}
+
+	// subscribe client to queue
+	err := c.backend.Subscribe(c, pkt.Subscriptions, func() {
+		select {
+		case c.ackQueue <- suback:
+		case <-c.tomb.Dying():
+		}
+	})
+	if err != nil {
+		return c.die(BackendError, err)
+	}
+
+	return nil
+}
+
+// handle an incoming unsubscribe packet
+func (c *Client) processUnsubscribe(pkt *packet.Unsubscribe) error {
+	// acquire subscribe token
+	select {
+	case <-c.subscribeTokens:
+		// continue
+	case <-time.After(c.TokenTimeout):
+		return c.die(ClientError, ErrTokenTimeout)
+	case <-c.tomb.Dying():
+		return tomb.ErrDying
+	}
+
+	// prepare unsuback packet
+	unsuback := packet.NewUnsuback()
+	unsuback.ID = pkt.ID
+
+	// unsubscribe topics
+	err := c.backend.Unsubscribe(c, pkt.Topics, func() {
+		select {
+		case c.ackQueue <- unsuback:
+		case <-c.tomb.Dying():
+		}
+	})
+	if err != nil {
+		return c.die(BackendError, err)
+	}
+
+	return nil
+}
+
+// handle an incoming publish packet
+func (c *Client) processPublish(publish *packet.Publish) error {
+	// handle qos 0 flow
+	if publish.Message.QOS == 0 {
+		// publish message
+		err := c.backend.Publish(c, &publish.Message, nil)
+		if err != nil {
+			return c.die(BackendError, err)
+		}
+
+		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)
+
+		return nil
+	}
+
+	// acquire publish token
+	select {
+	case <-c.publishTokens:
+		// continue
+	case <-time.After(c.TokenTimeout):
+		return c.die(ClientError, ErrTokenTimeout)
+	case <-c.tomb.Dying():
+		return tomb.ErrDying
+	}
+
+	// handle qos 1 flow
+	if publish.Message.QOS == 1 {
+		// prepare puback
+		puback := packet.NewPuback()
+		puback.ID = publish.ID
+
+		// publish message and queue puback if ack is called
+		err := c.backend.Publish(c, &publish.Message, func() {
+			c.backend.Log(MessageAcknowledged, c, nil, &publish.Message, nil)
+
+			select {
+			case c.ackQueue <- puback:
+			case <-c.tomb.Dying():
+			}
+		})
+		if err != nil {
+			return c.die(BackendError, err)
+		}
+
+		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)
+	}
+
+	// handle qos 2 flow
+	if publish.Message.QOS == 2 {
+		// store received publish packet in session
+		err := c.session.SavePacket(session.Incoming, publish)
+		if err != nil {
+			return c.die(SessionError, err)
+		}
+
+		// prepare pubrec packet
+		pubrec := packet.NewPubrec()
+		pubrec.ID = publish.ID
+
+		// signal qos 2 pubrec
+		err = c.send(pubrec, true)
+		if err != nil {
+			return c.die(TransportError, err)
+		}
+	}
+
+	return nil
+}
+
+// handle an incoming p or pubcomp packet
+func (c *Client) processPubackAndPubcomp(id packet.ID) error {
+	// remove packet from store
+	err := c.session.DeletePacket(session.Outgoing, id)
+	if err != nil {
+		return c.die(SessionError, err)
+	}
+
+	// put back dequeue token
+	select {
+	case c.dequeueTokens <- struct{}{}:
+	default:
+		// continue if full for some reason
+	}
+
+	return nil
+}
+
+// handle an incoming pubrec packet
+func (c *Client) processPubrec(id packet.ID) error {
+	// allocate packet
+	pubrel := packet.NewPubrel()
+	pubrel.ID = id
+
+	// overwrite stored publish with the pubrel packet
+	err := c.session.SavePacket(session.Outgoing, pubrel)
+	if err != nil {
+		return c.die(SessionError, err)
+	}
+
+	// send packet
+	err = c.send(pubrel, true)
+	if err != nil {
+		return c.die(TransportError, err)
+	}
+
+	return nil
+}
+
+// handle an incoming pubrel packet
+func (c *Client) processPubrel(id packet.ID) error {
+	// get stored publish packet from session
+	pkt, err := c.session.LookupPacket(session.Incoming, id)
+	if err != nil {
+		return c.die(SessionError, err)
+	}
+
+	// prepare pubcomp packet
+	pubcomp := packet.NewPubcomp()
+	pubcomp.ID = id
+
+	// get packet from store
+	publish, ok := pkt.(*packet.Publish)
+	if !ok {
+		// immediately send pubcomp for missing packets
+		err = c.send(pubcomp, true)
+		if err != nil {
+			return c.die(TransportError, err)
+		}
+
+		return nil
+	}
+
+	// publish message and queue pubcomp if ack is called
+	err = c.backend.Publish(c, &publish.Message, func() {
+		c.backend.Log(MessageAcknowledged, c, nil, &publish.Message, nil)
+
+		select {
+		case c.ackQueue <- pubcomp:
+		case <-c.tomb.Dying():
+		}
+	})
+	if err != nil {
+		return c.die(BackendError, err)
+	}
+
+	c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)
+
+	return nil
+}
+
+// handle an incoming disconnect packet
+func (c *Client) processDisconnect() error {
+	// clear will
+	c.will = nil
+
+	// mark client as cleanly disconnected
+	atomic.StoreUint32(&c.state, clientDisconnected)
+
+	// close underlying connection (triggers cleanup)
+	c.conn.Close()
+
+	c.backend.Log(ClientDisconnected, c, nil, nil, nil)
+
+	return ErrClientDisconnected
+}
+
+/* helpers */
+
+// send a packet
+func (c *Client) send(pkt packet.Generic, async bool) error {
+	// send packet
+	err := c.conn.Send(pkt, async)
+	if err != nil {
+		return err
+	}
+
+	c.backend.Log(PacketSent, c, pkt, nil, nil)
+
+	return nil
+}
+
+/* error handling and logging */
+
+// used for closing and cleaning up from internal goroutines
+func (c *Client) die(event LogEvent, err error) error {
+	// log error
+	c.backend.Log(event, c, nil, nil, err)
+
+	// close connection if requested
+	c.conn.Close()
+
+	return err
+}
+
+// will try to cleanup as many resources as possible
+func (c *Client) cleanup() {
+	// check if not cleanly connected and will is present
+	if atomic.LoadUint32(&c.state) == clientConnected && c.will != nil {
+		// publish message
+		err := c.backend.Publish(c, c.will, nil)
+		if err != nil {
+			c.backend.Log(BackendError, c, nil, nil, err)
+		}
+
+		c.backend.Log(MessagePublished, c, nil, c.will, nil)
+	}
+
+	// remove client from the queue
+	if atomic.LoadUint32(&c.state) >= clientConnected {
+		err := c.backend.Terminate(c)
+		if err != nil {
+			c.backend.Log(BackendError, c, nil, nil, err)
+		}
+	}
+
+	c.backend.Log(LostConnection, c, nil, nil, nil)
+}
diff --git a/vendor/github.com/256dpi/gomqtt/broker/engine.go b/vendor/github.com/256dpi/gomqtt/broker/engine.go
new file mode 100644
index 000000000..90eb2a324
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/broker/engine.go
@@ -0,0 +1,148 @@
+// Package broker implements an extensible MQTT broker.
+package broker
+
+import (
+	"net"
+	"sync"
+	"time"
+
+	"github.com/256dpi/gomqtt/transport"
+
+	"gopkg.in/tomb.v2"
+)
+
+// The Engine handles incoming connections and connects them to the backend.
+type Engine struct {
+	// The Backend that will be passed to accepted clients.
+	Backend Backend
+
+	// ConnectTimeout defines the timeout to receive the first packet.
+	ConnectTimeout time.Duration
+
+	// The DefaultReadLimit defines the initial read limit.
+	DefaultReadLimit int64
+
+	// OnError can be used to receive errors from engine. If an error is received
+	// the server should be restarted.
+	OnError func(error)
+
+	mutex sync.Mutex
+	tomb  tomb.Tomb
+}
+
+// NewEngine returns a new Engine.
+func NewEngine(backend Backend) *Engine {
+	return &Engine{
+		Backend:        backend,
+		ConnectTimeout: 10 * time.Second,
+	}
+}
+
+// Accept begins accepting connections from the passed server.
+func (e *Engine) Accept(server transport.Server) {
+	e.tomb.Go(func() error {
+		for {
+			// return if dying
+			if !e.tomb.Alive() {
+				return tomb.ErrDying
+			}
+
+			// accept next connection
+			conn, err := server.Accept()
+			if err != nil {
+				// call error callback if available
+				if e.OnError != nil {
+					e.OnError(err)
+				}
+
+				return err
+			}
+
+			// handle connection
+			if !e.Handle(conn) {
+				return nil
+			}
+		}
+	})
+}
+
+// Handle takes over responsibility and handles a transport.Conn. It returns
+// false if the engine is closing and the connection has been closed.
+func (e *Engine) Handle(conn transport.Conn) bool {
+	// check conn
+	if conn == nil {
+		panic("passed conn is nil")
+	}
+
+	// acquire mutex
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
+
+	// close conn immediately when dying
+	if !e.tomb.Alive() {
+		conn.Close()
+		return false
+	}
+
+	// set default read limit
+	conn.SetReadLimit(e.DefaultReadLimit)
+
+	// set initial read timeout
+	conn.SetReadTimeout(e.ConnectTimeout)
+
+	// handle client
+	NewClient(e.Backend, conn)
+
+	return true
+}
+
+// Close will stop handling incoming connections and close all acceptors. The
+// call will block until all acceptors returned.
+//
+// Note: All passed servers to Accept must be closed before calling this method.
+func (e *Engine) Close() {
+	// acquire mutex
+	e.mutex.Lock()
+	defer e.mutex.Unlock()
+
+	// stop acceptors
+	e.tomb.Kill(nil)
+	e.tomb.Wait()
+}
+
+// Run runs the passed engine on a random available port and returns a channel
+// that can be closed to shutdown the engine. This method is intended to be used
+// in testing scenarios.
+func Run(engine *Engine, protocol string) (string, chan struct{}, chan struct{}) {
+	// launch server
+	server, err := transport.Launch(protocol + "://localhost:0")
+	if err != nil {
+		panic(err)
+	}
+
+	// prepare channels
+	quit := make(chan struct{})
+	done := make(chan struct{})
+
+	// start accepting connections
+	engine.Accept(server)
+
+	// prepare shutdown
+	go func() {
+		// wait for signal
+		<-quit
+
+		// errors from close are ignored
+		server.Close()
+
+		// close broker
+		engine.Close()
+
+		close(done)
+	}()
+
+	// get random port
+	_, port, _ := net.SplitHostPort(server.Addr().String())
+
+	return port, quit, done
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/connack.go b/vendor/github.com/256dpi/gomqtt/packet/connack.go
new file mode 100644
index 000000000..0fdf1a059
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/connack.go
@@ -0,0 +1,148 @@
+package packet
+
+import "fmt"
+
+// The ConnackCode represents the return code in a Connack packet.
+type ConnackCode uint8
+
+// All available ConnackCodes.
+const (
+	ConnectionAccepted ConnackCode = iota
+	InvalidProtocolVersion
+	IdentifierRejected
+	ServerUnavailable
+	BadUsernameOrPassword
+	NotAuthorized
+)
+
+// Valid checks if the ConnackCode is valid.
+func (cc ConnackCode) Valid() bool {
+	return cc <= 5
+}
+
+// String returns the corresponding error string for the ConnackCode.
+func (cc ConnackCode) String() string {
+	switch cc {
+	case ConnectionAccepted:
+		return "connection accepted"
+	case InvalidProtocolVersion:
+		return "connection refused: unacceptable protocol version"
+	case IdentifierRejected:
+		return "connection refused: identifier rejected"
+	case ServerUnavailable:
+		return "connection refused: server unavailable"
+	case BadUsernameOrPassword:
+		return "connection refused: bad user name or password"
+	case NotAuthorized:
+		return "connection refused: not authorized"
+	}
+
+	return "invalid connack code"
+}
+
+// A Connack packet is sent by the server in response to a Connect packet
+// received from a client.
+type Connack struct {
+	// The SessionPresent flag enables a client to establish whether the
+	// client and server have a consistent view about whether there is already
+	// stored session state.
+	SessionPresent bool
+
+	// If a well formed Connect packet is received by the server, but the server
+	// is unable to process it for some reason, then the server should attempt
+	// to send a Connack containing a non-zero ReturnCode.
+	ReturnCode ConnackCode
+}
+
+// NewConnack creates a new Connack packet.
+func NewConnack() *Connack {
+	return &Connack{}
+}
+
+// Type returns the packets type.
+func (cp *Connack) Type() Type {
+	return CONNACK
+}
+
+// String returns a string representation of the packet.
+func (cp *Connack) String() string {
+	return fmt.Sprintf("<Connack SessionPresent=%t ReturnCode=%d>",
+		cp.SessionPresent, cp.ReturnCode)
+}
+
+// Len returns the byte length of the encoded packet.
+func (cp *Connack) Len() int {
+	return headerLen(2) + 2
+}
+
+// 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 (cp *Connack) Decode(src []byte) (int, error) {
+	total := 0
+
+	// decode header
+	hl, _, rl, err := headerDecode(src, CONNACK)
+	total += hl
+	if err != nil {
+		return total, err
+	}
+
+	// check remaining length
+	if rl != 2 {
+		return total, makeError(cp.Type(), "expected remaining length to be 2")
+	}
+
+	// read connack flags
+	connackFlags := src[total]
+	cp.SessionPresent = connackFlags&0x1 == 1
+	total++
+
+	// check flags
+	if connackFlags&254 != 0 {
+		return 0, makeError(cp.Type(), "bits 7-1 in acknowledge flags are not 0")
+	}
+
+	// read return code
+	cp.ReturnCode = ConnackCode(src[total])
+	total++
+
+	// check return code
+	if !cp.ReturnCode.Valid() {
+		return 0, makeError(cp.Type(), "invalid return code (%d)", cp.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 (cp *Connack) Encode(dst []byte) (int, error) {
+	total := 0
+
+	// encode header
+	n, err := headerEncode(dst[total:], 0, 2, cp.Len(), CONNACK)
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// set session present flag
+	if cp.SessionPresent {
+		dst[total] = 1 // 00000001
+	} else {
+		dst[total] = 0 // 00000000
+	}
+	total++
+
+	// check return code
+	if !cp.ReturnCode.Valid() {
+		return total, makeError(cp.Type(), "invalid return code (%d)", cp.ReturnCode)
+	}
+
+	// set return code
+	dst[total] = byte(cp.ReturnCode)
+	total++
+
+	return total, nil
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/connect.go b/vendor/github.com/256dpi/gomqtt/packet/connect.go
new file mode 100644
index 000000000..354c7e1e6
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/connect.go
@@ -0,0 +1,408 @@
+package packet
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+)
+
+// The supported MQTT versions.
+const (
+	Version311 byte = 4
+	Version31  byte = 3
+)
+
+var version311Name = []byte("MQTT")
+var version31Name = []byte("MQIsdp")
+
+// A Connect packet is sent by a client to the server after a network
+// connection has been established.
+type Connect struct {
+	// The clients client id.
+	ClientID string
+
+	// The keep alive value.
+	KeepAlive uint16
+
+	// The authentication username.
+	Username string
+
+	// The authentication password.
+	Password string
+
+	// The clean session flag.
+	CleanSession bool
+
+	// The will message.
+	Will *Message
+
+	// The MQTT version 3 or 4 (defaults to 4 when 0).
+	Version byte
+}
+
+// NewConnect creates a new Connect packet.
+func NewConnect() *Connect {
+	return &Connect{
+		CleanSession: true,
+		Version:      4,
+	}
+}
+
+// Type returns the packets type.
+func (cp *Connect) Type() Type {
+	return CONNECT
+}
+
+// String returns a string representation of the packet.
+func (cp *Connect) String() string {
+	will := "nil"
+
+	if cp.Will != nil {
+		will = cp.Will.String()
+	}
+
+	return fmt.Sprintf("<Connect ClientID=%q KeepAlive=%d Username=%q "+
+		"Password=%q CleanSession=%t Will=%s Version=%d>",
+		cp.ClientID,
+		cp.KeepAlive,
+		cp.Username,
+		cp.Password,
+		cp.CleanSession,
+		will,
+		cp.Version,
+	)
+}
+
+// Len returns the byte length of the encoded packet.
+func (cp *Connect) Len() int {
+	ml := cp.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 (cp *Connect) Decode(src []byte) (int, error) {
+	total := 0
+
+	// decode header
+	hl, _, _, err := headerDecode(src[total:], CONNECT)
+	total += hl
+	if err != nil {
+		return total, err
+	}
+
+	// read protocol string
+	protoName, n, err := readLPBytes(src[total:], false, cp.Type())
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// check buffer length
+	if len(src) < total+1 {
+		return total, makeError(cp.Type(), "insufficient buffer size, expected %d, got %d", total+1, len(src))
+	}
+
+	// read version
+	versionByte := src[total]
+	total++
+
+	// check protocol string and version
+	if versionByte != Version311 && versionByte != Version31 {
+		return total, makeError(cp.Type(), "invalid protocol version (%d)", versionByte)
+	}
+
+	// set version
+	cp.Version = versionByte
+
+	// check protocol version string
+	if !bytes.Equal(protoName, version311Name) && !bytes.Equal(protoName, version31Name) {
+		return total, makeError(cp.Type(), "invalid protocol version description (%s)", protoName)
+	}
+
+	// check buffer length
+	if len(src) < total+1 {
+		return total, makeError(cp.Type(), "insufficient buffer size, expected %d, got %d", total+1, len(src))
+	}
+
+	// read connect flags
+	connectFlags := src[total]
+	total++
+
+	// read flags
+	usernameFlag := ((connectFlags >> 7) & 0x1) == 1
+	passwordFlag := ((connectFlags >> 6) & 0x1) == 1
+	willFlag := ((connectFlags >> 2) & 0x1) == 1
+	willRetain := ((connectFlags >> 5) & 0x1) == 1
+	willQOS := QOS((connectFlags >> 3) & 0x3)
+	cp.CleanSession = ((connectFlags >> 1) & 0x1) == 1
+
+	// check reserved bit
+	if connectFlags&0x1 != 0 {
+		return total, makeError(cp.Type(), "reserved bit 0 is not 0")
+	}
+
+	// check will qos
+	if !willQOS.Successful() {
+		return total, makeError(cp.Type(), "invalid QOS level (%d) for will message", willQOS)
+	}
+
+	// check will flags
+	if !willFlag && (willRetain || willQOS != 0) {
+		return total, makeError(cp.Type(), "if the will flag is set to 0 the will qos and will retain fields must be set to zero")
+	}
+
+	// create will if present
+	if willFlag {
+		cp.Will = &Message{QOS: willQOS, Retain: willRetain}
+	}
+
+	// check auth flags
+	if !usernameFlag && passwordFlag {
+		return total, makeError(cp.Type(), "password flag is set but username flag is not set")
+	}
+
+	// check buffer length
+	if len(src) < total+2 {
+		return total, makeError(cp.Type(), "insufficient buffer size, expected %d, got %d", total+2, len(src))
+	}
+
+	// read keep alive
+	cp.KeepAlive = binary.BigEndian.Uint16(src[total:])
+	total += 2
+
+	// read client id
+	cp.ClientID, n, err = readLPString(src[total:], cp.Type())
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// if the client supplies a zero-byte clientID, the client must also set CleanSession to 1
+	if len(cp.ClientID) == 0 && !cp.CleanSession {
+		return total, makeError(cp.Type(), "clean session must be 1 if client id is zero length")
+	}
+
+	// read will topic and payload
+	if cp.Will != nil {
+		cp.Will.Topic, n, err = readLPString(src[total:], cp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+
+		cp.Will.Payload, n, err = readLPBytes(src[total:], true, cp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+
+	// read username
+	if usernameFlag {
+		cp.Username, n, err = readLPString(src[total:], cp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+
+	// read password
+	if passwordFlag {
+		cp.Password, n, err = readLPString(src[total:], cp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+
+	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 (cp *Connect) Encode(dst []byte) (int, error) {
+	total := 0
+
+	// encode header
+	n, err := headerEncode(dst[total:], 0, cp.len(), cp.Len(), CONNECT)
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// set default version byte
+	if cp.Version == 0 {
+		cp.Version = Version311
+	}
+
+	// check version byte
+	if cp.Version != Version311 && cp.Version != Version31 {
+		return total, makeError(cp.Type(), "unsupported protocol version %d", cp.Version)
+	}
+
+	// write version string, length has been checked beforehand
+	if cp.Version == Version311 {
+		n, _ = writeLPBytes(dst[total:], version311Name, cp.Type())
+		total += n
+	} else if cp.Version == Version31 {
+		n, _ = writeLPBytes(dst[total:], version31Name, cp.Type())
+		total += n
+	}
+
+	// write version value
+	dst[total] = cp.Version
+	total++
+
+	var connectFlags byte
+
+	// set username flag
+	if len(cp.Username) > 0 {
+		connectFlags |= 128 // 10000000
+	} else {
+		connectFlags &= 127 // 01111111
+	}
+
+	// set password flag
+	if len(cp.Password) > 0 {
+		connectFlags |= 64 // 01000000
+	} else {
+		connectFlags &= 191 // 10111111
+	}
+
+	// set will flag
+	if cp.Will != nil {
+		connectFlags |= 0x4 // 00000100
+
+		// check will topic length
+		if len(cp.Will.Topic) == 0 {
+			return total, makeError(cp.Type(), "will topic is empty")
+		}
+
+		// check will qos
+		if !cp.Will.QOS.Successful() {
+			return total, makeError(cp.Type(), "invalid will qos level %d", cp.Will.QOS)
+		}
+
+		// set will qos flag
+		connectFlags = (connectFlags & 231) | (byte(cp.Will.QOS) << 3) // 231 = 11100111
+
+		// set will retain flag
+		if cp.Will.Retain {
+			connectFlags |= 32 // 00100000
+		} else {
+			connectFlags &= 223 // 11011111
+		}
+
+	} else {
+		connectFlags &= 251 // 11111011
+	}
+
+	// check client id and clean session
+	if len(cp.ClientID) == 0 && !cp.CleanSession {
+		return total, makeError(cp.Type(), "clean session must be 1 if client id is zero length")
+	}
+
+	// set clean session flag
+	if cp.CleanSession {
+		connectFlags |= 0x2 // 00000010
+	} else {
+		connectFlags &= 253 // 11111101
+	}
+
+	// write connect flags
+	dst[total] = connectFlags
+	total++
+
+	// write keep alive
+	binary.BigEndian.PutUint16(dst[total:], cp.KeepAlive)
+	total += 2
+
+	// write client id
+	n, err = writeLPString(dst[total:], cp.ClientID, cp.Type())
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// write will topic and payload
+	if cp.Will != nil {
+		n, err = writeLPString(dst[total:], cp.Will.Topic, cp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+
+		n, err = writeLPBytes(dst[total:], cp.Will.Payload, cp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+
+	if len(cp.Username) == 0 && len(cp.Password) > 0 {
+		return total, makeError(cp.Type(), "password set without username")
+	}
+
+	// write username
+	if len(cp.Username) > 0 {
+		n, err = writeLPString(dst[total:], cp.Username, cp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+
+	// write password
+	if len(cp.Password) > 0 {
+		n, err = writeLPString(dst[total:], cp.Password, cp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+
+	return total, nil
+}
+
+// Returns the payload length.
+func (cp *Connect) len() int {
+	total := 0
+
+	if cp.Version == Version31 {
+		// 2 bytes protocol name length
+		// 6 bytes protocol name
+		// 1 byte protocol version
+		total += 2 + 6 + 1
+	} else {
+		// 2 bytes protocol name length
+		// 4 bytes protocol name
+		// 1 byte protocol version
+		total += 2 + 4 + 1
+	}
+
+	// 1 byte connect flags
+	// 2 bytes keep alive timer
+	total += 1 + 2
+
+	// add the clientID length
+	total += 2 + len(cp.ClientID)
+
+	// add the will topic and will message length
+	if cp.Will != nil {
+		total += 2 + len(cp.Will.Topic) + 2 + len(cp.Will.Payload)
+	}
+
+	// add the username length
+	if len(cp.Username) > 0 {
+		total += 2 + len(cp.Username)
+	}
+
+	// add the password length
+	if len(cp.Password) > 0 {
+		total += 2 + len(cp.Password)
+	}
+
+	return total
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/error.go b/vendor/github.com/256dpi/gomqtt/packet/error.go
new file mode 100644
index 000000000..71ea255d9
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/error.go
@@ -0,0 +1,20 @@
+package packet
+
+import "fmt"
+
+// Error represents decoding and encoding errors.
+type Error struct {
+	Type Type
+
+	format    string
+	arguments []interface{}
+}
+
+func makeError(typ Type, format string, arguments ...interface{}) *Error {
+	return &Error{Type: typ, format: format, arguments: arguments}
+}
+
+// Error implements the error interface.
+func (e *Error) Error() string {
+	return fmt.Sprintf(e.format, e.arguments...)
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/header.go b/vendor/github.com/256dpi/gomqtt/packet/header.go
new file mode 100644
index 000000000..f176c1a2d
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/header.go
@@ -0,0 +1,98 @@
+package packet
+
+import (
+	"encoding/binary"
+)
+
+const maxRemainingLength = 268435455 // 256 MB
+
+func headerLen(rl int) int {
+	// packet type and flag byte
+	total := 1
+
+	if rl <= 127 {
+		total++
+	} else if rl <= 16383 {
+		total += 2
+	} else if rl <= 2097151 {
+		total += 3
+	} else {
+		total += 4
+	}
+
+	return total
+}
+
+func headerEncode(dst []byte, flags byte, rl int, tl int, t Type) (int, error) {
+	total := 0
+
+	// check buffer length
+	if len(dst) < tl {
+		return total, makeError(t, "insufficient buffer size, expected %d, got %d", tl, len(dst))
+	}
+
+	// check remaining length
+	if rl > maxRemainingLength || rl < 0 {
+		return total, makeError(t, "remaining length (%d) out of bound (max %d, min 0)", rl, maxRemainingLength)
+	}
+
+	// check header length
+	hl := headerLen(rl)
+	if len(dst) < hl {
+		return total, makeError(t, "insufficient buffer size, expected %d, got %d", hl, len(dst))
+	}
+
+	// write type and flags
+	typeAndFlags := byte(t)<<4 | (t.defaultFlags() & 0xf)
+	typeAndFlags |= flags
+	dst[total] = typeAndFlags
+	total++
+
+	// write remaining length
+	n := binary.PutUvarint(dst[total:], uint64(rl))
+	total += n
+
+	return total, nil
+}
+
+func headerDecode(src []byte, t Type) (int, byte, int, error) {
+	total := 0
+
+	// check buffer size
+	if len(src) < 2 {
+		return total, 0, 0, makeError(t, "insufficient buffer size, expected %d, got %d", 2, len(src))
+	}
+
+	// read type and flags
+	typeAndFlags := src[total : total+1]
+	decodedType := Type(typeAndFlags[0] >> 4)
+	flags := typeAndFlags[0] & 0x0f
+	total++
+
+	// 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, m := binary.Uvarint(src[total:])
+	rl := int(_rl)
+	total += m
+
+	// check resulting remaining length
+	if m <= 0 {
+		return total, 0, 0, makeError(t, "error reading remaining length")
+	}
+
+	// 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
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/identified.go b/vendor/github.com/256dpi/gomqtt/packet/identified.go
new file mode 100644
index 000000000..564ce0e6d
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/identified.go
@@ -0,0 +1,275 @@
+package packet
+
+import (
+	"encoding/binary"
+	"fmt"
+)
+
+// returns the byte length of an identified packet
+func identifiedLen() int {
+	return headerLen(2) + 2
+}
+
+// decodes an identified packet
+func identifiedDecode(src []byte, t Type) (int, ID, error) {
+	total := 0
+
+	// decode header
+	hl, _, rl, err := headerDecode(src, t)
+	total += hl
+	if err != nil {
+		return total, 0, err
+	}
+
+	// check remaining length
+	if rl != 2 {
+		return total, 0, makeError(t, "expected remaining length to be 2")
+	}
+
+	// read packet id
+	packetID := ID(binary.BigEndian.Uint16(src[total:]))
+	total += 2
+
+	// check packet id
+	if !packetID.Valid() {
+		return total, 0, makeError(t, "packet id must be grater than zero")
+	}
+
+	return total, packetID, nil
+}
+
+// encodes an identified packet
+func identifiedEncode(dst []byte, id ID, t Type) (int, error) {
+	total := 0
+
+	// check packet id
+	if !id.Valid() {
+		return total, makeError(t, "packet id must be grater than zero")
+	}
+
+	// encode header
+	n, err := headerEncode(dst[total:], 0, 2, identifiedLen(), t)
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// write packet id
+	binary.BigEndian.PutUint16(dst[total:], uint16(id))
+	total += 2
+
+	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 (pp *Puback) Type() Type {
+	return PUBACK
+}
+
+// Len returns the byte length of the encoded packet.
+func (pp *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 (pp *Puback) Decode(src []byte) (int, error) {
+	n, pid, err := identifiedDecode(src, PUBACK)
+	pp.ID = pid
+	return n, err
+}
+
+// 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 (pp *Puback) Encode(dst []byte) (int, error) {
+	return identifiedEncode(dst, pp.ID, PUBACK)
+}
+
+// String returns a string representation of the packet.
+func (pp *Puback) String() string {
+	return fmt.Sprintf("<Puback ID=%d>", pp.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
+}
+
+var _ Generic = (*Pubcomp)(nil)
+
+// NewPubcomp creates a new Pubcomp packet.
+func NewPubcomp() *Pubcomp {
+	return &Pubcomp{}
+}
+
+// Type returns the packets type.
+func (pp *Pubcomp) Type() Type {
+	return PUBCOMP
+}
+
+// Len returns the byte length of the encoded packet.
+func (pp *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 (pp *Pubcomp) Decode(src []byte) (int, error) {
+	n, pid, err := identifiedDecode(src, PUBCOMP)
+	pp.ID = pid
+	return n, err
+}
+
+// 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 (pp *Pubcomp) Encode(dst []byte) (int, error) {
+	return identifiedEncode(dst, pp.ID, PUBCOMP)
+}
+
+// String returns a string representation of the packet.
+func (pp *Pubcomp) String() string {
+	return fmt.Sprintf("<Pubcomp ID=%d>", pp.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 (pp *Pubrec) Type() Type {
+	return PUBREC
+}
+
+// Len returns the byte length of the encoded packet.
+func (pp *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 (pp *Pubrec) Decode(src []byte) (int, error) {
+	n, pid, err := identifiedDecode(src, PUBREC)
+	pp.ID = pid
+	return n, err
+}
+
+// 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 (pp *Pubrec) Encode(dst []byte) (int, error) {
+	return identifiedEncode(dst, pp.ID, PUBREC)
+}
+
+// String returns a string representation of the packet.
+func (pp *Pubrec) String() string {
+	return fmt.Sprintf("<Pubrec ID=%d>", pp.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
+}
+
+var _ Generic = (*Pubrel)(nil)
+
+// NewPubrel creates a new Pubrel packet.
+func NewPubrel() *Pubrel {
+	return &Pubrel{}
+}
+
+// Type returns the packets type.
+func (pp *Pubrel) Type() Type {
+	return PUBREL
+}
+
+// Len returns the byte length of the encoded packet.
+func (pp *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 (pp *Pubrel) Decode(src []byte) (int, error) {
+	n, pid, err := identifiedDecode(src, PUBREL)
+	pp.ID = pid
+	return n, err
+}
+
+// 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 (pp *Pubrel) Encode(dst []byte) (int, error) {
+	return identifiedEncode(dst, pp.ID, PUBREL)
+}
+
+// String returns a string representation of the packet.
+func (pp *Pubrel) String() string {
+	return fmt.Sprintf("<Pubrel ID=%d>", pp.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 (up *Unsuback) Type() Type {
+	return UNSUBACK
+}
+
+// Len returns the byte length of the encoded packet.
+func (up *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 (up *Unsuback) Decode(src []byte) (int, error) {
+	n, pid, err := identifiedDecode(src, UNSUBACK)
+	up.ID = pid
+	return n, err
+}
+
+// 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 (up *Unsuback) Encode(dst []byte) (int, error) {
+	return identifiedEncode(dst, up.ID, UNSUBACK)
+}
+
+// String returns a string representation of the packet.
+func (up *Unsuback) String() string {
+	return fmt.Sprintf("<Unsuback ID=%d>", up.ID)
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/message.go b/vendor/github.com/256dpi/gomqtt/packet/message.go
new file mode 100644
index 000000000..589187f8b
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/message.go
@@ -0,0 +1,31 @@
+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=%v>",
+		m.Topic, m.QOS, m.Retain, m.Payload)
+}
+
+// Copy returns a copy of the message.
+func (m Message) Copy() *Message {
+	return &m
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/naked.go b/vendor/github.com/256dpi/gomqtt/packet/naked.go
new file mode 100644
index 000000000..8a0843528
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/naked.go
@@ -0,0 +1,137 @@
+package packet
+
+// returns the byte length of a naked packet
+func nakedLen() int {
+	return headerLen(0)
+}
+
+// decodes a naked packet
+func nakedDecode(src []byte, t Type) (int, error) {
+	// decode header
+	hl, _, rl, err := headerDecode(src, t)
+
+	// check remaining length
+	if rl != 0 {
+		return hl, makeError(t, "expected zero remaining length")
+	}
+
+	return hl, err
+}
+
+// encodes a naked packet
+func nakedEncode(dst []byte, t Type) (int, error) {
+	// encode header
+	return headerEncode(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 (dp *Disconnect) Type() Type {
+	return DISCONNECT
+}
+
+// Len returns the byte length of the encoded packet.
+func (dp *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 (dp *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 (dp *Disconnect) Encode(dst []byte) (int, error) {
+	return nakedEncode(dst, DISCONNECT)
+}
+
+// String returns a string representation of the packet.
+func (dp *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 (pp *Pingreq) Type() Type {
+	return PINGREQ
+}
+
+// Len returns the byte length of the encoded packet.
+func (pp *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 (pp *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 (pp *Pingreq) Encode(dst []byte) (int, error) {
+	return nakedEncode(dst, PINGREQ)
+}
+
+// String returns a string representation of the packet.
+func (pp *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{}
+
+var _ Generic = (*Pingresp)(nil)
+
+// NewPingresp creates a new Pingresp packet.
+func NewPingresp() *Pingresp {
+	return &Pingresp{}
+}
+
+// Type returns the packets type.
+func (pp *Pingresp) Type() Type {
+	return PINGRESP
+}
+
+// Len returns the byte length of the encoded packet.
+func (pp *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 (pp *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 (pp *Pingresp) Encode(dst []byte) (int, error) {
+	return nakedEncode(dst, PINGRESP)
+}
+
+// String returns a string representation of the packet.
+func (pp *Pingresp) String() string {
+	return "<Pingresp>"
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/packet.go b/vendor/github.com/256dpi/gomqtt/packet/packet.go
new file mode 100644
index 000000000..5d0305c56
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/packet.go
@@ -0,0 +1,138 @@
+// 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
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/publish.go b/vendor/github.com/256dpi/gomqtt/packet/publish.go
new file mode 100644
index 000000000..ffde2dd69
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/publish.go
@@ -0,0 +1,186 @@
+package packet
+
+import (
+	"encoding/binary"
+	"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 (pp *Publish) Type() Type {
+	return PUBLISH
+}
+
+// String returns a string representation of the packet.
+func (pp *Publish) String() string {
+	return fmt.Sprintf("<Publish ID=%d Message=%s Dup=%t>",
+		pp.ID, pp.Message.String(), pp.Dup)
+}
+
+// Len returns the byte length of the encoded packet.
+func (pp *Publish) Len() int {
+	ml := pp.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 (pp *Publish) Decode(src []byte) (int, error) {
+	total := 0
+
+	// decode header
+	hl, flags, rl, err := headerDecode(src[total:], PUBLISH)
+	total += hl
+	if err != nil {
+		return total, err
+	}
+
+	// read flags
+	pp.Dup = ((flags >> 3) & 0x1) == 1
+	pp.Message.Retain = (flags & 0x1) == 1
+	pp.Message.QOS = QOS((flags >> 1) & 0x3)
+
+	// check qos
+	if !pp.Message.QOS.Successful() {
+		return total, makeError(pp.Type(), "invalid QOS level (%d)", pp.Message.QOS)
+	}
+
+	// check buffer length
+	if len(src) < total+2 {
+		return total, makeError(pp.Type(), "insufficient buffer size, expected %d, got %d", total+2, len(src))
+	}
+
+	n := 0
+
+	// read topic
+	pp.Message.Topic, n, err = readLPString(src[total:], pp.Type())
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	if pp.Message.QOS != 0 {
+		// check buffer length
+		if len(src) < total+2 {
+			return total, makeError(pp.Type(), "insufficient buffer size, expected %d, got %d", total+2, len(src))
+		}
+
+		// read packet id
+		pp.ID = ID(binary.BigEndian.Uint16(src[total:]))
+		total += 2
+
+		// check packet id
+		if !pp.ID.Valid() {
+			return total, makeError(pp.Type(), "packet id must be grater than zero")
+		}
+	}
+
+	// calculate payload length
+	l := int(rl) - (total - hl)
+
+	// read payload
+	if l > 0 {
+		pp.Message.Payload = make([]byte, l)
+		copy(pp.Message.Payload, src[total:total+l])
+		total += len(pp.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 (pp *Publish) Encode(dst []byte) (int, error) {
+	total := 0
+
+	// check topic length
+	if len(pp.Message.Topic) == 0 {
+		return total, makeError(pp.Type(), "topic name is empty")
+	}
+
+	flags := byte(0)
+
+	// set dup flag
+	if pp.Dup {
+		flags |= 0x8 // 00001000
+	} else {
+		flags &= 247 // 11110111
+	}
+
+	// set retain flag
+	if pp.Message.Retain {
+		flags |= 0x1 // 00000001
+	} else {
+		flags &= 254 // 11111110
+	}
+
+	// check qos
+	if !pp.Message.QOS.Successful() {
+		return 0, makeError(pp.Type(), "invalid QOS level %d", pp.Message.QOS)
+	}
+
+	// check packet id
+	if pp.Message.QOS > 0 && !pp.ID.Valid() {
+		return total, makeError(pp.Type(), "packet id must be grater than zero")
+	}
+
+	// set qos
+	flags = (flags & 249) | (byte(pp.Message.QOS) << 1) // 249 = 11111001
+
+	// encode header
+	n, err := headerEncode(dst[total:], flags, pp.len(), pp.Len(), PUBLISH)
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// write topic
+	n, err = writeLPString(dst[total:], pp.Message.Topic, pp.Type())
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// write packet id
+	if pp.Message.QOS != 0 {
+		binary.BigEndian.PutUint16(dst[total:], uint16(pp.ID))
+		total += 2
+	}
+
+	// write payload
+	copy(dst[total:], pp.Message.Payload)
+	total += len(pp.Message.Payload)
+
+	return total, nil
+}
+
+// Returns the payload length.
+func (pp *Publish) len() int {
+	total := 2 + len(pp.Message.Topic) + len(pp.Message.Payload)
+	if pp.Message.QOS != 0 {
+		total += 2
+	}
+
+	return total
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/stream.go b/vendor/github.com/256dpi/gomqtt/packet/stream.go
new file mode 100644
index 000000000..25d8b2139
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/stream.go
@@ -0,0 +1,155 @@
+package packet
+
+import (
+	"bufio"
+	"bytes"
+	"errors"
+	"io"
+	"time"
+
+	"github.com/256dpi/mercury"
+)
+
+// ErrDetectionOverflow is returned by the Decoder if the next packet couldn't
+// be detect from the initial header bytes.
+var ErrDetectionOverflow = errors.New("detection overflow")
+
+// ErrReadLimitExceeded can be returned during a Receive if the connection
+// exceeded its read limit.
+var ErrReadLimitExceeded = errors.New("read limit exceeded")
+
+// An Encoder wraps a Writer and continuously encodes packets.
+type Encoder struct {
+	writer *mercury.Writer
+	buffer bytes.Buffer
+}
+
+// NewEncoder creates a new Encoder.
+func NewEncoder(writer io.Writer) *Encoder {
+	return &Encoder{
+		writer: mercury.NewWriter(writer, time.Millisecond),
+	}
+}
+
+// Write encodes and writes the passed packet to the write buffer.
+func (e *Encoder) Write(pkt Generic, async bool) error {
+	// reset and eventually grow buffer
+	packetLength := pkt.Len()
+	e.buffer.Reset()
+	e.buffer.Grow(packetLength)
+	buf := e.buffer.Bytes()[0:packetLength]
+
+	// encode packet
+	_, err := pkt.Encode(buf)
+	if err != nil {
+		return err
+	}
+
+	// write buffer
+	if async {
+		_, err = e.writer.Write(buf)
+	} else {
+		_, err = e.writer.WriteAndFlush(buf)
+	}
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Flush flushes the writer buffer.
+func (e *Encoder) Flush() error {
+	return e.writer.Flush()
+}
+
+// A Decoder wraps a Reader and continuously decodes packets.
+type Decoder struct {
+	Limit int64
+
+	reader *bufio.Reader
+	buffer bytes.Buffer
+}
+
+// NewDecoder returns a new Decoder.
+func NewDecoder(reader io.Reader) *Decoder {
+	return &Decoder{
+		reader: bufio.NewReader(reader),
+	}
+}
+
+// Read reads the next packet from the buffered reader.
+func (d *Decoder) Read() (Generic, error) {
+	// initial detection length
+	detectionLength := 2
+
+	for {
+		// check length
+		if detectionLength > 5 {
+			return nil, ErrDetectionOverflow
+		}
+
+		// try read detection bytes
+		header, err := d.reader.Peek(detectionLength)
+		if err == io.EOF && len(header) != 0 {
+			// an EOF with some data is unexpected
+			return nil, io.ErrUnexpectedEOF
+		} else if err != nil {
+			return nil, err
+		}
+
+		// detect packet
+		packetLength, packetType := DetectPacket(header)
+
+		// on zero packet length:
+		// increment detection length and try again
+		if packetLength <= 0 {
+			detectionLength++
+			continue
+		}
+
+		// check read limit
+		if d.Limit > 0 && int64(packetLength) > d.Limit {
+			return nil, ErrReadLimitExceeded
+		}
+
+		// create packet
+		pkt, err := packetType.New()
+		if err != nil {
+			return nil, err
+		}
+
+		// reset and eventually grow buffer
+		d.buffer.Reset()
+		d.buffer.Grow(packetLength)
+		buf := d.buffer.Bytes()[0:packetLength]
+
+		// read whole packet (will not return EOF)
+		_, err = io.ReadFull(d.reader, buf)
+		if err != nil {
+			return nil, err
+		}
+
+		// decode buffer
+		_, err = pkt.Decode(buf)
+		if err != nil {
+			return nil, err
+		}
+
+		return pkt, nil
+	}
+}
+
+// A Stream combines an Encoder and Decoder
+type Stream struct {
+	*Decoder
+	*Encoder
+}
+
+// NewStream creates a new Stream.
+func NewStream(reader io.Reader, writer io.Writer) *Stream {
+	return &Stream{
+		Decoder: NewDecoder(reader),
+		Encoder: NewEncoder(writer),
+	}
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/strings.go b/vendor/github.com/256dpi/gomqtt/packet/strings.go
new file mode 100644
index 000000000..4b8764435
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/strings.go
@@ -0,0 +1,78 @@
+package packet
+
+import (
+	"encoding/binary"
+)
+
+const maxLPLength uint16 = 65535
+
+// read length prefixed bytes
+func readLPBytes(buf []byte, safe bool, t Type) ([]byte, int, error) {
+	if len(buf) < 2 {
+		return nil, 0, makeError(t, "insufficient buffer size, expected 2, got %d", len(buf))
+	}
+
+	n, total := 0, 0
+
+	n = int(binary.BigEndian.Uint16(buf))
+	total += 2
+	total += n
+
+	if len(buf) < total {
+		return nil, total, makeError(t, "insufficient buffer size, expected %d, got %d", total, len(buf))
+	}
+
+	// copy buffer in safe mode
+	if safe {
+		newBuf := make([]byte, total-2)
+		copy(newBuf, buf[2:total])
+		return newBuf, total, nil
+	}
+
+	return buf[2:total], total, nil
+}
+
+// read length prefixed string
+func readLPString(buf []byte, t Type) (string, int, error) {
+	if len(buf) < 2 {
+		return "", 0, makeError(t, "insufficient buffer size, expected 2, got %d", len(buf))
+	}
+
+	n, total := 0, 0
+
+	n = int(binary.BigEndian.Uint16(buf))
+	total += 2
+	total += n
+
+	if len(buf) < total {
+		return "", total, makeError(t, "insufficient buffer size, expected %d, got %d", total, len(buf))
+	}
+
+	return string(buf[2:total]), total, nil
+}
+
+// write length prefixed bytes
+func writeLPBytes(buf []byte, b []byte, t Type) (int, error) {
+	total, n := 0, len(b)
+
+	if n > int(maxLPLength) {
+		return 0, makeError(t, "length (%d) greater than %d bytes", n, maxLPLength)
+	}
+
+	if len(buf) < 2+n {
+		return 0, makeError(t, "insufficient buffer size, expected %d, got %d", 2+n, len(buf))
+	}
+
+	binary.BigEndian.PutUint16(buf, uint16(n))
+	total += 2
+
+	copy(buf[total:], b)
+	total += n
+
+	return total, nil
+}
+
+// write length prefixed string
+func writeLPString(buf []byte, str string, t Type) (int, error) {
+	return writeLPBytes(buf, []byte(str), t)
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/suback.go b/vendor/github.com/256dpi/gomqtt/packet/suback.go
new file mode 100644
index 000000000..dd27f64de
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/suback.go
@@ -0,0 +1,140 @@
+package packet
+
+import (
+	"encoding/binary"
+	"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 (sp *Suback) Type() Type {
+	return SUBACK
+}
+
+// String returns a string representation of the packet.
+func (sp *Suback) String() string {
+	var codes []string
+
+	for _, c := range sp.ReturnCodes {
+		codes = append(codes, fmt.Sprintf("%d", c))
+	}
+
+	return fmt.Sprintf("<Suback ID=%d ReturnCodes=[%s]>",
+		sp.ID, strings.Join(codes, ", "))
+}
+
+// Len returns the byte length of the encoded packet.
+func (sp *Suback) Len() int {
+	ml := sp.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 (sp *Suback) Decode(src []byte) (int, error) {
+	total := 0
+
+	// decode header
+	hl, _, rl, err := headerDecode(src[total:], SUBACK)
+	total += hl
+	if err != nil {
+		return total, err
+	}
+
+	// check buffer length
+	if len(src) < total+2 {
+		return total, makeError(sp.Type(), "insufficient buffer size, expected %d, got %d", total+2, len(src))
+	}
+
+	// check remaining length
+	if rl <= 2 {
+		return total, makeError(sp.Type(), "expected remaining length to be greater than 2, got %d", rl)
+	}
+
+	// read packet id
+	sp.ID = ID(binary.BigEndian.Uint16(src[total:]))
+	total += 2
+
+	// check packet id
+	if !sp.ID.Valid() {
+		return total, makeError(sp.Type(), "packet id must be grater than zero")
+	}
+
+	// calculate number of return codes
+	rcl := int(rl) - 2
+
+	// read return codes
+	sp.ReturnCodes = make([]QOS, rcl)
+	for i, rc := range src[total : total+rcl] {
+		sp.ReturnCodes[i] = QOS(rc)
+	}
+	total += len(sp.ReturnCodes)
+
+	// validate return codes
+	for i, code := range sp.ReturnCodes {
+		if !code.Successful() && code != QOSFailure {
+			return total, makeError(sp.Type(), "invalid return code %d for topic %d", code, i)
+		}
+	}
+
+	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 (sp *Suback) Encode(dst []byte) (int, error) {
+	total := 0
+
+	// check return codes
+	for i, code := range sp.ReturnCodes {
+		if !code.Successful() && code != QOSFailure {
+			return total, makeError(sp.Type(), "invalid return code %d for topic %d", code, i)
+		}
+	}
+
+	// check packet id
+	if !sp.ID.Valid() {
+		return total, makeError(sp.Type(), "packet id must be grater than zero")
+	}
+
+	// encode header
+	n, err := headerEncode(dst[total:], 0, sp.len(), sp.Len(), SUBACK)
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// write packet id
+	binary.BigEndian.PutUint16(dst[total:], uint16(sp.ID))
+	total += 2
+
+	// write return codes
+	for i, rc := range sp.ReturnCodes {
+		dst[total+i] = byte(rc)
+	}
+	total += len(sp.ReturnCodes)
+
+	return total, nil
+}
+
+// Returns the payload length.
+func (sp *Suback) len() int {
+	return 2 + len(sp.ReturnCodes)
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/subscribe.go b/vendor/github.com/256dpi/gomqtt/packet/subscribe.go
new file mode 100644
index 000000000..496161940
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/subscribe.go
@@ -0,0 +1,182 @@
+package packet
+
+import (
+	"encoding/binary"
+	"fmt"
+	"strings"
+)
+
+// A Subscription is a single subscription in a Subscribe packet.
+type Subscription struct {
+	// The topic to subscribe.
+	Topic string
+
+	// The requested maximum QOS level.
+	QOS QOS
+}
+
+func (s *Subscription) String() string {
+	return fmt.Sprintf("%q=>%d", s.Topic, s.QOS)
+}
+
+// A Subscribe packet is sent from the client to the server to create one or
+// more Subscriptions. The server will forward application messages that match
+// these subscriptions using PublishPackets.
+type Subscribe struct {
+	// The subscriptions.
+	Subscriptions []Subscription
+
+	// The packet identifier.
+	ID ID
+}
+
+// NewSubscribe creates a new Subscribe packet.
+func NewSubscribe() *Subscribe {
+	return &Subscribe{}
+}
+
+// Type returns the packets type.
+func (sp *Subscribe) Type() Type {
+	return SUBSCRIBE
+}
+
+// String returns a string representation of the packet.
+func (sp *Subscribe) String() string {
+	var subscriptions []string
+
+	for _, t := range sp.Subscriptions {
+		subscriptions = append(subscriptions, t.String())
+	}
+
+	return fmt.Sprintf("<Subscribe ID=%d Subscriptions=[%s]>",
+		sp.ID, strings.Join(subscriptions, ", "))
+}
+
+// Len returns the byte length of the encoded packet.
+func (sp *Subscribe) Len() int {
+	ml := sp.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 (sp *Subscribe) Decode(src []byte) (int, error) {
+	total := 0
+
+	// decode header
+	hl, _, rl, err := headerDecode(src[total:], SUBSCRIBE)
+	total += hl
+	if err != nil {
+		return total, err
+	}
+
+	// check buffer length
+	if len(src) < total+2 {
+		return total, makeError(sp.Type(), "insufficient buffer size, expected %d, got %d", total+2, len(src))
+	}
+
+	// read packet id
+	sp.ID = ID(binary.BigEndian.Uint16(src[total:]))
+	total += 2
+
+	// check packet id
+	if !sp.ID.Valid() {
+		return total, makeError(sp.Type(), "packet id must be grater than zero")
+	}
+
+	// reset subscriptions
+	sp.Subscriptions = sp.Subscriptions[:0]
+
+	// calculate number of subscriptions
+	sl := int(rl) - 2
+
+	for sl > 0 {
+		// read topic
+		t, n, err := readLPString(src[total:], sp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+
+		// check buffer length
+		if len(src) < total+1 {
+			return total, makeError(sp.Type(), "insufficient buffer size, expected %d, got %d", total+1, len(src))
+		}
+
+		// read qos
+		qos := QOS(src[total])
+		if !qos.Successful() {
+			return total, makeError(sp.Type(), "invalid QOS level (%d)", qos)
+		}
+
+		// read qos and add subscription
+		sp.Subscriptions = append(sp.Subscriptions, Subscription{t, qos})
+		total++
+
+		// decrement counter
+		sl = sl - n - 1
+	}
+
+	// check for empty subscription list
+	if len(sp.Subscriptions) == 0 {
+		return total, makeError(sp.Type(), "empty subscription 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 (sp *Subscribe) Encode(dst []byte) (int, error) {
+	total := 0
+
+	// check packet id
+	if !sp.ID.Valid() {
+		return total, makeError(sp.Type(), "packet id must be grater than zero")
+	}
+
+	// encode header
+	n, err := headerEncode(dst[total:], 0, sp.len(), sp.Len(), SUBSCRIBE)
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// write packet id
+	binary.BigEndian.PutUint16(dst[total:], uint16(sp.ID))
+	total += 2
+
+	for _, t := range sp.Subscriptions {
+		// write topic
+		n, err := writeLPString(dst[total:], t.Topic, sp.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+
+		// check qos
+		if !t.QOS.Successful() {
+			return total, makeError(sp.Type(), "invalid QOS level (%d)", t.QOS)
+		}
+
+		// write qos
+		dst[total] = byte(t.QOS)
+
+		total++
+	}
+
+	return total, nil
+}
+
+// Returns the payload length.
+func (sp *Subscribe) len() int {
+	// packet ID
+	total := 2
+
+	for _, t := range sp.Subscriptions {
+		total += 2 + len(t.Topic) + 1
+	}
+
+	return total
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/type.go b/vendor/github.com/256dpi/gomqtt/packet/type.go
new file mode 100644
index 000000000..b2abbd083
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/type.go
@@ -0,0 +1,141 @@
+package packet
+
+import "errors"
+
+// ErrInvalidPacketType is returned by New if the packet type is invalid.
+var ErrInvalidPacketType = errors.New("invalid packet type")
+
+// Type represents the MQTT packet types.
+type Type byte
+
+// All packet types.
+const (
+	_ Type = iota
+	CONNECT
+	CONNACK
+	PUBLISH
+	PUBACK
+	PUBREC
+	PUBREL
+	PUBCOMP
+	SUBSCRIBE
+	SUBACK
+	UNSUBSCRIBE
+	UNSUBACK
+	PINGREQ
+	PINGRESP
+	DISCONNECT
+)
+
+// String returns the type as a string.
+func (t Type) String() string {
+	switch t {
+	case CONNECT:
+		return "Connect"
+	case CONNACK:
+		return "Connack"
+	case PUBLISH:
+		return "Publish"
+	case PUBACK:
+		return "Puback"
+	case PUBREC:
+		return "Pubrec"
+	case PUBREL:
+		return "Pubrel"
+	case PUBCOMP:
+		return "Pubcomp"
+	case SUBSCRIBE:
+		return "Subscribe"
+	case SUBACK:
+		return "Suback"
+	case UNSUBSCRIBE:
+		return "Unsubscribe"
+	case UNSUBACK:
+		return "Unsuback"
+	case PINGREQ:
+		return "Pingreq"
+	case PINGRESP:
+		return "Pingresp"
+	case DISCONNECT:
+		return "Disconnect"
+	}
+
+	return "Unknown"
+}
+
+// DefaultFlags returns the default flag values for the packet type, as defined
+// by the MQTT spec, except for PUBLISH.
+func (t Type) defaultFlags() byte {
+	switch t {
+	case CONNECT:
+		return 0
+	case CONNACK:
+		return 0
+	case PUBACK:
+		return 0
+	case PUBREC:
+		return 0
+	case PUBREL:
+		return 2 // 00000010
+	case PUBCOMP:
+		return 0
+	case SUBSCRIBE:
+		return 2 // 00000010
+	case SUBACK:
+		return 0
+	case UNSUBSCRIBE:
+		return 2 // 00000010
+	case UNSUBACK:
+		return 0
+	case PINGREQ:
+		return 0
+	case PINGRESP:
+		return 0
+	case DISCONNECT:
+		return 0
+	}
+
+	return 0
+}
+
+// New creates a new packet based on the type. It is a shortcut to call one of
+// the New*Packet functions. An error is returned if the type is invalid.
+func (t Type) New() (Generic, error) {
+	switch t {
+	case CONNECT:
+		return NewConnect(), nil
+	case CONNACK:
+		return NewConnack(), nil
+	case PUBLISH:
+		return NewPublish(), nil
+	case PUBACK:
+		return NewPuback(), nil
+	case PUBREC:
+		return NewPubrec(), nil
+	case PUBREL:
+		return NewPubrel(), nil
+	case PUBCOMP:
+		return NewPubcomp(), nil
+	case SUBSCRIBE:
+		return NewSubscribe(), nil
+	case SUBACK:
+		return NewSuback(), nil
+	case UNSUBSCRIBE:
+		return NewUnsubscribe(), nil
+	case UNSUBACK:
+		return NewUnsuback(), nil
+	case PINGREQ:
+		return NewPingreq(), nil
+	case PINGRESP:
+		return NewPingresp(), nil
+	case DISCONNECT:
+		return NewDisconnect(), nil
+	}
+
+	return nil, ErrInvalidPacketType
+}
+
+// Valid returns a boolean indicating whether the type is valid or not.
+func (t Type) Valid() bool {
+	return t >= CONNECT && t <= DISCONNECT
+}
diff --git a/vendor/github.com/256dpi/gomqtt/packet/unsubscribe.go b/vendor/github.com/256dpi/gomqtt/packet/unsubscribe.go
new file mode 100644
index 000000000..7fa95a3d1
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/packet/unsubscribe.go
@@ -0,0 +1,145 @@
+package packet
+
+import (
+	"encoding/binary"
+	"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 (up *Unsubscribe) Type() Type {
+	return UNSUBSCRIBE
+}
+
+// String returns a string representation of the packet.
+func (up *Unsubscribe) String() string {
+	var topics []string
+
+	for _, t := range up.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 (up *Unsubscribe) Len() int {
+	ml := up.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 (up *Unsubscribe) Decode(src []byte) (int, error) {
+	total := 0
+
+	// decode header
+	hl, _, rl, err := headerDecode(src[total:], UNSUBSCRIBE)
+	total += hl
+	if err != nil {
+		return total, err
+	}
+
+	// check buffer length
+	if len(src) < total+2 {
+		return total, makeError(up.Type(), "insufficient buffer size, expected %d, got %d", total+2, len(src))
+	}
+
+	// read packet id
+	up.ID = ID(binary.BigEndian.Uint16(src[total:]))
+	total += 2
+
+	// check packet id
+	if !up.ID.Valid() {
+		return total, makeError(up.Type(), "packet id must be grater than zero")
+	}
+
+	// prepare counter
+	tl := int(rl) - 2
+
+	// reset topics
+	up.Topics = up.Topics[:0]
+
+	for tl > 0 {
+		// read topic
+		t, n, err := readLPString(src[total:], up.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+
+		// append to list
+		up.Topics = append(up.Topics, t)
+
+		// decrement counter
+		tl = tl - n - 1
+	}
+
+	// check for empty list
+	if len(up.Topics) == 0 {
+		return total, makeError(up.Type(), "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 (up *Unsubscribe) Encode(dst []byte) (int, error) {
+	total := 0
+
+	// check packet id
+	if !up.ID.Valid() {
+		return total, makeError(up.Type(), "packet id must be grater than zero")
+	}
+
+	// encode header
+	n, err := headerEncode(dst[total:], 0, up.len(), up.Len(), UNSUBSCRIBE)
+	total += n
+	if err != nil {
+		return total, err
+	}
+
+	// write packet id
+	binary.BigEndian.PutUint16(dst[total:], uint16(up.ID))
+	total += 2
+
+	for _, t := range up.Topics {
+		// write topic
+		n, err := writeLPString(dst[total:], t, up.Type())
+		total += n
+		if err != nil {
+			return total, err
+		}
+	}
+
+	return total, nil
+}
+
+// Returns the payload length.
+func (up *Unsubscribe) len() int {
+	// packet ID
+	total := 2
+
+	for _, t := range up.Topics {
+		total += 2 + len(t)
+	}
+
+	return total
+}
diff --git a/vendor/github.com/256dpi/gomqtt/session/id_counter.go b/vendor/github.com/256dpi/gomqtt/session/id_counter.go
new file mode 100644
index 000000000..c4777030b
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/session/id_counter.go
@@ -0,0 +1,53 @@
+package session
+
+import (
+	"sync"
+
+	"github.com/256dpi/gomqtt/packet"
+)
+
+// An IDCounter continuously counts packet ids.
+type IDCounter struct {
+	next  packet.ID
+	mutex sync.Mutex
+}
+
+// NewIDCounter returns a new counter.
+func NewIDCounter() *IDCounter {
+	return NewIDCounterWithNext(1)
+}
+
+// NewIDCounterWithNext returns a new counter that will emit the specified if
+// id as the next id.
+func NewIDCounterWithNext(next packet.ID) *IDCounter {
+	return &IDCounter{
+		next: next,
+	}
+}
+
+// NextID will return the next id.
+func (c *IDCounter) NextID() packet.ID {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	// ignore zeroes
+	if c.next == 0 {
+		c.next++
+	}
+
+	// cache next id
+	id := c.next
+
+	// increment id
+	c.next++
+
+	return id
+}
+
+// Reset will reset the counter.
+func (c *IDCounter) Reset() {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
+	c.next = 1
+}
diff --git a/vendor/github.com/256dpi/gomqtt/session/memory_session.go b/vendor/github.com/256dpi/gomqtt/session/memory_session.go
new file mode 100644
index 000000000..6a403aede
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/session/memory_session.go
@@ -0,0 +1,83 @@
+// Package session implements session objects to be used with MQTT clients and
+// brokers.
+package session
+
+import (
+	"github.com/256dpi/gomqtt/packet"
+)
+
+// Direction denotes a packets direction.
+type Direction int
+
+const (
+	// Incoming packets are being received.
+	Incoming Direction = iota
+
+	// Outgoing packets are being be sent.
+	Outgoing
+)
+
+// A MemorySession stores packets in memory.
+type MemorySession struct {
+	Counter  *IDCounter
+	Incoming *PacketStore
+	Outgoing *PacketStore
+}
+
+// NewMemorySession returns a new MemorySession.
+func NewMemorySession() *MemorySession {
+	return &MemorySession{
+		Counter:  NewIDCounter(),
+		Incoming: NewPacketStore(),
+		Outgoing: NewPacketStore(),
+	}
+}
+
+// NextID will return the next id for outgoing packets.
+func (s *MemorySession) NextID() packet.ID {
+	return s.Counter.NextID()
+}
+
+// SavePacket will store a packet in the session. An eventual existing
+// packet with the same id gets quietly overwritten.
+func (s *MemorySession) SavePacket(dir Direction, pkt packet.Generic) error {
+	s.storeForDirection(dir).Save(pkt)
+	return nil
+}
+
+// LookupPacket will retrieve a packet from the session using a packet id.
+func (s *MemorySession) LookupPacket(dir Direction, id packet.ID) (packet.Generic, error) {
+	return s.storeForDirection(dir).Lookup(id), nil
+}
+
+// DeletePacket will remove a packet from the session. The method must not
+// return an error if no packet with the specified id does exists.
+func (s *MemorySession) DeletePacket(dir Direction, id packet.ID) error {
+	s.storeForDirection(dir).Delete(id)
+	return nil
+}
+
+// AllPackets will return all packets currently saved in the session.
+func (s *MemorySession) AllPackets(dir Direction) ([]packet.Generic, error) {
+	return s.storeForDirection(dir).All(), nil
+}
+
+// Reset will completely reset the session.
+func (s *MemorySession) Reset() error {
+	// reset counter and stores
+	s.Counter.Reset()
+	s.Incoming.Reset()
+	s.Outgoing.Reset()
+
+	return nil
+}
+
+func (s *MemorySession) storeForDirection(dir Direction) *PacketStore {
+	if dir == Incoming {
+		return s.Incoming
+	} else if dir == Outgoing {
+		return s.Outgoing
+	}
+
+	panic("unknown direction")
+}
diff --git a/vendor/github.com/256dpi/gomqtt/session/packet_store.go b/vendor/github.com/256dpi/gomqtt/session/packet_store.go
new file mode 100644
index 000000000..5c8a1d633
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/session/packet_store.go
@@ -0,0 +1,85 @@
+package session
+
+import (
+	"sync"
+
+	"github.com/256dpi/gomqtt/packet"
+)
+
+// PacketStore is a goroutine safe packet store.
+type PacketStore struct {
+	packets map[packet.ID]packet.Generic
+	mutex   sync.RWMutex
+}
+
+// NewPacketStore returns a new PacketStore.
+func NewPacketStore() *PacketStore {
+	return &PacketStore{
+		packets: make(map[packet.ID]packet.Generic),
+	}
+}
+
+// NewPacketStoreWithPackets returns a new PacketStore with the provided packets.
+func NewPacketStoreWithPackets(packets []packet.Generic) *PacketStore {
+	// prepare store
+	store := &PacketStore{
+		packets: make(map[packet.ID]packet.Generic),
+	}
+
+	// add packets
+	for _, pkt := range packets {
+		store.Save(pkt)
+	}
+
+	return store
+}
+
+// Save will store a packet in the store. An eventual existing packet with the
+// same id gets quietly overwritten.
+func (s *PacketStore) Save(pkt packet.Generic) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	id, ok := packet.GetID(pkt)
+	if ok {
+		s.packets[id] = pkt
+	}
+}
+
+// Lookup will retrieve a packet from the store.
+func (s *PacketStore) Lookup(id packet.ID) packet.Generic {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+
+	return s.packets[id]
+}
+
+// Delete will remove a packet from the store.
+func (s *PacketStore) Delete(id packet.ID) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	delete(s.packets, id)
+}
+
+// All will return all packets currently saved in the store.
+func (s *PacketStore) All() []packet.Generic {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+
+	var all []packet.Generic
+
+	for _, pkt := range s.packets {
+		all = append(all, pkt)
+	}
+
+	return all
+}
+
+// Reset will reset the store.
+func (s *PacketStore) Reset() {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	s.packets = make(map[packet.ID]packet.Generic)
+}
diff --git a/vendor/github.com/256dpi/gomqtt/topic/topic.go b/vendor/github.com/256dpi/gomqtt/topic/topic.go
new file mode 100644
index 000000000..15ac86e37
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/topic/topic.go
@@ -0,0 +1,65 @@
+// Package topic implements common methods to handle MQTT topics.
+package topic
+
+import (
+	"errors"
+	"regexp"
+	"strings"
+)
+
+// ErrZeroLength is returned by Parse if a topics has a zero length.
+var ErrZeroLength = errors.New("zero length topic")
+
+// ErrWildcards is returned by Parse if a topic contains invalid wildcards.
+var ErrWildcards = errors.New("invalid use of wildcards")
+
+var multiSlashRegex = regexp.MustCompile(`/+`)
+
+// Parse removes duplicate and trailing slashes from the supplied
+// string and returns the normalized topic.
+func Parse(topic string, allowWildcards bool) (string, error) {
+	// check for zero length
+	if topic == "" {
+		return "", ErrZeroLength
+	}
+
+	// normalize topic
+	topic = multiSlashRegex.ReplaceAllString(topic, "/")
+
+	// remove trailing slashes
+	topic = strings.TrimRight(topic, "/")
+
+	// check again for zero length
+	if topic == "" {
+		return "", ErrZeroLength
+	}
+
+	// split to segments
+	segments := strings.Split(topic, "/")
+
+	// check all segments
+	for i, s := range segments {
+		// check use of wildcards
+		if (strings.Contains(s, "+") || strings.Contains(s, "#")) && len(s) > 1 {
+			return "", ErrWildcards
+		}
+
+		// check if wildcards are allowed
+		if !allowWildcards && (s == "#" || s == "+") {
+			return "", ErrWildcards
+		}
+
+		// check if hash is the last character
+		if s == "#" && i != len(segments)-1 {
+			return "", ErrWildcards
+		}
+	}
+
+	return topic, nil
+}
+
+// ContainsWildcards tests if the supplied topic contains wildcards. The topics
+// is expected to be tested and normalized using Parse beforehand.
+func ContainsWildcards(topic string) bool {
+	return strings.Contains(topic, "+") || strings.Contains(topic, "#")
+}
diff --git a/vendor/github.com/256dpi/gomqtt/topic/tree.go b/vendor/github.com/256dpi/gomqtt/topic/tree.go
new file mode 100644
index 000000000..eff87dfd2
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/topic/tree.go
@@ -0,0 +1,425 @@
+package topic
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+)
+
+type node struct {
+	children map[string]*node
+	values   []interface{}
+}
+
+func newNode() *node {
+	return &node{
+		children: make(map[string]*node),
+	}
+}
+
+func (n *node) removeValue(value interface{}) {
+	for i, v := range n.values {
+		if v == value {
+			// remove without preserving order
+			n.values[i] = n.values[len(n.values)-1]
+			n.values = n.values[:len(n.values)-1]
+			break
+		}
+	}
+}
+
+func (n *node) clearValues() {
+	n.values = []interface{}{}
+}
+
+func (n *node) string(i int) string {
+	str := ""
+
+	if i != 0 {
+		str = fmt.Sprintf("%d", len(n.values))
+	}
+
+	for key, node := range n.children {
+		str += fmt.Sprintf("\n| %s'%s' => %s", strings.Repeat(" ", i*2), key, node.string(i+1))
+	}
+
+	return str
+}
+
+// A Tree implements a thread-safe topic tree.
+type Tree struct {
+	// The separator character. Default: "/"
+	Separator string
+
+	// The single level wildcard character. Default: "+"
+	WildcardOne string
+
+	// The multi level wildcard character. Default "#"
+	WildcardSome string
+
+	root  *node
+	mutex sync.RWMutex
+}
+
+// NewTree returns a new Tree.
+func NewTree() *Tree {
+	return &Tree{
+		Separator:    "/",
+		WildcardOne:  "+",
+		WildcardSome: "#",
+
+		root: newNode(),
+	}
+}
+
+// Add registers the value for the supplied topic. This function will
+// automatically grow the tree. If value already exists for the given topic it
+// will not be added again.
+func (t *Tree) Add(topic string, value interface{}) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	t.add(value, 0, strings.Split(topic, t.Separator), t.root)
+}
+
+func (t *Tree) add(value interface{}, i int, segments []string, node *node) {
+	// add value to leaf
+	if i == len(segments) {
+		for _, v := range node.values {
+			if v == value {
+				return
+			}
+		}
+
+		node.values = append(node.values, value)
+		return
+	}
+
+	segment := segments[i]
+	child, ok := node.children[segment]
+
+	// create missing node
+	if !ok {
+		child = newNode()
+		node.children[segment] = child
+	}
+
+	t.add(value, i+1, segments, child)
+}
+
+// Set sets the supplied value as the only value for the supplied topic. This
+// function will automatically grow the tree.
+func (t *Tree) Set(topic string, value interface{}) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	t.set(value, 0, strings.Split(topic, t.Separator), t.root)
+}
+
+func (t *Tree) set(value interface{}, i int, segments []string, node *node) {
+	// set value on leaf
+	if i == len(segments) {
+		node.values = []interface{}{value}
+		return
+	}
+
+	segment := segments[i]
+	child, ok := node.children[segment]
+
+	// create missing node
+	if !ok {
+		child = newNode()
+		node.children[segment] = child
+	}
+
+	t.set(value, i+1, segments, child)
+}
+
+// Get gets the values from the topic that exactly matches the supplied topics.
+func (t *Tree) Get(topic string) []interface{} {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	return t.get(0, strings.Split(topic, t.Separator), t.root)
+}
+
+func (t *Tree) get(i int, segments []string, node *node) []interface{} {
+	// set value on leaf
+	if i == len(segments) {
+		return node.values
+	}
+
+	// get next segment
+	segment := segments[i]
+	child, ok := node.children[segment]
+	if !ok {
+		return nil
+	}
+
+	return t.get(i+1, segments, child)
+}
+
+// Remove un-registers the value from the supplied topic. This function will
+// automatically shrink the tree.
+func (t *Tree) Remove(topic string, value interface{}) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	t.remove(value, 0, strings.Split(topic, t.Separator), t.root)
+}
+
+// Empty will unregister all values from the supplied topic. This function will
+// automatically shrink the tree.
+func (t *Tree) Empty(topic string) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	t.remove(nil, 0, strings.Split(topic, t.Separator), t.root)
+}
+
+func (t *Tree) remove(value interface{}, i int, segments []string, node *node) bool {
+	// clear or remove value from leaf node
+	if i == len(segments) {
+		if value == nil {
+			node.clearValues()
+		} else {
+			node.removeValue(value)
+		}
+
+		return len(node.values) == 0 && len(node.children) == 0
+	}
+
+	segment := segments[i]
+	child, ok := node.children[segment]
+
+	// node not found
+	if !ok {
+		return false
+	}
+
+	if t.remove(value, i+1, segments, child) {
+		delete(node.children, segment)
+	}
+
+	return len(node.values) == 0 && len(node.children) == 0
+}
+
+// Clear will unregister the supplied value from all topics. This function will
+// automatically shrink the tree.
+func (t *Tree) Clear(value interface{}) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	t.clear(value, t.root)
+}
+
+func (t *Tree) clear(value interface{}, node *node) bool {
+	node.removeValue(value)
+
+	// remove value from all nodes
+	for segment, child := range node.children {
+		if t.clear(value, child) {
+			delete(node.children, segment)
+		}
+	}
+
+	return len(node.values) == 0 && len(node.children) == 0
+}
+
+// Match will return a set of values from topics that match the supplied topic.
+// The result set will be cleared from duplicate values.
+//
+// Note: In contrast to Search, Match does not respect wildcards in the query but
+// in the stored tree.
+func (t *Tree) Match(topic string) []interface{} {
+	t.mutex.RLock()
+	defer t.mutex.RUnlock()
+
+	segments := strings.Split(topic, t.Separator)
+	values := t.match([]interface{}{}, 0, segments, t.root)
+
+	return t.clean(values)
+}
+
+func (t *Tree) match(result []interface{}, i int, segments []string, node *node) []interface{} {
+	// add all values to the result set that match multiple levels
+	if child, ok := node.children[t.WildcardSome]; ok {
+		result = append(result, child.values...)
+	}
+
+	// when finished add all values to the result set
+	if i == len(segments) {
+		return append(result, node.values...)
+	}
+
+	// advance children that match a single level
+	if child, ok := node.children[t.WildcardOne]; ok {
+		result = t.match(result, i+1, segments, child)
+	}
+
+	segment := segments[i]
+
+	// match segments and get children
+	if segment != t.WildcardOne && segment != t.WildcardSome {
+		if child, ok := node.children[segment]; ok {
+			result = t.match(result, i+1, segments, child)
+		}
+	}
+
+	return result
+}
+
+// MatchFirst will run Match and return the first value or nil.
+func (t *Tree) MatchFirst(topic string) interface{} {
+	values := t.Match(topic)
+
+	if len(values) > 0 {
+		return values[0]
+	}
+
+	return nil
+}
+
+// Search will return a set of values from topics that match the supplied topic.
+// The result set will be cleared from duplicate values.
+//
+// Note: In contrast to Match, Search respects wildcards in the query but not in
+// the stored tree.
+func (t *Tree) Search(topic string) []interface{} {
+	t.mutex.RLock()
+	defer t.mutex.RUnlock()
+
+	segments := strings.Split(topic, t.Separator)
+	values := t.search([]interface{}{}, 0, segments, t.root)
+
+	return t.clean(values)
+}
+
+func (t *Tree) search(result []interface{}, i int, segments []string, node *node) []interface{} {
+	// when finished add all values to the result set
+	if i == len(segments) {
+		return append(result, node.values...)
+	}
+
+	// get segment
+	segment := segments[i]
+
+	// add all current and further values
+	if segment == t.WildcardSome {
+		result = append(result, node.values...)
+
+		for _, child := range node.children {
+			result = t.search(result, i, segments, child)
+		}
+	}
+
+	// add all current values and continue
+	if segment == t.WildcardOne {
+		result = append(result, node.values...)
+
+		for _, child := range node.children {
+			result = t.search(result, i+1, segments, child)
+		}
+	}
+
+	// match segments and get children
+	if segment != t.WildcardOne && segment != t.WildcardSome {
+		if child, ok := node.children[segment]; ok {
+			result = t.search(result, i+1, segments, child)
+		}
+	}
+
+	return result
+}
+
+// SearchFirst will run Search and return the first value or nil.
+func (t *Tree) SearchFirst(topic string) interface{} {
+	values := t.Search(topic)
+
+	if len(values) > 0 {
+		return values[0]
+	}
+
+	return nil
+}
+
+// clean will remove duplicates
+func (t *Tree) clean(values []interface{}) []interface{} {
+	result := values[:0]
+
+	for _, v := range values {
+		if contains(result, v) {
+			continue
+		}
+
+		result = append(result, v)
+	}
+
+	return result
+}
+
+// Count will count all stored values in the tree. It will not filter out
+// duplicate values and thus might return a different result to `len(All())`.
+func (t *Tree) Count() int {
+	t.mutex.RLock()
+	defer t.mutex.RUnlock()
+
+	return t.count(t.root)
+}
+
+func (t *Tree) count(node *node) int {
+	// prepare total
+	total := 0
+
+	// add children to results
+	for _, child := range node.children {
+		total += t.count(child)
+	}
+
+	// add values to result
+	return total + len(node.values)
+}
+
+// All will return all stored values in the tree.
+func (t *Tree) All() []interface{} {
+	t.mutex.RLock()
+	defer t.mutex.RUnlock()
+
+	return t.clean(t.all([]interface{}{}, t.root))
+}
+
+func (t *Tree) all(result []interface{}, node *node) []interface{} {
+	// add children to results
+	for _, child := range node.children {
+		result = t.all(result, child)
+	}
+
+	// add current node to results
+	return append(result, node.values...)
+}
+
+// Reset will completely clear the tree.
+func (t *Tree) Reset() {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	t.root = newNode()
+}
+
+// String will return a string representation of the tree.
+func (t *Tree) String() string {
+	t.mutex.RLock()
+	defer t.mutex.RUnlock()
+
+	return fmt.Sprintf("topic.Tree:%s", t.root.string(0))
+}
+
+func contains(list []interface{}, value interface{}) bool {
+	for _, v := range list {
+		if v == value {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/base_conn.go b/vendor/github.com/256dpi/gomqtt/transport/base_conn.go
new file mode 100644
index 000000000..fc66839c5
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/base_conn.go
@@ -0,0 +1,129 @@
+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
+
+	sMutex sync.Mutex
+	rMutex 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 {
+	c.sMutex.Lock()
+	defer c.sMutex.Unlock()
+
+	// write packet
+	err := c.stream.Write(pkt, async)
+	if err != nil {
+		// ensure connection 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 an Error if there was an error 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) {
+	c.rMutex.Lock()
+	defer c.rMutex.Unlock()
+
+	// read next packet
+	pkt, err := c.stream.Read()
+	if err != nil {
+		// ensure connection gets closed
+		c.carrier.Close()
+
+		return nil, err
+	}
+
+	// reset timeout
+	c.resetTimeout()
+
+	return pkt, nil
+}
+
+// Close will close the underlying connection and cleanup resources. It will
+// return an Error if there was an error while closing the underlying
+// connection.
+func (c *BaseConn) Close() error {
+	c.sMutex.Lock()
+	defer c.sMutex.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.Decoder.Limit = 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) {
+	c.readTimeout = timeout
+	c.resetTimeout()
+}
+
+func (c *BaseConn) resetTimeout() {
+	if c.readTimeout > 0 {
+		c.carrier.SetReadDeadline(time.Now().Add(c.readTimeout))
+	} else {
+		c.carrier.SetReadDeadline(time.Time{})
+	}
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/conn.go b/vendor/github.com/256dpi/gomqtt/transport/conn.go
new file mode 100644
index 000000000..5fd010f0c
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/conn.go
@@ -0,0 +1,48 @@
+package transport
+
+import (
+	"net"
+	"time"
+
+	"github.com/256dpi/gomqtt/packet"
+)
+
+// A Conn is a connection between a client and a broker. It abstracts an
+// existing underlying stream connection.
+type Conn interface {
+	// 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.
+	Send(pkt packet.Generic, async bool) error
+
+	// Receive will read from the underlying connection and return a fully read
+	// packet. It will return an Error if there was an error while decoding or
+	// reading from the underlying connection.
+	//
+	// Note: Only one goroutine can Receive at the same time.
+	Receive() (packet.Generic, error)
+
+	// Close will close the underlying connection and cleanup resources. It will
+	// return an Error if there was an error while closing the underlying
+	// connection.
+	Close() error
+
+	// 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.
+	SetReadLimit(limit int64)
+
+	// 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.
+	SetReadTimeout(timeout time.Duration)
+
+	// LocalAddr will return the underlying connection's local net address.
+	LocalAddr() net.Addr
+
+	// RemoteAddr will return the underlying connection's remote net address.
+	RemoteAddr() net.Addr
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/dialer.go b/vendor/github.com/256dpi/gomqtt/transport/dialer.go
new file mode 100644
index 000000000..1fcd19b6e
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/dialer.go
@@ -0,0 +1,117 @@
+package transport
+
+import (
+	"crypto/tls"
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+
+	"github.com/gorilla/websocket"
+)
+
+// The Dialer handles connecting to a server and creating a connection.
+type Dialer struct {
+	TLSConfig     *tls.Config
+	RequestHeader http.Header
+
+	DefaultTCPPort string
+	DefaultTLSPort string
+	DefaultWSPort  string
+	DefaultWSSPort string
+
+	webSocketDialer *websocket.Dialer
+}
+
+// NewDialer returns a new Dialer.
+func NewDialer() *Dialer {
+	return &Dialer{
+		DefaultTCPPort: "1883",
+		DefaultTLSPort: "8883",
+		DefaultWSPort:  "80",
+		DefaultWSSPort: "443",
+		webSocketDialer: &websocket.Dialer{
+			Proxy:        http.ProxyFromEnvironment,
+			Subprotocols: []string{"mqtt"},
+		},
+	}
+}
+
+var sharedDialer *Dialer
+
+func init() {
+	sharedDialer = NewDialer()
+}
+
+// Dial is a shorthand function.
+func Dial(urlString string) (Conn, error) {
+	return sharedDialer.Dial(urlString)
+}
+
+// Dial initiates a connection based in information extracted from an URL.
+func (d *Dialer) Dial(urlString string) (Conn, error) {
+	urlParts, err := url.ParseRequestURI(urlString)
+	if err != nil {
+		return nil, err
+	}
+
+	host, port, err := net.SplitHostPort(urlParts.Host)
+	if err != nil {
+		host = urlParts.Host
+		port = ""
+	}
+
+	switch urlParts.Scheme {
+	case "tcp", "mqtt":
+		if port == "" {
+			port = d.DefaultTCPPort
+		}
+
+		conn, err := net.Dial("tcp", net.JoinHostPort(host, port))
+		if err != nil {
+			return nil, err
+		}
+
+		return NewNetConn(conn), nil
+	case "tls", "mqtts":
+		if port == "" {
+			port = d.DefaultTLSPort
+		}
+
+		conn, err := tls.Dial("tcp", net.JoinHostPort(host, port), d.TLSConfig)
+		if err != nil {
+			return nil, err
+		}
+
+		return NewNetConn(conn), nil
+	case "ws":
+		if port == "" {
+			port = d.DefaultWSPort
+		}
+
+		wsURL := fmt.Sprintf("ws://%s:%s%s", host, port, urlParts.Path)
+
+		conn, _, err := d.webSocketDialer.Dial(wsURL, d.RequestHeader)
+		if err != nil {
+			return nil, err
+		}
+
+		return NewWebSocketConn(conn), nil
+	case "wss":
+		if port == "" {
+			port = d.DefaultWSSPort
+		}
+
+		wsURL := fmt.Sprintf("wss://%s:%s%s", host, port, urlParts.Path)
+
+		d.webSocketDialer.TLSClientConfig = d.TLSConfig
+		conn, _, err := d.webSocketDialer.Dial(wsURL, d.RequestHeader)
+		if err != nil {
+			return nil, err
+		}
+
+		return NewWebSocketConn(conn), nil
+	}
+
+	return nil, ErrUnsupportedProtocol
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/launcher.go b/vendor/github.com/256dpi/gomqtt/transport/launcher.go
new file mode 100644
index 000000000..2e84f7936
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/launcher.go
@@ -0,0 +1,48 @@
+package transport
+
+import (
+	"crypto/tls"
+	"net/url"
+)
+
+// The Launcher helps with launching a server and accepting connections.
+type Launcher struct {
+	TLSConfig *tls.Config
+}
+
+// NewLauncher returns a new Launcher.
+func NewLauncher() *Launcher {
+	return &Launcher{}
+}
+
+var sharedLauncher *Launcher
+
+func init() {
+	sharedLauncher = NewLauncher()
+}
+
+// Launch is a shorthand function.
+func Launch(urlString string) (Server, error) {
+	return sharedLauncher.Launch(urlString)
+}
+
+// Launch will launch a server based on information extracted from an URL.
+func (l *Launcher) Launch(urlString string) (Server, error) {
+	urlParts, err := url.ParseRequestURI(urlString)
+	if err != nil {
+		return nil, err
+	}
+
+	switch urlParts.Scheme {
+	case "tcp", "mqtt":
+		return CreateNetServer(urlParts.Host)
+	case "tls", "mqtts":
+		return CreateSecureNetServer(urlParts.Host, l.TLSConfig)
+	case "ws":
+		return CreateWebSocketServer(urlParts.Host)
+	case "wss":
+		return CreateSecureWebSocketServer(urlParts.Host, l.TLSConfig)
+	}
+
+	return nil, ErrUnsupportedProtocol
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/net_conn.go b/vendor/github.com/256dpi/gomqtt/transport/net_conn.go
new file mode 100644
index 000000000..99ff53804
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/net_conn.go
@@ -0,0 +1,35 @@
+package transport
+
+import (
+	"net"
+)
+
+// A NetConn is a wrapper around a basic TCP connection.
+type NetConn struct {
+	*BaseConn
+
+	conn net.Conn
+}
+
+// NewNetConn returns a new NetConn.
+func NewNetConn(conn net.Conn) *NetConn {
+	return &NetConn{
+		BaseConn: NewBaseConn(conn),
+		conn:     conn,
+	}
+}
+
+// LocalAddr returns the local network address.
+func (c *NetConn) LocalAddr() net.Addr {
+	return c.conn.LocalAddr()
+}
+
+// RemoteAddr returns the remote network address.
+func (c *NetConn) RemoteAddr() net.Addr {
+	return c.conn.RemoteAddr()
+}
+
+// UnderlyingConn returns the underlying net.Conn.
+func (c *NetConn) UnderlyingConn() net.Conn {
+	return c.conn
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/net_server.go b/vendor/github.com/256dpi/gomqtt/transport/net_server.go
new file mode 100644
index 000000000..4719ce7b3
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/net_server.go
@@ -0,0 +1,65 @@
+package transport
+
+import (
+	"crypto/tls"
+	"net"
+)
+
+// A NetServer accepts net.Conn based connections.
+type NetServer struct {
+	listener net.Listener
+}
+
+// NewNetServer wraps the provided listener.
+func NewNetServer(listener net.Listener) *NetServer {
+	return &NetServer{
+		listener: listener,
+	}
+}
+
+// CreateNetServer creates a new TCP server that listens on the provided address.
+func CreateNetServer(address string) (*NetServer, error) {
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewNetServer(listener), nil
+}
+
+// CreateSecureNetServer creates a new TLS server that listens on the provided address.
+func CreateSecureNetServer(address string, config *tls.Config) (*NetServer, error) {
+	listener, err := tls.Listen("tcp", address, config)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewNetServer(listener), nil
+}
+
+// Accept will return the next available connection or block until a
+// connection becomes available, otherwise returns an Error.
+func (s *NetServer) Accept() (Conn, error) {
+	conn, err := s.listener.Accept()
+	if err != nil {
+		return nil, err
+	}
+
+	return NewNetConn(conn), nil
+}
+
+// Close will close the underlying listener and cleanup resources. It will
+// return an Error if the underlying listener didn't close cleanly.
+func (s *NetServer) Close() error {
+	err := s.listener.Close()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Addr returns the server's network address.
+func (s *NetServer) Addr() net.Addr {
+	return s.listener.Addr()
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/server.go b/vendor/github.com/256dpi/gomqtt/transport/server.go
new file mode 100644
index 000000000..c4b5a4068
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/server.go
@@ -0,0 +1,17 @@
+package transport
+
+import "net"
+
+// A Server is a local port on which incoming connections can be accepted.
+type Server interface {
+	// Accept will return the next available connection or block until a
+	// connection becomes available, otherwise returns an Error.
+	Accept() (Conn, error)
+
+	// Close will close the underlying listener and cleanup resources. It will
+	// return an Error if the underlying listener didn't close cleanly.
+	Close() error
+
+	// Addr returns the server's network address.
+	Addr() net.Addr
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/transport.go b/vendor/github.com/256dpi/gomqtt/transport/transport.go
new file mode 100644
index 000000000..a662640bd
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/transport.go
@@ -0,0 +1,14 @@
+// Package transport implements functionality for handling MQTT connections.
+package transport
+
+import "errors"
+
+// ErrUnsupportedProtocol is returned if either the launcher or dialer
+// couldn't infer the protocol from the URL.
+var ErrUnsupportedProtocol = errors.New("unsupported protocol")
+
+// ErrAcceptAfterClose can be returned by a WebSocketServer during Accept()
+// if the server has been already closed and the internal goroutine is dying.
+//
+// Note: this error is wrapped in an Error with NetworkError code.
+var ErrAcceptAfterClose = errors.New("accept after close")
diff --git a/vendor/github.com/256dpi/gomqtt/transport/websocket_conn.go b/vendor/github.com/256dpi/gomqtt/transport/websocket_conn.go
new file mode 100644
index 000000000..1f52b087f
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/websocket_conn.go
@@ -0,0 +1,126 @@
+package transport
+
+import (
+	"errors"
+	"io"
+	"net"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+// ErrNotBinary may be returned by WebSocket connection when a message is
+// received that is not binary.
+var ErrNotBinary = errors.New("received web socket message is not binary")
+
+type wsStream struct {
+	conn   *websocket.Conn
+	reader io.Reader
+}
+
+func (s *wsStream) Read(p []byte) (int, error) {
+	total := 0
+	buf := p
+
+	for {
+		// get next reader
+		if s.reader == nil {
+			messageType, reader, err := s.conn.NextReader()
+			if _, ok := err.(*websocket.CloseError); ok {
+				return 0, io.EOF
+			} else if err != nil {
+				return 0, err
+			} else if messageType != websocket.BinaryMessage {
+				return 0, ErrNotBinary
+			}
+
+			// set current reader
+			s.reader = reader
+		}
+
+		// read data
+		n, err := s.reader.Read(buf)
+
+		// increment counter
+		total += n
+		buf = buf[n:]
+
+		// handle EOF
+		if err == io.EOF {
+			// clear reader
+			s.reader = nil
+
+			continue
+		}
+
+		return total, err
+	}
+}
+
+func (s *wsStream) Write(p []byte) (n int, err error) {
+	// create writer if missing
+	writer, err := s.conn.NextWriter(websocket.BinaryMessage)
+	if err != nil {
+		return 0, err
+	}
+
+	// write packet to writer
+	n, err = writer.Write(p)
+	if err != nil {
+		return n, err
+	}
+
+	// close temporary writer
+	err = writer.Close()
+	if err != nil {
+		return n, err
+	}
+
+	return n, nil
+}
+
+func (s *wsStream) Close() error {
+	// Close can be called during read and write, therefore we cannot write a
+	// close message to the client without risking a concurrent write on the
+	// websocket conn. The MQTT spec anyway requires clients to terminate the
+	// connection, therefore we don't have to really care about announcing a
+	// server-side connection close.
+
+	return s.conn.Close()
+}
+
+func (s *wsStream) SetReadDeadline(t time.Time) error {
+	return s.conn.SetReadDeadline(t)
+}
+
+// The WebSocketConn wraps a websocket.Conn. The implementation supports packets
+// that are chunked over several WebSocket messages and packets that are coalesced
+// to one WebSocket message.
+type WebSocketConn struct {
+	*BaseConn
+
+	conn *websocket.Conn
+}
+
+// NewWebSocketConn returns a new WebSocketConn.
+func NewWebSocketConn(conn *websocket.Conn) *WebSocketConn {
+	return &WebSocketConn{
+		BaseConn: NewBaseConn(&wsStream{conn: conn}),
+		conn:     conn,
+	}
+}
+
+// LocalAddr returns the local network address.
+func (c *WebSocketConn) LocalAddr() net.Addr {
+	return c.conn.LocalAddr()
+}
+
+// RemoteAddr returns the remote network address.
+func (c *WebSocketConn) RemoteAddr() net.Addr {
+	return c.conn.RemoteAddr()
+}
+
+// UnderlyingConn returns the underlying websocket.Conn.
+func (c *WebSocketConn) UnderlyingConn() *websocket.Conn {
+	return c.conn
+}
diff --git a/vendor/github.com/256dpi/gomqtt/transport/websocket_server.go b/vendor/github.com/256dpi/gomqtt/transport/websocket_server.go
new file mode 100644
index 000000000..1a581d7c9
--- /dev/null
+++ b/vendor/github.com/256dpi/gomqtt/transport/websocket_server.go
@@ -0,0 +1,154 @@
+package transport
+
+import (
+	"crypto/tls"
+	"errors"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"gopkg.in/tomb.v2"
+)
+
+var errManualClose = errors.New("internal: manual close")
+
+// The WebSocketServer accepts websocket.Conn based connections.
+type WebSocketServer struct {
+	listener      net.Listener
+	mux           *http.ServeMux
+	fallback      http.Handler
+	upgrader      *websocket.Upgrader
+	incoming      chan *WebSocketConn
+	originChecker func(r *http.Request) bool
+
+	tomb tomb.Tomb
+}
+
+// NewWebSocketServer wraps the provided listener.
+func NewWebSocketServer(listener net.Listener) *WebSocketServer {
+	// create server
+	ws := &WebSocketServer{
+		listener: listener,
+		upgrader: &websocket.Upgrader{
+			HandshakeTimeout: 60 * time.Second,
+			Subprotocols:     []string{"mqtt", "mqttv3.1"},
+		},
+		incoming: make(chan *WebSocketConn),
+	}
+
+	// add check origin method that uses the optional check origin function
+	ws.upgrader.CheckOrigin = func(r *http.Request) bool {
+		if ws.originChecker != nil {
+			return ws.originChecker(r)
+		}
+
+		return true
+	}
+
+	// create http server
+	h := &http.Server{
+		Handler: http.HandlerFunc(ws.requestHandler),
+	}
+
+	// serve http traffic in background
+	ws.tomb.Go(func() error {
+		return h.Serve(ws.listener)
+	})
+
+	return ws
+}
+
+// CreateWebSocketServer creates a new WS server that listens on the provided address.
+func CreateWebSocketServer(address string) (*WebSocketServer, error) {
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewWebSocketServer(listener), nil
+}
+
+// CreateSecureWebSocketServer creates a new WSS server that listens on the
+// provided address.
+func CreateSecureWebSocketServer(address string, config *tls.Config) (*WebSocketServer, error) {
+	listener, err := tls.Listen("tcp", address, config)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewWebSocketServer(listener), nil
+}
+
+// SetFallback will register a http.Handler that gets called if a request is not
+// a WebSocket upgrade request.
+func (s *WebSocketServer) SetFallback(handler http.Handler) {
+	s.fallback = handler
+}
+
+// SetOriginChecker sets an optional function that allows check the request origin
+// before accepting the connection.
+func (s *WebSocketServer) SetOriginChecker(fn func(r *http.Request) bool) {
+	s.originChecker = fn
+}
+
+func (s *WebSocketServer) requestHandler(w http.ResponseWriter, r *http.Request) {
+	// run fallback if request is not an upgrade
+	if r.Header.Get("Upgrade") != "websocket" && s.fallback != nil {
+		s.fallback.ServeHTTP(w, r)
+		return
+	}
+
+	// run WebSocket upgrader
+	conn, err := s.upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		// upgrader already responded to request
+		return
+	}
+
+	// create connection
+	webSocketConn := NewWebSocketConn(conn)
+
+	select {
+	case s.incoming <- webSocketConn:
+	case <-s.tomb.Dying():
+		webSocketConn.Close()
+	}
+}
+
+// Accept will return the next available connection or block until a
+// connection becomes available, otherwise returns an Error.
+func (s *WebSocketServer) Accept() (Conn, error) {
+	select {
+	case <-s.tomb.Dying():
+		if s.tomb.Err() == errManualClose {
+			// server has been closed manually
+			return nil, ErrAcceptAfterClose
+		}
+
+		// return the previously caught error
+		return nil, s.tomb.Err()
+	case conn := <-s.incoming:
+		return conn, nil
+	}
+}
+
+// Close will close the underlying listener and cleanup resources. It will
+// return an Error if the underlying listener didn't close cleanly.
+func (s *WebSocketServer) Close() error {
+	s.tomb.Kill(errManualClose)
+
+	err := s.listener.Close()
+	s.tomb.Wait()
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Addr returns the server's network address.
+func (s *WebSocketServer) Addr() net.Addr {
+	return s.listener.Addr()
+}
diff --git a/vendor/github.com/256dpi/mercury/LICENSE.md b/vendor/github.com/256dpi/mercury/LICENSE.md
new file mode 100644
index 000000000..446c112c1
--- /dev/null
+++ b/vendor/github.com/256dpi/mercury/LICENSE.md
@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2018 Joël Gähwiler
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
diff --git a/vendor/github.com/256dpi/mercury/Makefile b/vendor/github.com/256dpi/mercury/Makefile
new file mode 100644
index 000000000..7dbe4abce
--- /dev/null
+++ b/vendor/github.com/256dpi/mercury/Makefile
@@ -0,0 +1,10 @@
+all: fmt vet lint
+
+fmt:
+	go fmt .
+
+vet:
+	go vet .
+
+lint:
+	golint .
diff --git a/vendor/github.com/256dpi/mercury/README.md b/vendor/github.com/256dpi/mercury/README.md
new file mode 100644
index 000000000..fdf4290f3
--- /dev/null
+++ b/vendor/github.com/256dpi/mercury/README.md
@@ -0,0 +1,11 @@
+# mercury
+
+[![Build Status](https://travis-ci.org/256dpi/mercury.svg?branch=master)](https://travis-ci.org/256dpi/mercury)
+[![Coverage Status](https://coveralls.io/repos/github/256dpi/mercury/badge.svg?branch=master)](https://coveralls.io/github/256dpi/mercury?branch=master)
+[![GoDoc](https://godoc.org/github.com/256dpi/mercury?status.svg)](http://godoc.org/github.com/256dpi/mercury)
+[![Release](https://img.shields.io/github/release/256dpi/mercury.svg)](https://github.com/256dpi/mercury/releases)
+[![Go Report Card](https://goreportcard.com/badge/github.com/256dpi/mercury)](https://goreportcard.com/report/github.com/256dpi/mercury)
+
+**An asynchronously flushing buffered writer for Go.**
+
+This library implements a simple asynchronously flushing buffered writer by extending the standard `bufio.Writer`. 
diff --git a/vendor/github.com/256dpi/mercury/go.mod b/vendor/github.com/256dpi/mercury/go.mod
new file mode 100644
index 000000000..666b6b5ec
--- /dev/null
+++ b/vendor/github.com/256dpi/mercury/go.mod
@@ -0,0 +1,7 @@
+module github.com/256dpi/mercury
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/stretchr/testify v1.2.2
+)
diff --git a/vendor/github.com/256dpi/mercury/go.sum b/vendor/github.com/256dpi/mercury/go.sum
new file mode 100644
index 000000000..e03ee77d9
--- /dev/null
+++ b/vendor/github.com/256dpi/mercury/go.sum
@@ -0,0 +1,6 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
diff --git a/vendor/github.com/256dpi/mercury/mercury.go b/vendor/github.com/256dpi/mercury/mercury.go
new file mode 100644
index 000000000..3a70f8cc0
--- /dev/null
+++ b/vendor/github.com/256dpi/mercury/mercury.go
@@ -0,0 +1,111 @@
+package mercury
+
+import (
+	"bufio"
+	"io"
+	"sync"
+	"time"
+)
+
+// Writer extends a buffered writer that flushes itself asynchronously. It uses
+// a timer to flush the buffered writer it it gets stale. Errors that occur
+// during the flush are returned on the next call to Write, Flush or WriteAndFlush.
+type Writer struct {
+	w *bufio.Writer
+	d time.Duration
+	t *time.Timer
+	e error
+	m sync.Mutex
+}
+
+// NewWriter wraps the provided writer and enable buffering and asynchronous
+// flushing using the specified maximum delay.
+func NewWriter(w io.Writer, maxDelay time.Duration) *Writer {
+	return &Writer{
+		w: bufio.NewWriter(w),
+		d: maxDelay,
+	}
+}
+
+// NewWriterSize wraps the provided writer and enable buffering and asynchronous
+// flushing using the specified maximum delay. This method allows configuration
+// of the initial buffer size.
+func NewWriterSize(w io.Writer, maxDelay time.Duration, size int) *Writer {
+	return &Writer{
+		w: bufio.NewWriterSize(w, size),
+		d: maxDelay,
+	}
+}
+
+// Write implements the io.Writer interface and writes data to the underlying
+// buffered writer and flushes it asynchronously.
+func (w *Writer) Write(p []byte) (int, error) {
+	return w.write(p, false)
+}
+
+// Flush flushes the buffered writer immediately.
+func (w *Writer) Flush() error {
+	_, err := w.write(nil, true)
+	return err
+}
+
+// WriteAndFlush writes data to the underlying buffered writer and flushes it
+// immediately after writing.
+func (w *Writer) WriteAndFlush(p []byte) (int, error) {
+	return w.write(p, true)
+}
+
+func (w *Writer) write(p []byte, flush bool) (n int, err error) {
+	w.m.Lock()
+	defer w.m.Unlock()
+
+	// clear and return any error from flush
+	if w.e != nil {
+		err = w.e
+		w.e = nil
+		return 0, err
+	}
+
+	// write data if available
+	if len(p) > 0 {
+		n, err = w.w.Write(p)
+		if err != nil {
+			return n, err
+		}
+	}
+
+	// flush immediately if requested
+	if flush {
+		err = w.w.Flush()
+		if err != nil {
+			return n, err
+		}
+	}
+
+	// setup timer if data is buffered
+	if w.w.Buffered() > 0 && w.t == nil {
+		w.t = time.AfterFunc(w.d, w.flush)
+	}
+
+	// stop timer if no data is buffered
+	if w.w.Buffered() == 0 && w.t != nil {
+		w.t.Stop()
+		w.t = nil
+	}
+
+	return n, nil
+}
+
+func (w *Writer) flush() {
+	w.m.Lock()
+	defer w.m.Unlock()
+
+	// clear timer
+	w.t = nil
+
+	// flush buffer
+	err := w.w.Flush()
+	if err != nil && w.e == nil {
+		w.e = err
+	}
+}