func (c *Client) processDisconnect() error {
	c.will = nil
	atomic.StoreUint32(&c.state, clientDisconnected)
	_ = c.conn.Close()
	c.tomb.Kill(ErrClientDisconnected)
	c.backend.Log(ClientDisconnected, c, nil, nil, nil)
	return ErrClientDisconnected
}
handle an incoming disconnect packet
package broker
import (
	"errors"
	"sync"
	"sync/atomic"
	"time"
	"github.com/256dpi/gomqtt/packet"
	"github.com/256dpi/gomqtt/session"
	"github.com/256dpi/gomqtt/transport"
	"gopkg.in/tomb.v2"
)
type LogEvent string
const (
	NewConnection LogEvent = "new connection"
	PacketReceived LogEvent = "packet received"
	MessagePublished LogEvent = "message published"
	MessageAcknowledged LogEvent = "message acknowledged"
	MessageDequeued LogEvent = "message dequeued"
	MessageForwarded LogEvent = "message forwarded"
	PacketSent LogEvent = "packet sent"
	ClientDisconnected LogEvent = "client disconnected"
	TransportError LogEvent = "transport error"
	SessionError LogEvent = "session error"
	BackendError LogEvent = "backend error"
	ClientError LogEvent = "client error"
	LostConnection LogEvent = "lost connection"
)
type Session interface {
	NextID() packet.ID
	SavePacket(session.Direction, packet.Generic) error
	LookupPacket(session.Direction, packet.ID) (packet.Generic, error)
	DeletePacket(session.Direction, packet.ID) error
	AllPackets(session.Direction) ([]packet.Generic, error)
}
type Ack func()
type Backend interface {
	Authenticate(client *Client, user, password string) (ok bool, err error)
	Setup(client *Client, id string, clean bool) (a Session, resumed bool, err error)
	Restore(client *Client) error
	Subscribe(client *Client, subs []packet.Subscription, ack Ack) error
	Unsubscribe(client *Client, topics []string, ack Ack) error
	Publish(client *Client, msg *packet.Message, ack Ack) error
	Dequeue(client *Client) (*packet.Message, Ack, error)
	Terminate(client *Client) error
	Log(event LogEvent, client *Client, pkt packet.Generic, msg *packet.Message, err error)
}
var ErrUnexpectedPacket = errors.New("unexpected packet")
var ErrNotAuthorized = errors.New("not authorized")
var ErrMissingSession = errors.New("missing session")
var ErrTokenTimeout = errors.New("token timeout")
var ErrClientDisconnected = errors.New("client disconnected")
var ErrClientClosed = errors.New("client closed")
const (
	clientConnecting uint32 = iota
	clientConnected
	clientDisconnected
)
type Client struct {
	state uint32
	MaximumKeepAlive time.Duration
	ParallelPublishes int
	ParallelSubscribes int
	InflightMessages int
	TokenTimeout time.Duration
	PacketCallback func(packet.Generic) error
	Ref interface{}
	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
	closed          chan struct{}
}
func NewClient(backend Backend, conn transport.Conn) *Client {
	c := &Client{
		state:   clientConnecting,
		backend: backend,
		conn:    conn,
		closed:  make(chan struct{}),
	}
	c.tomb.Go(c.processor)
	go func() {
		_ = c.tomb.Wait()
		c.cleanup()
		close(c.closed)
	}()
	return c
}
func (c *Client) Session() Session {
	return c.session
}
func (c *Client) ID() string {
	return c.id
}
func (c *Client) Conn() transport.Conn {
	return c.conn
}
func (c *Client) Close() {
	_ = c.conn.Close()
	c.tomb.Kill(ErrClientClosed)
}
func (c *Client) Closing() <-chan struct{} {
	return c.tomb.Dying()
}
func (c *Client) Closed() <-chan struct{} {
	return c.closed
}
func (c *Client) processor() error {
	c.backend.Log(NewConnection, c, nil, nil, nil)
	pkt, err := c.conn.Receive()
	if err != nil {
		return c.die(TransportError, err)
	}
	c.backend.Log(PacketReceived, c, pkt, nil, nil)
	connect, ok := pkt.(*packet.Connect)
	if !ok {
		return c.die(ClientError, ErrUnexpectedPacket)
	}
	err = c.processConnect(connect)
	if err != nil {
		return err  
	}
	c.tomb.Go(c.dequeuer)
	c.tomb.Go(c.acker)
	for {
		if !c.tomb.Alive() {
			return tomb.ErrDying
		}
		pkt, err := c.conn.Receive()
		if err != nil {
			return c.die(TransportError, err)
		}
		c.backend.Log(PacketReceived, c, pkt, nil, nil)
		if c.PacketCallback != nil && pkt.Type() != packet.DISCONNECT {
			err = c.PacketCallback(pkt)
			if err != nil {
				return c.die(ClientError, err)
			}
		}
		err = c.processPacket(pkt)
		if err != nil {
			return err  
		}
	}
}
func (c *Client) dequeuer() error {
	for {
		select {
		case <-c.dequeueTokens:
		default:
			select {
			case <-c.dequeueTokens:
			case <-time.After(c.TokenTimeout):
				return c.die(ClientError, ErrTokenTimeout)
			case <-c.tomb.Dying():
				return tomb.ErrDying
			}
		}
		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)
		publish := packet.NewPublish()
		publish.Message = *msg
		if publish.Message.QOS > 0 {
			publish.ID = c.session.NextID()
		}
		if publish.Message.QOS > 0 {
			err := c.session.SavePacket(session.Outgoing, publish)
			if err != nil {
				return c.die(SessionError, err)
			}
		}
		if ack != nil {
			ack()
			c.backend.Log(MessageAcknowledged, c, nil, msg, nil)
		}
		err = c.send(publish, true)
		if err != nil {
			return c.die(TransportError, err)
		}
		if publish.Message.QOS == 0 {
			select {
			case c.dequeueTokens <- struct{}{}:
			default:
			}
		}
		c.backend.Log(MessageForwarded, c, nil, msg, nil)
	}
}
func (c *Client) acker() error {
	for {
		select {
		case pkt := <-c.ackQueue:
			err := c.send(pkt, true)
			if err != nil {
				return c.die(TransportError, err)
			}
			if pubcomp, ok := pkt.(*packet.Pubcomp); ok {
				err = c.session.DeletePacket(session.Incoming, pubcomp.ID)
				if err != nil {
					return c.die(SessionError, err)
				}
			}
			switch pkt.(type) {
			case *packet.Suback, *packet.Unsuback:
				select {
				case c.subscribeTokens <- struct{}{}:
				default:
				}
			case *packet.Puback, *packet.Pubcomp:
				select {
				case c.publishTokens <- struct{}{}:
				default:
				}
			}
		case <-c.tomb.Dying():
			return tomb.ErrDying
		}
	}
}
func (c *Client) processConnect(pkt *packet.Connect) error {
	c.id = pkt.ClientID
	ok, err := c.backend.Authenticate(c, pkt.Username, pkt.Password)
	if err != nil {
		return c.die(BackendError, err)
	}
	connack := packet.NewConnack()
	connack.ReturnCode = packet.ConnectionAccepted
	connack.SessionPresent = false
	if !ok {
		connack.ReturnCode = packet.NotAuthorized
		err = c.send(connack, false)
		if err != nil {
			return c.die(TransportError, err)
		}
		return c.die(ClientError, ErrNotAuthorized)
	}
	atomic.StoreUint32(&c.state, clientConnected)
	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)
	}
	if c.MaximumKeepAlive <= 0 {
		c.MaximumKeepAlive = 5 * time.Minute
	}
	requestedKeepAlive := time.Duration(pkt.KeepAlive) * time.Second
	if requestedKeepAlive == 0 || requestedKeepAlive > c.MaximumKeepAlive {
		requestedKeepAlive = c.MaximumKeepAlive
	}
	c.conn.SetReadTimeout(requestedKeepAlive + time.Duration(float64(requestedKeepAlive)*0.5))
	connack.SessionPresent = !pkt.CleanSession && resumed
	c.session = s
	if c.ParallelPublishes <= 0 {
		c.ParallelPublishes = 10
	}
	if c.ParallelSubscribes <= 0 {
		c.ParallelSubscribes = 10
	}
	if c.InflightMessages <= 0 {
		c.InflightMessages = 10
	}
	if c.TokenTimeout == 0 {
		c.TokenTimeout = 30 * time.Second
	}
	c.publishTokens = make(chan struct{}, c.ParallelPublishes)
	for i := 0; i < c.ParallelPublishes; i++ {
		c.publishTokens <- struct{}{}
	}
	c.subscribeTokens = make(chan struct{}, c.ParallelSubscribes)
	for i := 0; i < c.ParallelSubscribes; i++ {
		c.subscribeTokens <- struct{}{}
	}
	c.dequeueTokens = make(chan struct{}, c.InflightMessages)
	for i := 0; i < c.InflightMessages; i++ {
		c.dequeueTokens <- struct{}{}
	}
	c.ackQueue = make(chan packet.Generic, c.ParallelPublishes+c.ParallelSubscribes)
	if pkt.Will != nil {
		c.will = pkt.Will
	}
	err = c.send(connack, false)
	if err != nil {
		return c.die(TransportError, err)
	}
	packets, err := c.session.AllPackets(session.Outgoing)
	if err != nil {
		return c.die(SessionError, err)
	}
	for _, pkt := range packets {
		select {
		case <-c.dequeueTokens:
		default:
		}
		publish, ok := pkt.(*packet.Publish)
		if ok {
			publish.Dup = true
		}
		err = c.send(pkt, true)
		if err != nil {
			return c.die(TransportError, err)
		}
	}
	err = c.backend.Restore(c)
	if err != nil {
		return c.die(BackendError, err)
	}
	return nil
}
func (c *Client) processPacket(pkt packet.Generic) error {
	var err error
	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)
	}
	if err != nil {
		return err  
	}
	return nil
}
func (c *Client) processPingreq() error {
	err := c.send(packet.NewPingresp(), true)
	if err != nil {
		return c.die(TransportError, err)
	}
	return nil
}
func (c *Client) processSubscribe(pkt *packet.Subscribe) error {
	select {
	case <-c.subscribeTokens:
	default:
		select {
		case <-c.subscribeTokens:
		case <-time.After(c.TokenTimeout):
			return c.die(ClientError, ErrTokenTimeout)
		case <-c.tomb.Dying():
			return tomb.ErrDying
		}
	}
	suback := packet.NewSuback()
	suback.ReturnCodes = make([]packet.QOS, len(pkt.Subscriptions))
	suback.ID = pkt.ID
	for i, subscription := range pkt.Subscriptions {
		suback.ReturnCodes[i] = subscription.QOS
	}
	var once sync.Once
	ack := func() {
		once.Do(func() {
			select {
			case c.ackQueue <- suback:
			case <-c.tomb.Dying():
			}
		})
	}
	err := c.backend.Subscribe(c, pkt.Subscriptions, ack)
	if err != nil {
		return c.die(BackendError, err)
	}
	return nil
}
func (c *Client) processUnsubscribe(pkt *packet.Unsubscribe) error {
	select {
	case <-c.subscribeTokens:
	default:
		select {
		case <-c.subscribeTokens:
		case <-time.After(c.TokenTimeout):
			return c.die(ClientError, ErrTokenTimeout)
		case <-c.tomb.Dying():
			return tomb.ErrDying
		}
	}
	unsuback := packet.NewUnsuback()
	unsuback.ID = pkt.ID
	var once sync.Once
	ack := func() {
		once.Do(func() {
			select {
			case c.ackQueue <- unsuback:
			case <-c.tomb.Dying():
			}
		})
	}
	err := c.backend.Unsubscribe(c, pkt.Topics, ack)
	if err != nil {
		return c.die(BackendError, err)
	}
	return nil
}
func (c *Client) processPublish(publish *packet.Publish) error {
	if publish.Message.QOS == 0 {
		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
	}
	select {
	case <-c.publishTokens:
	default:
		select {
		case <-c.publishTokens:
		case <-time.After(c.TokenTimeout):
			return c.die(ClientError, ErrTokenTimeout)
		case <-c.tomb.Dying():
			return tomb.ErrDying
		}
	}
	if publish.Message.QOS == 1 {
		puback := packet.NewPuback()
		puback.ID = publish.ID
		var once sync.Once
		ack := func() {
			once.Do(func() {
				c.backend.Log(MessageAcknowledged, c, nil, &publish.Message, nil)
				select {
				case c.ackQueue <- puback:
				case <-c.tomb.Dying():
				}
			})
		}
		err := c.backend.Publish(c, &publish.Message, ack)
		if err != nil {
			return c.die(BackendError, err)
		}
		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)
	}
	if publish.Message.QOS == 2 {
		err := c.session.SavePacket(session.Incoming, publish)
		if err != nil {
			return c.die(SessionError, err)
		}
		pubrec := packet.NewPubrec()
		pubrec.ID = publish.ID
		err = c.send(pubrec, true)
		if err != nil {
			return c.die(TransportError, err)
		}
	}
	return nil
}
func (c *Client) processPubackAndPubcomp(id packet.ID) error {
	err := c.session.DeletePacket(session.Outgoing, id)
	if err != nil {
		return c.die(SessionError, err)
	}
	select {
	case c.dequeueTokens <- struct{}{}:
	default:
	}
	return nil
}
func (c *Client) processPubrec(id packet.ID) error {
	pubrel := packet.NewPubrel()
	pubrel.ID = id
	err := c.session.SavePacket(session.Outgoing, pubrel)
	if err != nil {
		return c.die(SessionError, err)
	}
	err = c.send(pubrel, true)
	if err != nil {
		return c.die(TransportError, err)
	}
	return nil
}
func (c *Client) processPubrel(id packet.ID) error {
	pkt, err := c.session.LookupPacket(session.Incoming, id)
	if err != nil {
		return c.die(SessionError, err)
	}
	pubcomp := packet.NewPubcomp()
	pubcomp.ID = id
	publish, ok := pkt.(*packet.Publish)
	if !ok {
		err = c.send(pubcomp, true)
		if err != nil {
			return c.die(TransportError, err)
		}
		return nil
	}
	var once sync.Once
	ack := func() {
		once.Do(func() {
			c.backend.Log(MessageAcknowledged, c, nil, &publish.Message, nil)
			select {
			case c.ackQueue <- pubcomp:
			case <-c.tomb.Dying():
			}
		})
	}
	err = c.backend.Publish(c, &publish.Message, ack)
	if err != nil {
		return c.die(BackendError, err)
	}
	c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)
	return nil
}