server.go

// package broker provides a MQTT 3.1.1 compliant MQTT broker.
package broker

import (
	"errors"
	"fmt"
	"io"
	"net"
	"strconv"
	"sync/atomic"
	"time"

	"github.com/crtrpt/mqtt/broker/events"
	"github.com/crtrpt/mqtt/broker/internal/circ"
	"github.com/crtrpt/mqtt/broker/internal/clients"
	"github.com/crtrpt/mqtt/broker/internal/packets"
	"github.com/crtrpt/mqtt/broker/internal/topics"
	"github.com/crtrpt/mqtt/broker/internal/utils"
	"github.com/crtrpt/mqtt/broker/listeners"
	"github.com/crtrpt/mqtt/broker/listeners/auth"
	"github.com/crtrpt/mqtt/broker/persistence"
	"github.com/crtrpt/mqtt/broker/system"
)

const (
	// 当前服务器的版本
	Version = "1.1.1"

	// defaultInflightTTL is the number of seconds a pending inflight message should last.
	// 默认飞行时间  如果一条消息还没有到达那么 就删除掉
	defaultInflightTTL int64 = 60 * 60 * 24
)

var (
	// ErrListenerIDExists indicates that a listener with the same id already exists.
	ErrListenerIDExists = errors.New("监听器已经退出")

	// ErrReadConnectInvalid indicates that the connection packet was invalid.
	ErrReadConnectInvalid = errors.New("connect packet 错误")

	// ErrConnectNotAuthorized indicates that the connection packet had incorrect auth values.
	ErrConnectNotAuthorized = errors.New("授权错误")

	// ErrInvalidTopic indicates that the specified topic was not valid.
	ErrInvalidTopic = errors.New("cannot publish to $ and $SYS topics")

	// ErrRejectPacket indicates that a packet should be dropped instead of processed.
	ErrRejectPacket = errors.New("packet 被拒绝")

	// ErrClientDisconnect indicates that a client disconnected from the server.
	ErrClientDisconnect = errors.New("客户端断开连接")

	// ErrClientReconnect indicates that a client attempted to reconnect while still connected.
	ErrClientReconnect = errors.New("client sent connect while connected")

	// ErrServerShutdown is propagated when the server shuts down.
	ErrServerShutdown = errors.New("服务器已经下线")

	// ErrSessionReestablished indicates that an existing client was replaced by a newly connected
	// client. The existing client is disconnected.
	ErrSessionReestablished = errors.New("client session re-established")

	// ErrConnectionFailed indicates that a client connection attempt failed for other reasons.
	ErrConnectionFailed = errors.New("connection attempt failed")

	// 定时发送$Sys 间隔
	SysTopicInterval time.Duration = 30000

	// 消息重发间隔 最多6小时内重发
	inflightResendBackoff = []int64{0, 1, 2, 10, 60, 120, 600, 3600, 21600}

	// QoS 重发时间
	inflightMaxResends = 6
)

// 使用 server.New() 初始化 broker
type Server struct {
	inline               inlineMessages       // 内部消息 服务器直接发给 client的消息
	Events               events.Events        // 事件回调列表
	Store                persistence.Store    // 持久化后端.
	Options              *Options             // 服务器配置信息.
	Listeners            *listeners.Listeners // tcp服务监听器.
	Clients              *clients.Clients     // 当前broker下面的全部client
	Topics               *topics.Index        // topic树
	System               *system.Info         // 系统的信息 $SYS 内容.
	bytepool             *circ.BytesPool      // 字节池
	sysTicker            *time.Ticker         // 定时更新 $SYS .
	inflightExpiryTicker *time.Ticker         // 定时删除过期消息.
	inflightResendTicker *time.Ticker         // 消息定时重发.
	done                 chan bool            // broker是否下线.
}

type Options struct {
	// BufferSize overrides the default buffer size (circ.DefaultBufferSize) for the client buffers.
	BufferSize int

	// BufferBlockSize overrides the default buffer block size (DefaultBlockSize) for the client buffers.
	BufferBlockSize int

	// 消息飞行时间
	InflightTTL int64
}

// inlineMessages contains channels for handling inline (direct) publishing.
type inlineMessages struct {
	done chan bool           // indicate that the server is ending.
	pub  chan packets.Packet // a channel of packets to publish to clients
}

// 仅限测试用
func New() *Server {
	return NewServer(nil)
}

// 返回一个server实例
func NewServer(opts *Options) *Server {
	if opts == nil {
		opts = new(Options)
	}

	if opts.InflightTTL < 1 {
		opts.InflightTTL = defaultInflightTTL
	}

	s := &Server{
		done: make(chan bool),
		//字节池
		bytepool: circ.NewBytesPool(opts.BufferSize),
		//client列表
		Clients: clients.New(),
		//topic树
		Topics: topics.New(),
		//服务器信息
		System: &system.Info{
			Version: Version,
			Started: time.Now().Unix(),
		},
		//定时发送服务器信息
		sysTicker: time.NewTicker(SysTopicInterval * time.Millisecond),
		//定时清楚过期消息
		inflightExpiryTicker: time.NewTicker(time.Duration(opts.InflightTTL) * time.Second),
		//10纳秒跑一次重发队列
		inflightResendTicker: time.NewTicker(time.Duration(10) * time.Second),
		//内联消息
		inline: inlineMessages{
			done: make(chan bool),
			pub:  make(chan packets.Packet, 4096),
		},
		//事件列表
		Events: events.Events{},
		//参数
		Options: opts,
	}

	// Expose server stats using the system listener so it can be used in the
	// dashboard and other more experimental listeners.
	s.Listeners = listeners.New(s.System)

	return s
}

// 设置持久化后端
// 必须在启动server之前调用.
func (s *Server) AddStore(p persistence.Store) error {
	s.Store = p
	s.Store.SetInflightTTL(s.Options.InflightTTL)

	err := s.Store.Open()
	if err != nil {
		return err
	}

	return nil
}

// 开启多个网络监听器
func (s *Server) AddListener(listener listeners.Listener, config *listeners.Config) error {
	if _, ok := s.Listeners.Get(listener.ID()); ok {
		return ErrListenerIDExists
	}

	if config != nil {
		listener.SetConfig(config)
	}

	s.Listeners.Add(listener)
	err := listener.Listen(s.System)
	if err != nil {
		return err
	}

	return nil
}

// 开始事件调度
// on all attached listeners, and publishing the system topics.
func (s *Server) Serve() error {
	if s.Store != nil {
		err := s.readStore()
		if err != nil {
			return err
		}
	}
	//开启事件循环
	go s.eventLoop() // spin up event loop for issuing $SYS values and closing server.

	go s.inlineClient()                         // spin up inline client for direct message publishing.
	s.Listeners.ServeAll(s.EstablishConnection) // start listening on all listeners.
	s.publishSysTopics()                        // begin publishing $SYS system values.

	return nil
}

// 开启时间调度, running various server processes at different intervals.
func (s *Server) eventLoop() {
	for {
		select {
		case <-s.done:
			s.sysTicker.Stop()
			close(s.inline.done)
			return
		//定时发布系统状态
		case <-s.sysTicker.C:
			s.publishSysTopics()
		//删除过期消息
		case <-s.inflightExpiryTicker.C:
			s.clearExpiredInflights(time.Now().Unix())
		//重发消息
		case <-s.inflightResendTicker.C:
			s.resendPendingInflights()
		}
	}
}

// inlineClient loops forever, sending directly-published messages
// from the Publish method to subscribers.
func (s *Server) inlineClient() {
	for {
		select {
		case <-s.inline.done:
			close(s.inline.pub)
			return
		case pk := <-s.inline.pub:
			s.publishToSubscribers(pk)
		}
	}
}

// 读取连接的第一个传入标头，如果可接受，则返回有效的连接数据包。
func (s *Server) readConnectionPacket(cl *clients.Client) (pk packets.Packet, err error) {
	fh := new(packets.FixedHeader)
	err = cl.ReadFixedHeader(fh)
	if err != nil {
		return
	}

	pk, err = cl.ReadPacket(fh)
	if err != nil {
		return
	}

	//如果类型错误 返回连接
	if pk.FixedHeader.Type != packets.Connect {
		return pk, ErrReadConnectInvalid
	}

	return
}

// onError is a pass-through method which triggers the OnError
// event hook (if applicable), and returns the provided error.
func (s *Server) onError(cl events.Client, err error) error {
	if err == nil {
		return err
	}
	// Note: if the error originates from a real cause, it will
	// have been captured as the StopCause. The two cases ignored
	// below are ordinary consequences of closing the connection.
	// If one of these ordinary conditions stops the connection,
	// then the client closed or broke the connection.
	if s.Events.OnError != nil &&
		!errors.Is(err, io.EOF) {
		s.Events.OnError(cl, err)
	}

	return err
}

// onStorage is a pass-through method which delegates errors from
// the persistent storage adapter to the onError event hook.
func (s *Server) onStorage(cl events.Clientlike, err error) {
	if err == nil {
		return
	}

	_ = s.onError(cl.Info(), fmt.Errorf("storage: %w", err))
}

// 处理每一个新的客户端连接
func (s *Server) EstablishConnection(lid string, c net.Conn, ac auth.Controller) error {

	xbr := s.bytepool.Get() // 从连接池获取一个接收的
	xbw := s.bytepool.Get() // 从连接池获取一个发送的
	defer s.bytepool.Put(xbr)
	defer s.bytepool.Put(xbw)

	//创建一个客户端映射
	cl := clients.NewClient(c,
		circ.NewReaderFromSlice(s.Options.BufferBlockSize, xbr),
		circ.NewWriterFromSlice(s.Options.BufferBlockSize, xbw),
		s.System,
	)

	cl.Start()
	defer cl.ClearBuffers()
	defer cl.Stop(nil)

	//从客户端读取数据
	pk, err := s.readConnectionPacket(cl)
	if err != nil {
		return s.onError(cl.Info(), fmt.Errorf("读取连接数据包异常: %w", err))
	}

	//对连接进行验证
	ackCode, err := pk.ConnectValidate()
	if err != nil {
		if err := s.ackConnection(cl, ackCode, false); err != nil {
			return s.onError(cl.Info(), fmt.Errorf("发送非法连接 数据包: %w", err))
		}
		return s.onError(cl.Info(), fmt.Errorf("验证连接数据包的时候发生了异常: %w", err))
	}

	//设置客户端的唯一标识符
	cl.Identify(lid, pk, ac) // Set client identity values from the connection packet.

	//开始用户名密码认证
	if !ac.Authenticate(pk.Username, pk.Password) {
		//返回授权认证错误给客户端
		if err := s.ackConnection(cl, packets.CodeConnectBadAuthValues, false); err != nil {
			return s.onError(cl.Info(), fmt.Errorf("invalid connection send ack: %w", err))
		}
		return s.onError(cl.Info(), ErrConnectionFailed)
	}
	//连接总数+1
	atomic.AddInt64(&s.System.ConnectionsTotal, 1)
	//客户端总数+1;
	atomic.AddInt64(&s.System.ClientsConnected, 1)
	//断开连接后 客户端总数-1
	defer atomic.AddInt64(&s.System.ClientsConnected, -1)
	//断开连接后 断开连接数+1
	defer atomic.AddInt64(&s.System.ClientsDisconnected, 1)

	//授权完成之后 获取 client session
	sessionPresent := s.inheritClientSession(pk, cl)
	//broker 客户端+1
	s.Clients.Add(cl)
	//给 client返回 ack
	err = s.ackConnection(cl, ackCode, sessionPresent)

	if err != nil {
		return s.onError(cl.Info(), fmt.Errorf("ack connection packet: %w", err))
	}

	if sessionPresent {
		// 重新发送正在处理的消息
		err = s.ResendClientInflight(cl, true)
		if err != nil {
			s.onError(cl.Info(), fmt.Errorf("resend in flight: %w", err)) // pass-through, no return.
		}
	}

	if s.Store != nil {
		//持久化 client信息
		s.onStorage(cl, s.Store.WriteClient(persistence.Client{
			ID:       "cl_" + cl.ID,
			ClientID: cl.ID,
			T:        persistence.KClient,
			Listener: cl.Listener,
			Username: cl.Username,
			LWT:      persistence.LWT(cl.LWT),
		}))
	}

	//触发 connect 事件
	if s.Events.OnConnect != nil {
		s.Events.OnConnect(cl.Info(), events.Packet(pk))
	}
	// 现在连接状态已经变成正常状态了
	//处理消息---
	if err := cl.Read(s.processPacket); err != nil {
		//如果发生了错误 发送遗言 并且断开客户端连接
		s.sendLWT(cl)
		cl.Stop(err)
	}

	//获取断开原因
	err = cl.StopCause() // Determine true cause of stop.

	//如果清楚连接状态的话执行相应的逻辑
	if cl.CleanSession {
		s.clearAbandonedInflights(cl)
	}

	//触发断开连接逻辑
	if s.Events.OnDisconnect != nil {
		s.Events.OnDisconnect(cl.Info(), err)
	}

	return err
}

// 发送 connack数据包
func (s *Server) ackConnection(cl *clients.Client, ack byte, present bool) error {
	return s.writeClient(cl, packets.Packet{
		FixedHeader: packets.FixedHeader{
			Type: packets.Connack,
		},
		SessionPresent: present,
		ReturnCode:     ack,
	})
}

// 如果当前的 cleanSession 为true 复用以前的 下一个client 也会复用当前的
// cleanSession 为 true 复用以前的 否则 不复用以前的
func (s *Server) inheritClientSession(pk packets.Packet, cl *clients.Client) bool {
	if existing, ok := s.Clients.Get(pk.ClientIdentifier); ok {
		existing.Lock()
		defer existing.Unlock()
		//如果client相同 停止老的客户端
		existing.Stop(ErrSessionReestablished) // Issue a stop on the old client.

		// Per [MQTT-3.1.2-6]:
		// If CleanSession is set to 1, the Client and Server MUST discard any previous Session and start a new one.
		// The state associated with a CleanSession MUST NOT be reused in any subsequent session.
		// 如果当前不复用 或者 老的client 不复用的话
		if pk.CleanSession || existing.CleanSession {
			//退订全部订阅
			s.unsubscribeClient(existing)
			//清楚未完成的消息
			s.clearAbandonedInflights(existing)
			return false
		}
		//否则不清楚老的订阅和消息继续处理
		cl.Inflight = existing.Inflight // Take address of existing session.
		cl.Subscriptions = existing.Subscriptions
		return true

	} else {
		//全新的客户端 增加总客户端数量
		atomic.AddInt64(&s.System.ClientsTotal, 1)
		//如果系统客户端连接数>系统客户端最大连接数量
		if atomic.LoadInt64(&s.System.ClientsConnected) > atomic.LoadInt64(&s.System.ClientsMax) {
			//系统客户端最大连接数+1
			atomic.AddInt64(&s.System.ClientsMax, 1)
		}
		return false
	}
}

// unsubscribeClient unsubscribes a client from all of their subscriptions.
func (s *Server) unsubscribeClient(cl *clients.Client) {
	for k := range cl.Subscriptions {
		delete(cl.Subscriptions, k)
		if s.Topics.Unsubscribe(k, cl.ID) {
			if s.Events.OnUnsubscribe != nil {
				s.Events.OnUnsubscribe(k, cl.Info())
			}
			atomic.AddInt64(&s.System.Subscriptions, -1)
		}
	}
}

// writeClient writes packets to a client connection.
func (s *Server) writeClient(cl *clients.Client, pk packets.Packet) error {
	_, err := cl.WritePacket(pk)
	if err != nil {
		return fmt.Errorf("write: %w", err)
	}

	return nil
}

// processPacket processes an inbound packet for a client. Since the method is
// typically called as a goroutine, errors are primarily for test checking purposes.
func (s *Server) processPacket(cl *clients.Client, pk packets.Packet) error {
	switch pk.FixedHeader.Type {
	case packets.Connect:
		return s.processConnect(cl, pk)
	case packets.Disconnect:
		return s.processDisconnect(cl, pk)
	case packets.Pingreq:
		return s.processPingreq(cl, pk)
	case packets.Publish:
		r, err := pk.PublishValidate()
		if r != packets.Accepted {
			return err
		}
		return s.processPublish(cl, pk)
	case packets.Puback:
		return s.processPuback(cl, pk)
	case packets.Pubrec:
		return s.processPubrec(cl, pk)
	case packets.Pubrel:
		return s.processPubrel(cl, pk)
	case packets.Pubcomp:
		return s.processPubcomp(cl, pk)
	case packets.Subscribe:
		r, err := pk.SubscribeValidate()
		if r != packets.Accepted {
			return err
		}
		return s.processSubscribe(cl, pk)
	case packets.Unsubscribe:
		r, err := pk.UnsubscribeValidate()
		if r != packets.Accepted {
			return err
		}
		return s.processUnsubscribe(cl, pk)
	default:
		return fmt.Errorf("No valid packet available; %v", pk.FixedHeader.Type)
	}
}

// processConnect processes a Connect packet. The packet cannot be used to
// establish a new connection on an existing connection. See EstablishConnection
// instead.
func (s *Server) processConnect(cl *clients.Client, pk packets.Packet) error {
	s.sendLWT(cl)
	cl.Stop(ErrClientReconnect)
	return nil
}

// processDisconnect processes a Disconnect packet.
func (s *Server) processDisconnect(cl *clients.Client, pk packets.Packet) error {
	cl.Stop(ErrClientDisconnect)
	return nil
}

// processPingreq processes a Pingreq packet.
func (s *Server) processPingreq(cl *clients.Client, pk packets.Packet) error {
	err := s.writeClient(cl, packets.Packet{
		FixedHeader: packets.FixedHeader{
			Type: packets.Pingresp,
		},
	})
	if err != nil {
		return err
	}

	return nil
}

// Publish creates a publish packet from a payload and sends it to the inline.pub
// channel, where it is written directly to the outgoing byte buffers of any
// clients subscribed to the given topic. Because the message is written directly
// within the server, QoS is inherently 2 (exactly once).
func (s *Server) Publish(topic string, payload []byte, retain bool) error {
	if len(topic) >= 4 && topic[0:4] == "$SYS" {
		return ErrInvalidTopic
	}

	pk := packets.Packet{
		FixedHeader: packets.FixedHeader{
			Type:   packets.Publish,
			Retain: retain,
		},
		TopicName: topic,
		Payload:   payload,
	}

	if retain {
		s.retainMessage(&s.inline, pk)
	}

	// handoff packet to s.inline.pub channel for writing to client buffers
	// to avoid blocking the calling function.
	s.inline.pub <- pk

	return nil
}

// Info provides pseudo-client information for the inline messages processor.
// It provides a 'client' to which inline retained messages can be assigned.
func (*inlineMessages) Info() events.Client {
	return events.Client{
		ID:       "inline",
		Remote:   "inline",
		Listener: "inline",
	}
}

// processPublish processes a Publish packet.
func (s *Server) processPublish(cl *clients.Client, pk packets.Packet) error {
	if len(pk.TopicName) >= 4 && pk.TopicName[0:4] == "$SYS" {
		return nil // Clients can't publish to $SYS topics, so fail silently as per spec.
	}

	if !cl.AC.ACL(cl.Username, pk.TopicName, true) {
		return nil
	}

	// if an OnProcessMessage hook exists, potentially modify the packet.
	if s.Events.OnProcessMessage != nil {
		pkx, err := s.Events.OnProcessMessage(cl.Info(), events.Packet(pk))
		if err == nil {
			pk = packets.Packet(pkx) // Only use the new package changes if there's no errors.
		} else {
			// If the ErrRejectPacket is return, abandon processing the packet.
			if err == ErrRejectPacket {
				return nil
			}

			if s.Events.OnError != nil {
				s.Events.OnError(cl.Info(), err)
			}
		}
	}

	if pk.FixedHeader.Retain {
		s.retainMessage(cl, pk)
	}

	if pk.FixedHeader.Qos > 0 {
		ack := packets.Packet{
			FixedHeader: packets.FixedHeader{
				Type: packets.Puback,
			},
			PacketID: pk.PacketID,
		}

		if pk.FixedHeader.Qos == 2 {
			ack.FixedHeader.Type = packets.Pubrec
		}

		// omit errors in case of broken connection / LWT publish. ack send failures
		// will be handled by in-flight resending on next reconnect.
		s.onError(cl.Info(), s.writeClient(cl, ack))
	}

	// if an OnMessage hook exists, potentially modify the packet.
	if s.Events.OnMessage != nil {
		if pkx, err := s.Events.OnMessage(cl.Info(), events.Packet(pk)); err == nil {
			pk = packets.Packet(pkx)
		}
	}

	// write packet to the byte buffers of any clients with matching topic filters.
	s.publishToSubscribers(pk)

	return nil
}

// retainMessage adds a message to a topic, and if a persistent store is provided,
// adds the message to the store so it can be reloaded if necessary.
func (s *Server) retainMessage(cl events.Clientlike, pk packets.Packet) {
	out := pk.PublishCopy()
	r := s.Topics.RetainMessage(out)
	atomic.AddInt64(&s.System.Retained, r)

	if s.Store != nil {
		id := "ret_" + out.TopicName
		if r == 1 {
			s.onStorage(cl, s.Store.WriteRetained(persistence.Message{
				ID:          id,
				T:           persistence.KRetained,
				FixedHeader: persistence.FixedHeader(out.FixedHeader),
				TopicName:   out.TopicName,
				Payload:     out.Payload,
			}))
		} else {
			s.onStorage(cl, s.Store.DeleteRetained(id))
		}
	}
}

// publishToSubscribers publishes a publish packet to all subscribers with
// matching topic filters.
func (s *Server) publishToSubscribers(pk packets.Packet) {
	for id, qos := range s.Topics.Subscribers(pk.TopicName) {
		if client, ok := s.Clients.Get(id); ok {

			// If the AllowClients value is set, only deliver the packet if the subscribed
			// client exists in the AllowClients value. For use with the OnMessage event hook
			// in cases where you want to publish messages to clients selectively.
			if pk.AllowClients != nil && !utils.InSliceString(pk.AllowClients, id) {
				continue
			}

			out := pk.PublishCopy()
			if qos > out.FixedHeader.Qos { // Inherit higher desired qos values.
				out.FixedHeader.Qos = qos
			}

			if out.FixedHeader.Qos > 0 { // If QoS required, save to inflight index.
				if out.PacketID == 0 {
					out.PacketID = uint16(client.NextPacketID())
				}

				// If a message has a QoS, we need to ensure it is delivered to
				// the client at some point, one way or another. Store the publish
				// packet in the client's inflight queue and attempt to redeliver
				// if an appropriate ack is not received (or if the client is offline).
				sent := time.Now().Unix()
				q := client.Inflight.Set(out.PacketID, clients.InflightMessage{
					Packet:  out,
					Created: time.Now().Unix(),
					Sent:    sent,
				})
				if q {
					atomic.AddInt64(&s.System.Inflight, 1)
				}

				if s.Store != nil {
					s.onStorage(client, s.Store.WriteInflight(persistence.Message{
						ID:          persistentID(client, out),
						T:           persistence.KInflight,
						FixedHeader: persistence.FixedHeader(out.FixedHeader),
						TopicName:   out.TopicName,
						Payload:     out.Payload,
						Sent:        sent,
					}))
				}
			}

			s.onError(client.Info(), s.writeClient(client, out))
		}
	}
}

// processPuback processes a Puback packet.
func (s *Server) processPuback(cl *clients.Client, pk packets.Packet) error {
	q := cl.Inflight.Delete(pk.PacketID)
	if q {
		atomic.AddInt64(&s.System.Inflight, -1)
	}
	if s.Store != nil {
		s.onStorage(cl, s.Store.DeleteInflight(persistentID(cl, pk)))
	}
	return nil
}

// processPubrec processes a Pubrec packet.
func (s *Server) processPubrec(cl *clients.Client, pk packets.Packet) error {
	out := packets.Packet{
		FixedHeader: packets.FixedHeader{
			Type: packets.Pubrel,
			Qos:  1,
		},
		PacketID: pk.PacketID,
	}

	err := s.writeClient(cl, out)
	if err != nil {
		return err
	}

	return nil
}

// processPubrel processes a Pubrel packet.
func (s *Server) processPubrel(cl *clients.Client, pk packets.Packet) error {
	out := packets.Packet{
		FixedHeader: packets.FixedHeader{
			Type: packets.Pubcomp,
		},
		PacketID: pk.PacketID,
	}

	err := s.writeClient(cl, out)
	if err != nil {
		return err
	}
	q := cl.Inflight.Delete(pk.PacketID)
	if q {
		atomic.AddInt64(&s.System.Inflight, -1)
	}

	if s.Store != nil {
		s.onStorage(cl, s.Store.DeleteInflight(persistentID(cl, pk)))
	}

	return nil
}

// processPubcomp processes a Pubcomp packet.
func (s *Server) processPubcomp(cl *clients.Client, pk packets.Packet) error {
	q := cl.Inflight.Delete(pk.PacketID)
	if q {
		atomic.AddInt64(&s.System.Inflight, -1)
	}
	if s.Store != nil {
		s.onStorage(cl, s.Store.DeleteInflight(persistentID(cl, pk)))
	}
	return nil
}

// processSubscribe processes a Subscribe packet.
func (s *Server) processSubscribe(cl *clients.Client, pk packets.Packet) error {
	retCodes := make([]byte, len(pk.Topics))
	for i := 0; i < len(pk.Topics); i++ {
		if !cl.AC.ACL(cl.Username, pk.Topics[i], false) {
			retCodes[i] = packets.ErrSubAckNetworkError
		} else {
			r := s.Topics.Subscribe(pk.Topics[i], cl.ID, pk.Qoss[i])
			if r {
				if s.Events.OnSubscribe != nil {
					s.Events.OnSubscribe(pk.Topics[i], cl.Info(), pk.Qoss[i])
				}
				atomic.AddInt64(&s.System.Subscriptions, 1)
			}
			cl.NoteSubscription(pk.Topics[i], pk.Qoss[i])
			retCodes[i] = pk.Qoss[i]

			if s.Store != nil {
				s.onStorage(cl, s.Store.WriteSubscription(persistence.Subscription{
					ID:     "sub_" + cl.ID + ":" + pk.Topics[i],
					T:      persistence.KSubscription,
					Filter: pk.Topics[i],
					Client: cl.ID,
					QoS:    pk.Qoss[i],
				}))
			}
		}
	}

	err := s.writeClient(cl, packets.Packet{
		FixedHeader: packets.FixedHeader{
			Type: packets.Suback,
		},
		PacketID:    pk.PacketID,
		ReturnCodes: retCodes,
	})
	if err != nil {
		return err
	}

	// Publish out any retained messages matching the subscription filter and the user has
	// been allowed to subscribe to.
	for i := 0; i < len(pk.Topics); i++ {
		if retCodes[i] == packets.ErrSubAckNetworkError {
			continue
		}

		for _, pkv := range s.Topics.Messages(pk.Topics[i]) {
			s.onError(cl.Info(), s.writeClient(cl, pkv))
		}
	}

	return nil
}

// processUnsubscribe processes an unsubscribe packet.
func (s *Server) processUnsubscribe(cl *clients.Client, pk packets.Packet) error {
	for i := 0; i < len(pk.Topics); i++ {
		q := s.Topics.Unsubscribe(pk.Topics[i], cl.ID)
		if q {
			if s.Events.OnUnsubscribe != nil {
				s.Events.OnUnsubscribe(pk.Topics[i], cl.Info())
			}
			atomic.AddInt64(&s.System.Subscriptions, -1)
		}
		cl.ForgetSubscription(pk.Topics[i])
	}

	err := s.writeClient(cl, packets.Packet{
		FixedHeader: packets.FixedHeader{
			Type: packets.Unsuback,
		},
		PacketID: pk.PacketID,
	})
	if err != nil {
		return err
	}

	return nil
}

// atomicItoa reads an *int64 and formats a decimal string.
func atomicItoa(ptr *int64) string {
	return strconv.FormatInt(atomic.LoadInt64(ptr), 10)
}

// persistentID return a string combining the client and packet
// identifiers for use with the persistence layer.
func persistentID(client *clients.Client, pk packets.Packet) string {
	return "if_" + client.ID + "_" + pk.FormatID()
}

// publishSysTopics publishes the current values to the server $SYS topics.
// Due to the int to string conversions this method is not as cheap as
// some of the others so the publishing interval should be set appropriately.
func (s *Server) publishSysTopics() {
	pk := packets.Packet{
		FixedHeader: packets.FixedHeader{
			Type:   packets.Publish,
			Retain: true,
		},
	}

	uptime := time.Now().Unix() - atomic.LoadInt64(&s.System.Started)
	atomic.StoreInt64(&s.System.Uptime, uptime)
	topics := map[string]string{
		"$SYS/broker/version":                   s.System.Version,
		"$SYS/broker/uptime":                    atomicItoa(&s.System.Uptime),
		"$SYS/broker/timestamp":                 atomicItoa(&s.System.Started),
		"$SYS/broker/load/bytes/received":       atomicItoa(&s.System.BytesRecv),
		"$SYS/broker/load/bytes/sent":           atomicItoa(&s.System.BytesSent),
		"$SYS/broker/clients/connected":         atomicItoa(&s.System.ClientsConnected),
		"$SYS/broker/clients/disconnected":      atomicItoa(&s.System.ClientsDisconnected),
		"$SYS/broker/clients/maximum":           atomicItoa(&s.System.ClientsMax),
		"$SYS/broker/clients/total":             atomicItoa(&s.System.ClientsTotal),
		"$SYS/broker/connections/total":         atomicItoa(&s.System.ConnectionsTotal),
		"$SYS/broker/messages/received":         atomicItoa(&s.System.MessagesRecv),
		"$SYS/broker/messages/sent":             atomicItoa(&s.System.MessagesSent),
		"$SYS/broker/messages/publish/dropped":  atomicItoa(&s.System.PublishDropped),
		"$SYS/broker/messages/publish/received": atomicItoa(&s.System.PublishRecv),
		"$SYS/broker/messages/publish/sent":     atomicItoa(&s.System.PublishSent),
		"$SYS/broker/messages/retained/count":   atomicItoa(&s.System.Retained),
		"$SYS/broker/messages/inflight":         atomicItoa(&s.System.Inflight),
		"$SYS/broker/subscriptions/count":       atomicItoa(&s.System.Subscriptions),
	}

	for topic, payload := range topics {
		pk.TopicName = topic
		pk.Payload = []byte(payload)
		q := s.Topics.RetainMessage(pk.PublishCopy())
		atomic.AddInt64(&s.System.Retained, q)
		s.publishToSubscribers(pk)
	}

	if s.Store != nil {
		s.onStorage(&s.inline, s.Store.WriteServerInfo(persistence.ServerInfo{
			Info: *s.System,
			ID:   persistence.KServerInfo,
		}))
	}
}

// ResendClientInflight attempts to resend all undelivered inflight messages
// to a client.
func (s *Server) ResendClientInflight(cl *clients.Client, force bool) error {
	if atomic.LoadUint32(&cl.State.Done) == 1 || cl.Inflight.Len() == 0 {
		return nil
	}

	nt := time.Now().Unix()
	for _, tk := range cl.Inflight.GetAll() {
		if tk.Resends >= inflightMaxResends { // After a reasonable time, drop inflight packets.
			cl.Inflight.Delete(tk.Packet.PacketID)
			if tk.Packet.FixedHeader.Type == packets.Publish {
				atomic.AddInt64(&s.System.PublishDropped, 1)
			}

			if s.Store != nil {
				s.onStorage(cl, s.Store.DeleteInflight(persistentID(cl, tk.Packet)))
			}

			continue
		}

		// Only continue if the resend backoff time has passed and there's a backoff time.
		if !force && (nt-tk.Sent < inflightResendBackoff[tk.Resends] || len(inflightResendBackoff) < tk.Resends) {
			continue
		}

		if tk.Packet.FixedHeader.Type == packets.Publish {
			tk.Packet.FixedHeader.Dup = true
		}

		tk.Resends++
		tk.Sent = nt
		cl.Inflight.Set(tk.Packet.PacketID, tk)
		_, err := cl.WritePacket(tk.Packet)
		if err != nil {
			return err
		}

		if s.Store != nil {
			s.onStorage(cl, s.Store.WriteInflight(persistence.Message{
				ID:          persistentID(cl, tk.Packet),
				T:           persistence.KInflight,
				FixedHeader: persistence.FixedHeader(tk.Packet.FixedHeader),
				TopicName:   tk.Packet.TopicName,
				Payload:     tk.Packet.Payload,
				Sent:        tk.Sent,
				Resends:     tk.Resends,
			}))
		}
	}

	return nil
}

// Close attempts to gracefully shutdown the server, all listeners, clients, and stores.
func (s *Server) Close() error {
	close(s.done)
	s.Listeners.CloseAll(s.closeListenerClients)

	if s.Store != nil {
		s.Store.Close()
	}

	return nil
}

// closeListenerClients closes all clients on the specified listener.
func (s *Server) closeListenerClients(listener string) {
	clients := s.Clients.GetByListener(listener)
	for _, cl := range clients {
		cl.Stop(ErrServerShutdown)
	}
}

// sendLWT issues an LWT message to a topic when a client disconnects.
func (s *Server) sendLWT(cl *clients.Client) error {
	if cl.LWT.Topic != "" {
		err := s.processPublish(cl, packets.Packet{
			FixedHeader: packets.FixedHeader{
				Type:   packets.Publish,
				Retain: cl.LWT.Retain,
				Qos:    cl.LWT.Qos,
			},
			TopicName: cl.LWT.Topic,
			Payload:   cl.LWT.Message,
		})
		if err != nil {
			return s.onError(cl.Info(), fmt.Errorf("send lwt: %s %w; %+v", cl.ID, err, cl.LWT))
		}
	}

	return nil
}

// readStore reads in any data from the persistent datastore (if applicable).
func (s *Server) readStore() error {
	info, err := s.Store.ReadServerInfo()
	if err != nil {
		return fmt.Errorf("load server info; %w", err)
	}
	s.loadServerInfo(info)

	clients, err := s.Store.ReadClients()
	if err != nil {
		return fmt.Errorf("load clients; %w", err)
	}
	s.loadClients(clients)

	subs, err := s.Store.ReadSubscriptions()
	if err != nil {
		return fmt.Errorf("load subscriptions; %w", err)
	}
	s.loadSubscriptions(subs)

	inflight, err := s.Store.ReadInflight()
	if err != nil {
		return fmt.Errorf("load inflight; %w", err)
	}
	s.loadInflight(inflight)

	retained, err := s.Store.ReadRetained()
	if err != nil {
		return fmt.Errorf("load retained; %w", err)
	}
	s.loadRetained(retained)

	return nil
}

// loadServerInfo restores server info from the datastore.
func (s *Server) loadServerInfo(v persistence.ServerInfo) {
	version := s.System.Version
	s.System = &v.Info
	s.System.Version = version
}

// loadSubscriptions restores subscriptions from the datastore.
func (s *Server) loadSubscriptions(v []persistence.Subscription) {
	for _, sub := range v {
		if s.Topics.Subscribe(sub.Filter, sub.Client, sub.QoS) {
			if cl, ok := s.Clients.Get(sub.Client); ok {
				cl.NoteSubscription(sub.Filter, sub.QoS)
				if s.Events.OnSubscribe != nil {
					s.Events.OnSubscribe(sub.Filter, cl.Info(), sub.QoS)
				}
			}
		}
	}
}

// loadClients restores clients from the datastore.
func (s *Server) loadClients(v []persistence.Client) {
	for _, c := range v {
		cl := clients.NewClientStub(s.System)
		cl.ID = c.ClientID
		cl.Listener = c.Listener
		cl.Username = c.Username
		cl.LWT = clients.LWT(c.LWT)
		s.Clients.Add(cl)
	}
}

// loadInflight restores inflight messages from the datastore.
func (s *Server) loadInflight(v []persistence.Message) {
	for _, msg := range v {
		if client, ok := s.Clients.Get(msg.Client); ok {
			client.Inflight.Set(msg.PacketID, clients.InflightMessage{
				Packet: packets.Packet{
					FixedHeader: packets.FixedHeader(msg.FixedHeader),
					PacketID:    msg.PacketID,
					TopicName:   msg.TopicName,
					Payload:     msg.Payload,
				},
				Created: msg.Created,
				Sent:    msg.Sent,
				Resends: msg.Resends,
			})
		}
	}
}

// loadRetained restores retained messages from the datastore.
func (s *Server) loadRetained(v []persistence.Message) {
	for _, msg := range v {
		s.Topics.RetainMessage(packets.Packet{
			FixedHeader: packets.FixedHeader(msg.FixedHeader),
			TopicName:   msg.TopicName,
			Payload:     msg.Payload,
		})
	}
}

// clearExpiredInflights deletes all inflight messages older than server inflight TTL.
func (s *Server) clearExpiredInflights(dt int64) {
	expiry := dt - s.Options.InflightTTL

	//便利全部客户端删除过期消息
	for _, client := range s.Clients.GetAll() {
		deleted := client.Inflight.ClearExpired(expiry)
		atomic.AddInt64(&s.System.Inflight, deleted*-1)
	}

	if s.Store != nil {
		//删除过期消息
		s.Store.ClearExpiredInflight(expiry)
	}
}

// clearAbandonedInflights deletes all inflight messages for a disconnected user (eg. with a clean session).
func (s *Server) clearAbandonedInflights(cl *clients.Client) {
	for i := range cl.Inflight.GetAll() {
		cl.Inflight.Delete(i)
		atomic.AddInt64(&s.System.Inflight, -1)
	}
}

// resendPendingInflights attempts resends of any pending and due inflight messages.
func (s *Server) resendPendingInflights() {
	for _, client := range s.Clients.GetAll() {
		err := s.ResendClientInflight(client, false)
		if err != nil {
			// TODO add log-level debugging.
			continue
		}
	}
}