game2006go/server/mqproxy/service/mqmgr.go

package service

import (
	"f5"
	"q5"
	"sync"
	"sync/atomic"
	"time"
)

type mqManager struct {
	topicMap     q5.ConcurrentMap[string, *topicCache]
	lock         *sync.Mutex
	publishcount int32
	consumecount int32
	expirecount  int32
}

type topicCache struct {
	lock       *sync.Mutex
	channelMap q5.ConcurrentMap[string, int64]
	msgList    []*msgItem
}

func (this *topicCache) Lock() {
	this.lock.Lock()
}

func (this *topicCache) Unlock() {
	this.lock.Unlock()
}

type msgItem struct {
	channels *q5.ConcurrentMap[string, int64]
	content  string
	addTime  int64
	expire   int64
}

func (this *mqManager) init() {
	this.topicMap = q5.ConcurrentMap[string, *topicCache]{}
	this.lock = new(sync.Mutex)
	go this.outputMonitorLog()
}

func (this *mqManager) unInit() {
}

func (this *mqManager) Lock() {
	this.lock.Lock()
}

func (this *mqManager) Unlock() {
	this.lock.Unlock()
}

func (this *mqManager) PublishTopic(topic, message string, life int32) {
	this.Lock()
	t, exist := this.topicMap.Load(topic)
	if !exist {
		tc := new(topicCache)
		tc.lock = new(sync.Mutex)
		tc.channelMap = q5.ConcurrentMap[string, int64]{}
		tc.msgList = make([]*msgItem, 0)
		this.topicMap.Store(topic, tc)
		t = &tc
	}
	newitem := new(msgItem)
	newitem.addTime = f5.GetApp().GetRealSeconds()
	newitem.content = message
	newitem.expire = newitem.addTime + int64(life)
	if (*t).channelMap.GetSize() > 0 {
		newitem.channels = new(q5.ConcurrentMap[string, int64])
		(*t).channelMap.Range(func(key string, value int64) bool {
			newitem.channels.Store(key, value)
			return true
		})
	}
	this.Unlock()

	(*t).Lock()
	(*t).msgList = append((*t).msgList, newitem)
	(*t).Unlock()
	this.IncPublishTimes()
}

func (this *mqManager) SubscribeChannel(topic, channel string) {
	this.Lock()
	t, exist := this.topicMap.Load(topic)
	if !exist {
		tc := new(topicCache)
		tc.lock = new(sync.Mutex)
		tc.msgList = make([]*msgItem, 0)
		this.topicMap.Store(topic, tc)
		t = &tc
	}
	(*t).channelMap.Store(channel, 0)
	this.Unlock()
}

func (this *mqManager) UnsubscribeChannel(topic, channel string) {
	this.Lock()
	t, exist := this.topicMap.Load(topic)
	if exist {
		(*t).channelMap.Delete(channel)
	}
	this.Unlock()
}

func (this *mqManager) ConsumeTopic(topic, channel string) (msg string) {
	t, exist := this.topicMap.Load(topic)
	if !exist {
		return
	}

	(*t).Lock()
	defer (*t).Unlock()

	if len((*t).msgList) == 0 {
		return
	}

	msgidx := -1
	del := false
	for index := range (*t).msgList {
		if (*t).msgList[index].channels == nil {
			msgidx = index
			del = true
		} else {
			_, exist := (*t).msgList[index].channels.LoadAndDelete(channel)
			if exist {
				msgidx = index
				del = (*t).msgList[index].channels.GetSize() == 0
			}
		}
	}

	if msgidx > -1 {
		msg = (*t).msgList[msgidx].content
		if del {
			if msgidx+1 < len((*t).msgList) {
				(*t).msgList = append((*t).msgList[:msgidx], (*t).msgList[msgidx+1:]...)
			} else {
				(*t).msgList = (*t).msgList[:msgidx]
			}
		}
	}

	this.IncConsumeTimes()
	return
}

func (this *mqManager) IncPublishTimes() {
	atomic.AddInt32(&this.publishcount, 1)
}

func (this *mqManager) IncConsumeTimes() {
	atomic.AddInt32(&this.consumecount, 1)
}

func (this *mqManager) IncExpireTimes() {
	atomic.AddInt32(&this.expirecount, 1)

}

func (this *mqManager) outputMonitorLog() {
	logtimes := 0
	for {
		f5.GetSysLog().Info("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<")
		f5.GetSysLog().Info("publishcount:%d, consumecount:%d, expirecount:%d",
			this.publishcount,
			this.consumecount,
			this.expirecount)
		f5.GetSysLog().Info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")

		logtimes++
		if logtimes > 6 {
			logtimes = 0
			this.publishcount = 0
			this.consumecount = 0
			this.expirecount = 0

			this.Lock()
			now := f5.GetApp().GetRealSeconds()
			this.topicMap.Range(func(key string, value *topicCache) bool {
				value.Lock()
				defer value.Unlock()
				for i := 0; i < len(value.msgList); {
					if value.msgList[i].expire < now {
						value.msgList = append(value.msgList[:i], value.msgList[i+1:]...)
						this.IncExpireTimes()
					} else {
						i++
					}
				}
				return true
			})
			this.Unlock()
		}
		time.Sleep(time.Second * 10)
	}
}