func (cli *Client) handleSUBACK(p packet.Packet) error {
	cli.muConn.Lock()
	cli.muSess.Lock()
	defer cli.muConn.Unlock()
	defer cli.muSess.Unlock()
	id := p.(*packet.SUBACK).PacketID
	if err := cli.validatePacketID(cli.sess.sendingPackets, id, packet.TypeSUBSCRIBE); err != nil {
		return err
	}
	subreqs := cli.sess.sendingPackets[id].(*packet.SUBSCRIBE).SubReqs
	delete(cli.sess.sendingPackets, id)
	returnCodes := p.(*packet.SUBACK).ReturnCodes
	if len(returnCodes) != len(subreqs) {
		return ErrInvalidSUBACK
	}
	for i, code := range returnCodes {
		if code == packet.SUBACKRetFailure {
			continue
		}
		topicFilter := string(subreqs[i].TopicFilter)
		cli.conn.ackedSubs[topicFilter] = cli.conn.unackSubs[topicFilter]
		delete(cli.conn.unackSubs, topicFilter)
	}
	return nil
}
handleSUBACK handles the SUBACK Packet.
package client
import (
	"errors"
	"io"
	"strings"
	"sync"
	"time"
	"github.com/yosssi/gmq/mqtt"
	"github.com/yosssi/gmq/mqtt/packet"
)
const (
	minPacketID uint16 = 1
	maxPacketID uint16 = 65535
)
var (
	ErrAlreadyConnected = errors.New("the Client has already connected to the Server")
	ErrNotYetConnected  = errors.New("the Client has not yet connected to the Server")
	ErrCONNACKTimeout   = errors.New("the CONNACK Packet was not received within a reasonalbe amount of time")
	ErrPINGRESPTimeout  = errors.New("the PINGRESP Packet was not received within a reasonalbe amount of time")
	ErrPacketIDExhaused = errors.New("Packet Identifiers are exhausted")
	ErrInvalidPINGRESP  = errors.New("invalid PINGRESP Packet")
	ErrInvalidSUBACK    = errors.New("invalid SUBACK Packet")
)
type Client struct {
	muConn sync.RWMutex
	conn *connection
	muSess sync.RWMutex
	sess *session
	wg sync.WaitGroup
	disconnc chan struct{}
	disconnEndc chan struct{}
	errorHandler ErrorHandler
}
func (cli *Client) Connect(opts *ConnectOptions) error {
	cli.muConn.Lock()
	defer cli.muConn.Unlock()
	if cli.conn != nil {
		return ErrAlreadyConnected
	}
	if opts == nil {
		opts = &ConnectOptions{}
	}
	conn, err := newConnection(opts.Network, opts.Address, opts.TLSConfig)
	if err != nil {
		return err
	}
	cli.conn = conn
	cli.muSess.Lock()
	if opts.CleanSession || cli.sess == nil {
		cli.sess = newSession(opts.CleanSession, opts.ClientID)
	} else {
		opts.ClientID = cli.sess.clientID
	}
	cli.muSess.Unlock()
	err = cli.sendCONNECT(&packet.CONNECTOptions{
		ClientID:     opts.ClientID,
		UserName:     opts.UserName,
		Password:     opts.Password,
		CleanSession: opts.CleanSession,
		KeepAlive:    opts.KeepAlive,
		WillTopic:    opts.WillTopic,
		WillMessage:  opts.WillMessage,
		WillQoS:      opts.WillQoS,
		WillRetain:   opts.WillRetain,
	})
	if err != nil {
		cli.conn.Close()
		cli.clean()
		return err
	}
	cli.conn.wg.Add(1)
	go cli.waitPacket(cli.conn.connack, opts.CONNACKTimeout, ErrCONNACKTimeout)
	cli.conn.wg.Add(1)
	go cli.receivePackets()
	cli.conn.wg.Add(1)
	go cli.sendPackets(time.Duration(opts.KeepAlive), opts.PINGRESPTimeout)
	if !opts.CleanSession {
		cli.muSess.Lock()
		defer cli.muSess.Unlock()
		for id, p := range cli.sess.sendingPackets {
			ptype, err := p.Type()
			if err != nil {
				return err
			}
			switch ptype {
			case packet.TypePUBLISH:
				p.(*packet.PUBLISH).DUP = true
				cli.conn.send <- p
			case packet.TypePUBREL:
				cli.conn.send <- p
			default:
				delete(cli.sess.sendingPackets, id)
			}
		}
	}
	return nil
}
func (cli *Client) Disconnect() error {
	cli.muConn.Lock()
	if cli.conn == nil {
		cli.muConn.Unlock()
		return ErrNotYetConnected
	}
	cli.send(packet.NewDISCONNECT())
	if err := cli.conn.Close(); err != nil {
		cli.muConn.Unlock()
		return err
	}
	cli.conn.disconnected = true
	select {
	case cli.conn.sendEnd <- struct{}{}:
	default:
	}
	cli.muConn.Unlock()
	cli.conn.wg.Wait()
	cli.muConn.Lock()
	cli.muSess.Lock()
	cli.clean()
	cli.muSess.Unlock()
	cli.muConn.Unlock()
	return nil
}
func (cli *Client) Publish(opts *PublishOptions) error {
	cli.muConn.RLock()
	defer cli.muConn.RUnlock()
	if cli.conn == nil {
		return ErrNotYetConnected
	}
	if opts == nil {
		opts = &PublishOptions{}
	}
	p, err := cli.newPUBLISHPacket(opts)
	if err != nil {
		return err
	}
	cli.conn.send <- p
	return nil
}
func (cli *Client) Subscribe(opts *SubscribeOptions) error {
	cli.muConn.Lock()
	defer cli.muConn.Unlock()
	if cli.conn == nil {
		return ErrNotYetConnected
	}
	if opts == nil || len(opts.SubReqs) == 0 {
		return packet.ErrInvalidNoSubReq
	}
	var packetID uint16
	var err error
	cli.muSess.Lock()
	defer cli.muSess.Unlock()
	if packetID, err = cli.generatePacketID(); err != nil {
		return err
	}
	var subReqs []*packet.SubReq
	for _, s := range opts.SubReqs {
		subReqs = append(subReqs, &packet.SubReq{
			TopicFilter: s.TopicFilter,
			QoS:         s.QoS,
		})
	}
	p, err := packet.NewSUBSCRIBE(&packet.SUBSCRIBEOptions{
		PacketID: packetID,
		SubReqs:  subReqs,
	})
	if err != nil {
		return err
	}
	cli.sess.sendingPackets[packetID] = p
	for _, s := range opts.SubReqs {
		cli.conn.unackSubs[string(s.TopicFilter)] = s.Handler
	}
	cli.conn.send <- p
	return nil
}
func (cli *Client) Unsubscribe(opts *UnsubscribeOptions) error {
	cli.muConn.Lock()
	defer cli.muConn.Unlock()
	if cli.conn == nil {
		return ErrNotYetConnected
	}
	if opts == nil || len(opts.TopicFilters) == 0 {
		return packet.ErrNoTopicFilter
	}
	var packetID uint16
	var err error
	cli.muSess.Lock()
	defer cli.muSess.Unlock()
	if packetID, err = cli.generatePacketID(); err != nil {
		return err
	}
	p, err := packet.NewUNSUBSCRIBE(&packet.UNSUBSCRIBEOptions{
		PacketID:     packetID,
		TopicFilters: opts.TopicFilters,
	})
	if err != nil {
		return err
	}
	cli.sess.sendingPackets[packetID] = p
	cli.conn.send <- p
	return nil
}
func (cli *Client) Terminate() {
	cli.disconnEndc <- struct{}{}
	cli.wg.Wait()
}
func (cli *Client) send(p packet.Packet) error {
	if cli.conn == nil {
		return ErrNotYetConnected
	}
	if _, err := p.WriteTo(cli.conn.w); err != nil {
		return err
	}
	return cli.conn.w.Flush()
}
func (cli *Client) sendCONNECT(opts *packet.CONNECTOptions) error {
	if opts == nil {
		opts = &packet.CONNECTOptions{}
	}
	p, err := packet.NewCONNECT(opts)
	if err != nil {
		return err
	}
	return cli.send(p)
}
func (cli *Client) receive() (packet.Packet, error) {
	if cli.conn == nil {
		return nil, ErrNotYetConnected
	}
	b, err := cli.conn.r.ReadByte()
	if err != nil {
		return nil, err
	}
	fixedHeader := packet.FixedHeader([]byte{b})
	var mp uint32 = 1  
	var rl uint32      
	for {
		b, err = cli.conn.r.ReadByte()
		if err != nil {
			return nil, err
		}
		fixedHeader = append(fixedHeader, b)
		rl += uint32(b&0x7F) * mp
		if b&0x80 == 0 {
			break
		}
		mp *= 128
	}
	remaining := make([]byte, rl)
	if rl > 0 {
		if _, err = io.ReadFull(cli.conn.r, remaining); err != nil {
			return nil, err
		}
	}
	return packet.NewFromBytes(fixedHeader, remaining)
}
func (cli *Client) clean() {
	cli.conn = nil
	if cli.sess != nil && cli.sess.cleanSession {
		cli.sess = nil
	}
}
func (cli *Client) waitPacket(packetc <-chan struct{}, timeout time.Duration, errTimeout error) {
	defer cli.conn.wg.Done()
	var timeoutc <-chan time.Time
	if timeout > 0 {
		timeoutc = time.After(timeout * time.Second)
	}
	select {
	case <-packetc:
	case <-timeoutc:
		cli.handleErrorAndDisconn(errTimeout)
	}
}
func (cli *Client) receivePackets() {
	defer func() {
		close(cli.conn.connack)
		cli.conn.wg.Done()
	}()
	for {
		p, err := cli.receive()
		if err != nil {
			cli.handleErrorAndDisconn(err)
			return
		}
		if err := cli.handlePacket(p); err != nil {
			cli.handleErrorAndDisconn(err)
			return
		}
	}
}
func (cli *Client) handlePacket(p packet.Packet) error {
	ptype, err := p.Type()
	if err != nil {
		return err
	}
	switch ptype {
	case packet.TypeCONNACK:
		cli.handleCONNACK()
		return nil
	case packet.TypePUBLISH:
		return cli.handlePUBLISH(p)
	case packet.TypePUBACK:
		return cli.handlePUBACK(p)
	case packet.TypePUBREC:
		return cli.handlePUBREC(p)
	case packet.TypePUBREL:
		return cli.handlePUBREL(p)
	case packet.TypePUBCOMP:
		return cli.handlePUBCOMP(p)
	case packet.TypeSUBACK:
		return cli.handleSUBACK(p)
	case packet.TypeUNSUBACK:
		return cli.handleUNSUBACK(p)
	case packet.TypePINGRESP:
		return cli.handlePINGRESP()
	default:
		return packet.ErrInvalidPacketType
	}
}
func (cli *Client) handleCONNACK() {
	select {
	case cli.conn.connack <- struct{}{}:
	default:
	}
}
func (cli *Client) handlePUBLISH(p packet.Packet) error {
	publish := p.(*packet.PUBLISH)
	switch publish.QoS {
	case mqtt.QoS0:
		cli.muConn.RLock()
		defer cli.muConn.RUnlock()
		cli.handleMessage(publish.TopicName, publish.Message)
		return nil
	case mqtt.QoS1:
		cli.muConn.RLock()
		defer cli.muConn.RUnlock()
		cli.handleMessage(publish.TopicName, publish.Message)
		puback, err := packet.NewPUBACK(&packet.PUBACKOptions{
			PacketID: publish.PacketID,
		})
		if err != nil {
			return err
		}
		cli.conn.send <- puback
		return nil
	default:
		cli.muSess.Lock()
		defer cli.muSess.Unlock()
		if _, exist := cli.sess.receivingPackets[publish.PacketID]; exist {
			return packet.ErrInvalidPacketID
		}
		cli.sess.receivingPackets[publish.PacketID] = p
		pubrec, err := packet.NewPUBREC(&packet.PUBRECOptions{
			PacketID: publish.PacketID,
		})
		if err != nil {
			return err
		}
		cli.conn.send <- pubrec
		return nil
	}
}
func (cli *Client) handlePUBACK(p packet.Packet) error {
	cli.muSess.Lock()
	defer cli.muSess.Unlock()
	id := p.(*packet.PUBACK).PacketID
	if err := cli.validatePacketID(cli.sess.sendingPackets, id, packet.TypePUBLISH); err != nil {
		return err
	}
	delete(cli.sess.sendingPackets, id)
	return nil
}
func (cli *Client) handlePUBREC(p packet.Packet) error {
	cli.muSess.Lock()
	defer cli.muSess.Unlock()
	id := p.(*packet.PUBREC).PacketID
	if err := cli.validatePacketID(cli.sess.sendingPackets, id, packet.TypePUBLISH); err != nil {
		return err
	}
	pubrel, err := packet.NewPUBREL(&packet.PUBRELOptions{
		PacketID: id,
	})
	if err != nil {
		return err
	}
	cli.sess.sendingPackets[id] = pubrel
	cli.conn.send <- pubrel
	return nil
}
func (cli *Client) handlePUBREL(p packet.Packet) error {
	cli.muSess.Lock()
	defer cli.muSess.Unlock()
	id := p.(*packet.PUBREL).PacketID
	if err := cli.validatePacketID(cli.sess.receivingPackets, id, packet.TypePUBLISH); err != nil {
		return err
	}
	publish := cli.sess.receivingPackets[id].(*packet.PUBLISH)
	cli.muConn.RLock()
	cli.handleMessage(publish.TopicName, publish.Message)
	cli.muConn.RUnlock()
	delete(cli.sess.receivingPackets, id)
	pubcomp, err := packet.NewPUBCOMP(&packet.PUBCOMPOptions{
		PacketID: id,
	})
	if err != nil {
		return err
	}
	cli.conn.send <- pubcomp
	return nil
}
func (cli *Client) handlePUBCOMP(p packet.Packet) error {
	cli.muSess.Lock()
	defer cli.muSess.Unlock()
	id := p.(*packet.PUBCOMP).PacketID
	if err := cli.validatePacketID(cli.sess.sendingPackets, id, packet.TypePUBREL); err != nil {
		return err
	}
	delete(cli.sess.sendingPackets, id)
	return nil
}