dsbqj_admin/pkg/sender/subscribe.go

package sender

import (
	"dsbqj-admin/model/mongo/subscribe"
	"dsbqj-admin/pkg/helper/wechat"
	"dsbqj-admin/pkg/logger"
	"github.com/kamva/mgm/v3"
	"go.mongodb.org/mongo-driver/bson"
	"log"
)

type SubscribeSend struct {
	DeviceId string
	OpenIds  []string
	Module   string
}

type SubscribeSender struct {
	workerNum int
	queue     chan *SubscribeSend
	wxHelper  *wechat.WechatHelper
}

func NewSubscribeSender(workerNum int) *SubscribeSender {
	return &SubscribeSender{
		workerNum: workerNum,
		queue:     make(chan *SubscribeSend, 1000),
		wxHelper:  wechat.NewWechatHelper(),
	}
}

func (this *SubscribeSender) Send(send *SubscribeSend) {
	select {
	case this.queue <- send:
	default:
		logger.Info("subscribe sender queue full, drop task=%+v", send)
	}
}

func (this *SubscribeSender) Start() {
	for i := 0; i < this.workerNum; i++ {
		go this.worker(i)
	}
}

func (this *SubscribeSender) worker(id int) {
	for task := range this.queue {
		this.safeSend(task)
	}
}

func (this *SubscribeSender) safeSend(send *SubscribeSend) {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("send subscribe panic: %v", r)
		}
	}()

	switch send.Module {
	case "hangup":
		this.SendHangupSubscribe(send.DeviceId)
	case "autofight":
		this.SendHangupSubscribe(send.DeviceId)
	case "guildgame":
		this.SendGuildGameSubscribe(send.OpenIds)
	case "alliance":
		this.SendAllianceSubscribe(send.OpenIds)
	case "warheavens":
		this.SendWarHeavensSubscribe(send.OpenIds)
	default:
		logger.Info("[push] unknown type=%s, openId=%s",
			send.Module, send.DeviceId)
	}
}

func (this *SubscribeSender) SendHangupSubscribe(deviceId string) {
	subscribeOne := new(subscribe.Subscribe)
	err := mgm.Coll(&subscribe.Subscribe{}).First(bson.M{"device_id": deviceId, "modules.hangup.enabled": true}, subscribeOne)
	if err != nil {
		return
	}

	msg := make(map[string]map[string]string)
	msg["thing1"] = make(map[string]string)
	msg["thing1"]["value"] = "挂机奖励时长已满"
	msg["thing3"] = make(map[string]string)
	msg["thing3"]["value"] = "您的挂机奖励时长已满，请打开游戏领取"

	this.wxHelper.SendWechatSubscribe(subscribeOne.OpenId, "0ByvFSV3B3U5-1XyvruRe7AMHAxiwQBFUL50lmoatCY", msg)
}

func (this *SubscribeSender) SendAutoFightSubscribe(deviceId string) {
	subscribeOne := new(subscribe.Subscribe)
	err := mgm.Coll(&subscribe.Subscribe{}).First(bson.M{"device_id": deviceId, "modules.hangup.enabled": true}, subscribeOne)
	if err != nil {
		return
	}
	msg := make(map[string]map[string]string)
	msg["thing2"] = make(map[string]string)
	msg["thing2"]["value"] = "离线闯关即将结束，请及时收取您的离线闯关奖励。"
	msg["thing1"] = make(map[string]string)
	msg["thing1"]["value"] = "离线闯关提醒"

	this.wxHelper.SendWechatSubscribe(subscribeOne.OpenId, "5rwMjLs6nVQ0SMtS0EAPEpmXZduOdEgSWbIjSNG5T7M", msg)
}

func (this *SubscribeSender) SendGuildGameSubscribe(openIds []string) {
	msg := make(map[string]map[string]string)
	msg["thing2"] = make(map[string]string)
	msg["thing2"]["value"] = "门派攻防战即将开始，为门派荣誉而战吧。"
	msg["thing4"] = make(map[string]string)
	msg["thing4"]["value"] = "门派攻防战"

	for _, openId := range openIds {
		this.wxHelper.SendWechatSubscribe(openId, "e6D_r9_5tKNZK0cvM8fjzMXy9z1-aoNYQ9TSvAEuklU", msg)
	}
}

func (this *SubscribeSender) SendWarHeavensSubscribe(openIds []string) {
	msg := make(map[string]map[string]string)
	msg["thing2"] = make(map[string]string)
	msg["thing2"]["value"] = "三界战场即将开始，争夺人间道统。"
	msg["thing4"] = make(map[string]string)
	msg["thing4"]["value"] = "三界争峰"

	for _, openId := range openIds {
		this.wxHelper.SendWechatSubscribe(openId, "e6D_r9_5tKNZK0cvM8fjzMXy9z1-aoNYQ9TSvAEuklU", msg)
	}
}

func (this *SubscribeSender) SendAllianceSubscribe(openIds []string) {
	msg := make(map[string]map[string]string)
	msg["thing2"] = make(map[string]string)
	msg["thing2"]["value"] = "独战诸仙即将开始！"
	msg["thing4"] = make(map[string]string)
	msg["thing4"]["value"] = "决战诸仙"

	for _, openId := range openIds {
		this.wxHelper.SendWechatSubscribe(openId, "e6D_r9_5tKNZK0cvM8fjzMXy9z1-aoNYQ9TSvAEuklU", msg)
	}
}