func (this *snode) matchQos(qos byte, subs *[]interface{}, qoss *[]byte) {
	for i, sub := range this.subs {
		if qos <= this.qos[i] {
			*subs = append(*subs, sub)
			*qoss = append(*qoss, qos)
		}
	}
}
The QoS of the payload messages sent in response to a subscription must be the
minimum of the QoS of the originally published message (in this case, it's the
qos parameter) and the maximum QoS granted by the server (in this case, it's
the QoS in the topic tree).

It's also possible that even if the topic matches, the subscriber is not included
due to the QoS granted is lower than the published message QoS. For example,
if the client is granted only QoS 0, and the publish message is QoS 1, then this
client is not to be send the published message.
package topics
import (
	"fmt"
	"reflect"
	"sync"
	"github.com/surgemq/message"
)
var (
	MaxQosAllowed = message.QosExactlyOnce
)
var _ TopicsProvider = (*memTopics)(nil)
type memTopics struct {
	smu sync.RWMutex
	sroot *snode
	rmu sync.RWMutex
	rroot *rnode
}
func init() {
	Register("mem", NewMemProvider())
}
var _ TopicsProvider = (*memTopics)(nil)
func NewMemProvider() *memTopics {
	return &memTopics{
		sroot: newSNode(),
		rroot: newRNode(),
	}
}
func (this *memTopics) Subscribe(topic []byte, qos byte, sub interface{}) (byte, error) {
	if !message.ValidQos(qos) {
		return message.QosFailure, fmt.Errorf("Invalid QoS %d", qos)
	}
	if sub == nil {
		return message.QosFailure, fmt.Errorf("Subscriber cannot be nil")
	}
	this.smu.Lock()
	defer this.smu.Unlock()
	if qos > MaxQosAllowed {
		qos = MaxQosAllowed
	}
	if err := this.sroot.sinsert(topic, qos, sub); err != nil {
		return message.QosFailure, err
	}
	return qos, nil
}
func (this *memTopics) Unsubscribe(topic []byte, sub interface{}) error {
	this.smu.Lock()
	defer this.smu.Unlock()
	return this.sroot.sremove(topic, sub)
}
func (this *memTopics) Subscribers(topic []byte, qos byte, subs *[]interface{}, qoss *[]byte) error {
	if !message.ValidQos(qos) {
		return fmt.Errorf("Invalid QoS %d", qos)
	}
	this.smu.RLock()
	defer this.smu.RUnlock()
	*subs = (*subs)[0:0]
	*qoss = (*qoss)[0:0]
	return this.sroot.smatch(topic, qos, subs, qoss)
}
func (this *memTopics) Retain(msg *message.PublishMessage) error {
	this.rmu.Lock()
	defer this.rmu.Unlock()
	if len(msg.Payload()) == 0 {
		return this.rroot.rremove(msg.Topic())
	}
	return this.rroot.rinsert(msg.Topic(), msg)
}
func (this *memTopics) Retained(topic []byte, msgs *[]*message.PublishMessage) error {
	this.rmu.RLock()
	defer this.rmu.RUnlock()
	return this.rroot.rmatch(topic, msgs)
}
func (this *memTopics) Close() error {
	this.sroot = nil
	this.rroot = nil
	return nil
}
type snode struct {
	subs []interface{}
	qos  []byte
	snodes map[string]*snode
}
func newSNode() *snode {
	return &snode{
		snodes: make(map[string]*snode),
	}
}
func (this *snode) sinsert(topic []byte, qos byte, sub interface{}) error {
	if len(topic) == 0 {
		for i := range this.subs {
			if equal(this.subs[i], sub) {
				this.qos[i] = qos
				return nil
			}
		}
		this.subs = append(this.subs, sub)
		this.qos = append(this.qos, qos)
		return nil
	}
	ntl, rem, err := nextTopicLevel(topic)
	if err != nil {
		return err
	}
	level := string(ntl)
	n, ok := this.snodes[level]
	if !ok {
		n = newSNode()
		this.snodes[level] = n
	}
	return n.sinsert(rem, qos, sub)
}
func (this *snode) sremove(topic []byte, sub interface{}) error {
	if len(topic) == 0 {
		if sub == nil {
			this.subs = this.subs[0:0]
			this.qos = this.qos[0:0]
			return nil
		}
		for i := range this.subs {
			if equal(this.subs[i], sub) {
				this.subs = append(this.subs[:i], this.subs[i+1:]...)
				this.qos = append(this.qos[:i], this.qos[i+1:]...)
				return nil
			}
		}
		return fmt.Errorf("memtopics/remove: No topic found for subscriber")
	}
	ntl, rem, err := nextTopicLevel(topic)
	if err != nil {
		return err
	}
	level := string(ntl)
	n, ok := this.snodes[level]
	if !ok {
		return fmt.Errorf("memtopics/remove: No topic found")
	}
	if err := n.sremove(rem, sub); err != nil {
		return err
	}
	if len(n.subs) == 0 && len(n.snodes) == 0 {
		delete(this.snodes, level)
	}
	return nil
}
func (this *snode) smatch(topic []byte, qos byte, subs *[]interface{}, qoss *[]byte) error {
	if len(topic) == 0 {
		this.matchQos(qos, subs, qoss)
		return nil
	}
	ntl, rem, err := nextTopicLevel(topic)
	if err != nil {
		return err
	}
	level := string(ntl)
	for k, n := range this.snodes {
		if k == MWC {
			n.matchQos(qos, subs, qoss)
		} else if k == SWC || k == level {
			if err := n.smatch(rem, qos, subs, qoss); err != nil {
				return err
			}
		}
	}
	return nil
}
type rnode struct {
	msg *message.PublishMessage
	buf []byte
	rnodes map[string]*rnode
}
func newRNode() *rnode {
	return &rnode{
		rnodes: make(map[string]*rnode),
	}
}
func (this *rnode) rinsert(topic []byte, msg *message.PublishMessage) error {
	if len(topic) == 0 {
		l := msg.Len()
		if l > cap(this.buf) {
			this.buf = make([]byte, l)
		} else {
			this.buf = this.buf[0:l]
		}
		if _, err := msg.Encode(this.buf); err != nil {
			return err
		}
		if this.msg == nil {
			this.msg = message.NewPublishMessage()
		}
		if _, err := this.msg.Decode(this.buf); err != nil {
			return err
		}
		return nil
	}
	ntl, rem, err := nextTopicLevel(topic)
	if err != nil {
		return err
	}
	level := string(ntl)
	n, ok := this.rnodes[level]
	if !ok {
		n = newRNode()
		this.rnodes[level] = n
	}
	return n.rinsert(rem, msg)
}
func (this *rnode) rremove(topic []byte) error {
	if len(topic) == 0 {
		this.buf = nil
		this.msg = nil
		return nil
	}
	ntl, rem, err := nextTopicLevel(topic)
	if err != nil {
		return err
	}
	level := string(ntl)
	n, ok := this.rnodes[level]
	if !ok {
		return fmt.Errorf("memtopics/rremove: No topic found")
	}
	if err := n.rremove(rem); err != nil {
		return err
	}
	if len(n.rnodes) == 0 {
		delete(this.rnodes, level)
	}
	return nil
}
func (this *rnode) rmatch(topic []byte, msgs *[]*message.PublishMessage) error {
	if len(topic) == 0 {
		if this.msg != nil {
			*msgs = append(*msgs, this.msg)
		}
		return nil
	}
	ntl, rem, err := nextTopicLevel(topic)
	if err != nil {
		return err
	}
	level := string(ntl)
	if level == MWC {
		this.allRetained(msgs)
	} else if level == SWC {
		for _, n := range this.rnodes {
			if err := n.rmatch(rem, msgs); err != nil {
				return err
			}
		}
	} else {
		if n, ok := this.rnodes[level]; ok {
			if err := n.rmatch(rem, msgs); err != nil {
				return err
			}
		}
	}
	return nil
}
func (this *rnode) allRetained(msgs *[]*message.PublishMessage) {
	if this.msg != nil {
		*msgs = append(*msgs, this.msg)
	}
	for _, n := range this.rnodes {
		n.allRetained(msgs)
	}
}
const (
	stateCHR byte = iota  
	stateMWC              
	stateSWC              
	stateSEP              
	stateSYS              
)
func nextTopicLevel(topic []byte) ([]byte, []byte, error) {
	s := stateCHR
	for i, c := range topic {
		switch c {
		case '/':
			if s == stateMWC {
				return nil, nil, fmt.Errorf("memtopics/nextTopicLevel: Multi-level wildcard found in topic and it's not at the last level")
			}
			if i == 0 {
				return []byte(SWC), topic[i+1:], nil
			}
			return topic[:i], topic[i+1:], nil
		case '#':
			if i != 0 {
				return nil, nil, fmt.Errorf("memtopics/nextTopicLevel: Wildcard character '#' must occupy entire topic level")
			}
			s = stateMWC
		case '+':
			if i != 0 {
				return nil, nil, fmt.Errorf("memtopics/nextTopicLevel: Wildcard character '+' must occupy entire topic level")
			}
			s = stateSWC
		case '$':
			if i == 0 {
				return nil, nil, fmt.Errorf("memtopics/nextTopicLevel: Cannot publish to $ topics")
			}
			s = stateSYS
		default:
			if s == stateMWC || s == stateSWC {
				return nil, nil, fmt.Errorf("memtopics/nextTopicLevel: Wildcard characters '#' and '+' must occupy entire topic level")
			}
			s = stateCHR
		}
	}
	return topic, nil, nil
}