dsbqj_admin/pkg/sender/subscribe.go

package sender

import (
	"dsbqj-admin/model/mongo/subscribe"
	"dsbqj-admin/pkg/helper/wechat"
	"dsbqj-admin/pkg/logger"
	"fmt"
	"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)
	default:
		log.Printf("[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["thing12"] = make(map[string]string)
	msg["thing12"]["value"] = "天气不错"
	msg["thing13"] = make(map[string]string)
	msg["thing13"]["value"] = "明天下雨"

	fmt.Println(this.wxHelper.SendWechatSubscribe(subscribeOne.OpenId, "UeiRSO8hHAFvTodH3SlWb_KoqT3Z2F2MoSvsDkZv7AE", 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["thing12"] = make(map[string]string)
	msg["thing12"]["value"] = "天气不错"
	msg["thing13"] = make(map[string]string)
	msg["thing13"]["value"] = "明天下雨"

	fmt.Println(this.wxHelper.SendWechatSubscribe(subscribeOne.OpenId, "UeiRSO8hHAFvTodH3SlWb_KoqT3Z2F2MoSvsDkZv7AE", msg))
}

func (this *SubscribeSender) SendGuildGameSubscribe(openIds []string) {

}