GtDataStore/app/cmd/events/internal/logic/handler/event.go

package handler

import (
	"GtDataStore/app/cmd/events/internal/svc"
	"GtDataStore/app/model"
	"context"
	"encoding/json"
	"fmt"
	"github.com/go-redis/redis/v8"
	"github.com/zeromicro/go-zero/core/logx"
	"time"
)

func HandlerEventNotify(svcCtx *svc.ServiceContext) error {
	defer close(eventChan)
	fmt.Print("start HandlerEventNotify.....\n")
	queue := svcCtx.Cache

	for info := range eventChan {
		info.MsgTime = time.Now()
		// 数据通知到redis
		toMQ(queue, info)
		// 事件记录到mysql
		toDB(svcCtx, info)
	}

	return nil
}

func toMQ(queue *redis.Client, info eventInfo) {
	channel := fmt.Sprintf(EVTNE_NOTIFY_KEY_FORMAT, info.ProjectId, info.EventName)
	if bs, err := json.Marshal(info); err == nil {
		//queue.XAdd(context.Background(), &redis.XAddArgs{
		//	Stream: channel,
		//	MaxLen: 5000,
		//	Values: ``,
		//})

		if intCmd := queue.Publish(context.Background(), channel, string(bs)); intCmd == nil {
			logx.Errorf("handlerEvent queue.Publish toMQ error: %s %+v", channel, info)
		}
	} else {
		logx.Errorf("handlerEvent queue.Publish toMQ error: %s %+v", channel, info)
	}

}

func toDB(svcCtx *svc.ServiceContext, info eventInfo) {
	if _, err := svcCtx.DcEventList.Insert(context.Background(), &model.DcEventList{
		ProjectId:  info.ProjectId,
		DeviceCode: info.DeviceCode,
		Name:       info.EventName,
		Item:       info.Item,
		Val:        info.NewValue,
		OldVal:     info.OldValue,
		Time:       info.Time,
		EventId:    info.EventId,
		CTime:      info.MsgTime,
	}); err != nil {
		logx.Errorf("handlerEvent queue.Publish toDB error:%+v", info)
	}
}