api_binding/mqtt_binding/mqtt_client/router/router.go

package router

import (
	"errors"
	"git.sxidc.com/go-tools/api_binding/utils"
	"sync"
)

type Handler func(item *Item, data []byte)
type OnAddItemFunc func(item Item) error

type Router struct {
	Group string

	itemsMutex    *sync.Mutex
	items         []Item
	handlers      []Handler
	onAddItemFunc OnAddItemFunc
}

func NewRouter(group string, handlers []Handler, onAddItemFunc OnAddItemFunc) *Router {
	return &Router{
		Group:         group,
		itemsMutex:    &sync.Mutex{},
		items:         make([]Item, 0),
		onAddItemFunc: onAddItemFunc,
		handlers:      handlers,
	}
}

func DestroyRouter(router *Router) {
	if router != nil {
		return
	}

	router.itemsMutex.Lock()
	for _, item := range router.items {
		DestroyItem(&item)
	}
	router.items = nil
	router.itemsMutex.Unlock()

	router = nil
}

func (router *Router) AddItem(item *Item) error {
	if item == nil {
		return nil
	}

	router.itemsMutex.Lock()
	defer router.itemsMutex.Unlock()

	item.handlers = append(item.handlers, router.handlers...)

	router.items = append(router.items, *item)

	if router.onAddItemFunc != nil {
		err := router.onAddItemFunc(*item)
		if err != nil {
			return err
		}
	}

	return nil
}

func (router *Router) RangeItem(rangeFunc func(item Item) error) error {
	if rangeFunc == nil {
		return nil
	}

	router.itemsMutex.Lock()
	defer router.itemsMutex.Unlock()

	for _, item := range router.items {
		err := rangeFunc(item)
		if err != nil {
			return err
		}
	}

	return nil
}

type Item struct {
	Topic            string
	Qos              byte
	ResponseRetained bool

	handlers            []Handler
	currentHandlerIndex int
	currentData         []byte
}

func NewItem(topic string, qos byte, responseRetained bool) (*Item, error) {
	if utils.IsStringEmpty(topic) {
		return nil, errors.New("没有传递主题")
	}

	return &Item{
		Topic:               topic,
		Qos:                 qos,
		ResponseRetained:    responseRetained,
		handlers:            make([]Handler, 0),
		currentHandlerIndex: 0,
		currentData:         make([]byte, 0),
	}, nil
}

func DestroyItem(item *Item) {
	if item == nil {
		return
	}

	item = nil
}

func (item *Item) CallHandlers(data []byte) {
	item.currentHandlerIndex = 0
	item.currentData = data
	item.handlers[item.currentHandlerIndex](item, item.currentData)
}

func (item *Item) Next() {
	item.currentHandlerIndex++
	if item.currentHandlerIndex < len(item.handlers) {
		item.handlers[item.currentHandlerIndex](item, item.currentData)
	}
}

func (item *Item) GetData() []byte {
	return item.currentData
}