dsbqj_admin/app/task/subscribe.go

package task

import (
	"dsbqj-admin/pkg/helper/wechat"
	"dsbqj-admin/pkg/logger"
	"dsbqj-admin/pkg/sender"
	"dsbqj-admin/pkg/util"
	"sync"
)

var SubscribeTask *Subscribe

type PushItem struct {
	Key       string // 玩家设备id
	DeviceId  string
	Module    string // 推送类型
	Timestamp int64  // 时间戳
}

type Subscribe struct {
	sync.Mutex
	waiting  map[string]*PushItem
	wxHelper *wechat.WechatHelper
	sender   *sender.SubscribeSender
}

func SubscribeInit() *Subscribe {
	SubscribeTask = new(Subscribe)
	SubscribeTask.waiting = make(map[string]*PushItem)
	SubscribeTask.sender = sender.NewSubscribeSender(10)
	SubscribeTask.sender.Start()
	return SubscribeTask
}

func (this *Subscribe) Exec() {
	now := util.GetNowSecond()

	var needPush []*PushItem

	this.Lock()
	for key, item := range this.waiting {
		if item.Timestamp <= now {
			needPush = append(needPush, item)
			delete(this.waiting, key)
		}
	}
	this.Unlock()

	for _, item := range needPush {
		logger.Info("Subscribe exec neePush module %s, device id %s", item.Module, item.DeviceId)
		this.sender.Send(&sender.SubscribeSend{
			DeviceId: item.DeviceId,
			Module:   item.Module,
		}) // 触发后移除，防止重复
	}
}

func (this *Subscribe) Append(key string, deviceId string, timestamp int64, module string) {
	this.Lock()
	defer this.Unlock()

	newPushItem := &PushItem{key, deviceId, module, timestamp}
	this.waiting[key] = newPushItem
}

func (this *Subscribe) Remove(key string) {
	this.Lock()
	defer this.Unlock()

	delete(this.waiting, key)
}

func (this *Subscribe) Send(openIds []string, module string) {
	this.sender.Send(&sender.SubscribeSend{
		OpenIds: openIds,
		Module:  module,
	})
}