func (c *Client) SubscribeMultiple(subscriptions []packet.Subscription) (SubscribeFuture, error) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	if atomic.LoadUint32(&c.state) != clientConnected {
		return nil, ErrClientNotConnected
	}
	subscribe := packet.NewSubscribe()
	subscribe.ID = c.Session.NextID()
	subscribe.Subscriptions = subscriptions
	subFuture := future.New()
	c.futureStore.Put(subscribe.ID, subFuture)
	err := c.send(subscribe, true)
	if err != nil {
		return nil, c.cleanup(err, false, false)
	}
	wrappedFuture := &subscribeFuture{subFuture}
	return wrappedFuture, nil
}
SubscribeMultiple will send a Subscribe packet containing multiple topics to
subscribe. It will return a SubscribeFuture that gets completed once a
Suback packet has been received.
package client
import (
	"errors"
	"fmt"
	"net/url"
	"sync"
	"sync/atomic"
	"time"
	"github.com/256dpi/gomqtt/client/future"
	"github.com/256dpi/gomqtt/packet"
	"github.com/256dpi/gomqtt/session"
	"github.com/256dpi/gomqtt/transport"
	"gopkg.in/tomb.v2"
)
var ErrClientAlreadyConnecting = errors.New("client already connecting")
var ErrClientNotConnected = errors.New("client not connected")
var ErrClientMissingID = errors.New("client missing id")
var ErrClientConnectionDenied = errors.New("client connection denied")
var ErrClientMissingPong = errors.New("client missing pong")
var ErrClientExpectedConnack = errors.New("client expected connack")
var ErrFailedSubscription = errors.New("failed subscription")
const (
	clientInitialized uint32 = iota
	clientConnecting
	clientConnacked
	clientConnected
	clientDisconnecting
	clientDisconnected
)
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)
	Reset() error
}
type Client struct {
	state uint32
	Session Session
	Callback func(msg *packet.Message, err error) error
	Logger func(msg string)
	config        *Config
	conn          transport.Conn
	clean         bool
	earlyCallback bool
	keepAlive     time.Duration
	tracker       *Tracker
	futureStore   *future.Store
	connectFuture *future.Future
	tomb          tomb.Tomb
	mutex         sync.Mutex
	finish        sync.Once
}
func New() *Client {
	return &Client{
		state:       clientInitialized,
		Session:     session.NewMemorySession(),
		futureStore: future.NewStore(),
	}
}
func (c *Client) Connect(config *Config) (ConnectFuture, error) {
	if config == nil {
		panic("missing config")
	}
	c.mutex.Lock()
	defer c.mutex.Unlock()
	if atomic.LoadUint32(&c.state) >= clientConnecting {
		return nil, ErrClientAlreadyConnecting
	}
	c.config = config
	urlParts, err := url.ParseRequestURI(config.BrokerURL)
	if err != nil {
		return nil, err
	}
	if !config.CleanSession && config.ClientID == "" {
		return nil, ErrClientMissingID
	}
	keepAlive, err := time.ParseDuration(config.KeepAlive)
	if err != nil {
		return nil, err
	}
	c.keepAlive = keepAlive
	c.tracker = NewTracker(keepAlive)
	if config.Dialer != nil {
		c.conn, err = config.Dialer.Dial(config.BrokerURL)
	} else {
		c.conn, err = transport.Dial(config.BrokerURL)
	}
	if err != nil {
		return nil, err
	}
	c.conn.SetReadLimit(c.config.ReadLimit)
	c.conn.SetMaxWriteDelay(c.config.MaxWriteDelay)
	atomic.StoreUint32(&c.state, clientConnecting)
	c.clean = config.CleanSession
	if c.clean {
		err = c.Session.Reset()
		if err != nil {
			return nil, c.cleanup(err, true, false)
		}
	}
	c.earlyCallback = config.AlwaysAnnounceOnPublish
	connect := packet.NewConnect()
	connect.ClientID = config.ClientID
	connect.KeepAlive = uint16(keepAlive.Seconds())
	connect.CleanSession = config.CleanSession
	if urlParts.User != nil {
		connect.Username = urlParts.User.Username()
		connect.Password, _ = urlParts.User.Password()
	}
	connect.Will = config.WillMessage
	c.connectFuture = future.New()
	err = c.send(connect, false)
	if err != nil {
		return nil, c.cleanup(err, false, false)
	}
	c.tomb.Go(c.processor)
	wrappedFuture := &connectFuture{c.connectFuture}
	return wrappedFuture, nil
}
func (c *Client) Publish(topic string, payload []byte, qos packet.QOS, retain bool) (GenericFuture, error) {
	return c.PublishMessage(&packet.Message{
		Topic:   topic,
		Payload: payload,
		QOS:     qos,
		Retain:  retain,
	})
}
func (c *Client) PublishMessage(msg *packet.Message) (GenericFuture, error) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	if atomic.LoadUint32(&c.state) != clientConnected {
		return nil, ErrClientNotConnected
	}
	publish := packet.NewPublish()
	publish.Message = *msg
	if msg.QOS > 0 {
		publish.ID = c.Session.NextID()
	}
	publishFuture := future.New()
	c.futureStore.Put(publish.ID, publishFuture)
	if msg.QOS > 0 {
		err := c.Session.SavePacket(session.Outgoing, publish)
		if err != nil {
			return nil, c.cleanup(err, true, false)
		}
	}
	err := c.send(publish, true)
	if err != nil {
		return nil, c.cleanup(err, false, false)
	}
	if msg.QOS == 0 {
		publishFuture.Complete(nil)
		c.futureStore.Delete(publish.ID)
	}
	return publishFuture, nil
}
func (c *Client) Subscribe(topic string, qos packet.QOS) (SubscribeFuture, error) {
	return c.SubscribeMultiple([]packet.Subscription{
		{Topic: topic, QOS: qos},
	})
}