更新通知系统

nsq/consumer.go → notify/consumer.go

-package nsq
+package notify
 
 import (
 	"github.com/nsqio/go-nsq"

notify/message.go

+package notify
+
+import (
+	"encoding/json"
+	"errors"
+	"github.com/gogf/gf/frame/g"
+	"github.com/gogf/gf/util/gconv"
+	"gitlab.jxhh.com/stbz/library.git/logs"
+	"math/rand"
+	"strconv"
+	"time"
+)
+
+const (
+	//通知服务topic
+	MsgOrderTopic = "messageOrder"
+	MsgGoodsTopic = "messageGoods"
+
+	//云仓消息topic
+	MsgCloudTopic = "shopApi"
+	//通知消息系统topic
+	NotifyTopic = "notifyMessage"
+
+	//消息类型
+	ProductExpire            = 101
+	ProductModify            = 102
+	ProductPrice             = 103
+	ProductRepost            = 104
+	OrderSendgoods           = 201
+	OrderComfirmReceiveGoods = 202
+	OrderSuccess             = 203
+	OrderCancel              = 204
+	RefundApply              = 300
+	RefundAgree              = 301
+	RefundRefuse             = 302
+	RefundSuccess            = 303
+
+	ImportTags = 401
+
+	ErrMsgParamEmpty = "该类型消息对应字段不能为空"
+	ErrMsgAppIDEmpty = "该类型消息appid字段不能为空"
+)
+
+var (
+	SendMsgType = g.MapIntStr{
+		ProductExpire: "goods.undercarriage",
+		ProductModify: "goods.alter",
+		ProductPrice:  "goods.price.alter",
+		ProductRepost: "goods.on.sale",
+
+		OrderSendgoods:           "order.delivery",
+		OrderComfirmReceiveGoods: "order.delivered",
+		OrderSuccess:             "order.success",
+		OrderCancel:              "order.cancel",
+
+		RefundAgree:   "afterSale.agree",
+		RefundRefuse:  "afterSale.refuse",
+		RefundSuccess: "afterSale.success",
+	}
+)
+
+func (p *NsqProducer) msgValidator(notifyMessage *NotifyMessage) (err error) {
+
+	switch gconv.Int(notifyMessage.Type) / 100 % 10 {
+	case 1:
+		var msgData *GoodsMsgData
+		if err = gconv.Struct(notifyMessage.Data, &msgData); err != nil {
+			return
+		}
+		if len(msgData.GoodsIDs) == 0 {
+			err = errors.New(ErrMsgParamEmpty)
+			return
+		}
+	case 2:
+		var msgData *OrderMsgData
+		if err = gconv.Struct(notifyMessage.Data, &msgData); err != nil {
+			return
+		}
+		if msgData.OrderSn == "" || msgData.Sku == 0 {
+			err = errors.New(ErrMsgParamEmpty)
+			return
+		}
+		if notifyMessage.AppID == 0 {
+			err = errors.New(ErrMsgAppIDEmpty)
+			return
+		}
+	case 3:
+		var msgData *RefundMsgData
+		if err = gconv.Struct(notifyMessage.Data, &msgData); err != nil {
+			return
+		}
+		if msgData.OrderSn == "" || msgData.Sku == 0 || msgData.AfterSaleID == 0 {
+			err = errors.New(ErrMsgParamEmpty)
+			return
+		}
+
+		if notifyMessage.AppID == 0 {
+			err = errors.New(ErrMsgAppIDEmpty)
+			return
+		}
+
+	case 4:
+		var msgData *TagsMsgData
+		if err = gconv.Struct(notifyMessage.Data, &msgData); err != nil {
+			return
+		}
+		if msgData.Tags == "" {
+			err = errors.New(ErrMsgParamEmpty)
+			return
+		}
+		if notifyMessage.AppID == 0 {
+			err = errors.New(ErrMsgAppIDEmpty)
+			return
+		}
+	default:
+		err = errors.New(ErrMsgAppIDEmpty)
+	}
+	return
+}
+
+/*
+	各服务推送通知到消息系统 20220114 gk
+*/
+
+func (p *NsqProducer) NotifyMessage(notifyMessage *NotifyMessage) (err error) {
+
+	err = p.msgValidator(notifyMessage)
+
+	if logs.CheckErr(err,"NotifyMessage"){
+		return
+	}
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	var randNums string
+	for i := 0; i <= 2; i++ {
+		randNums += strconv.Itoa(r.Intn(9))
+	}
+	notifyMessage.ID = gconv.Int(gconv.String(time.Now().UnixNano()/1e6) + randNums)
+
+	jsonBytes, _ := json.Marshal(notifyMessage)
+	err = NsqProducers.Publish(NotifyTopic, string(jsonBytes))
+	if !logs.CheckErr(err, "NotifyMessage") {
+		logs.Info("NotifyMessage", "消息内容:【%v】", string(jsonBytes))
+	}
+	return
+}
+
+/*
+	推送通知到服务系统系统 20220114 gk
+	*****************************************************
+    *******注意：该方法目前只有message-server使用 ***********
+    *****************************************************
+*/
+
+func (p *NsqProducer) NotifyServer(notifyServer *NotifyServer) (err error) {
+
+	jsonBytes, _ := json.Marshal(notifyServer)
+	var pushTopic string
+	switch notifyServer.MsgType / 100 % 10 {
+	case 1:
+		pushTopic = MsgGoodsTopic
+	case 2:
+		pushTopic = MsgOrderTopic
+	case 3:
+		pushTopic = MsgOrderTopic
+	}
+
+	if pushTopic == "" {
+		return
+	}
+	err = NsqProducers.Publish(NotifyTopic, string(jsonBytes))
+	if !logs.CheckErr(err, "NotifyServer") {
+		logs.Info("NotifyServer", "消息内容:【%v】", string(jsonBytes))
+	}
+	return
+}

nsq/model.go → notify/model.go

-package nsq
-
-
-type CommonReq struct {
-	MsgType        int     `json:"msgType"`
-	MsgData        string  `json:"msgData"`
-	Source         int     `json:"source"`
-	MsgSendTime    int64   `json:"msgSendTime"`
-}
-
-
-type CommonReply struct {
-	Success interface{} `json:"success"`
-	Msg     string      `json:"msg"`
-	Data    interface{} `json:"data"`
-	Code    int         `json:"code"`
-}
+package notify
 
 type NotifyReqData struct {
 	RealSource  int    `json:"real_source"`
@@ -30,17 +14,38 @@ type NotifyReq struct {
 	Data NotifyReqData `json:"data"`
 }
 
-type NotifyRes struct {
-	Code int    `json:"code"`
-	Msg  string `json:"msg"`
+//公共message消息
+type NotifyServer struct {
+	MsgType     int         `json:"msg_type"`
+	MsgData     interface{} `json:"msg_data"`
+	Source      int         `json:"source"`
+	MsgSendTime int64       `json:"msg_send_time"`
 }
-type NotifyResNew struct {
-	Msg string `json:"message"`
+
+//通知消息系统
+type NotifyMessage struct {
+	ID    int         `json:"id"`
+	Type  interface{} `json:"type"`
+	Data  interface{} `json:"data"`
+	URL   string      `json:"url"`
+	AppID int         `json:"appID"`
+}
+
+type GoodsMsgData struct {
+	GoodsIDs []int `json:"goods_id"`
 }
 
-type NotifyRefundData struct {
-	AfsServiceId   string `json:"afsServiceId"`
-	State          int    `json:"state"`
-	AfterSrvStatus string `json:"after_srv_status"`
+type OrderMsgData struct {
+	OrderSn     string `json:"orderSn"`
+	Sku         uint   `json:"sku"`
 }
 
+type RefundMsgData struct {
+	AfterSaleID uint   `json:"afterSaleId"`
+	OrderSn     string `json:"orderSn"`
+	Sku         uint   `json:"sku"`
+}
+
+type TagsMsgData struct {
+	Tags string `json:"tags"`
+}

nsq/producer.go → notify/producer.go

-package nsq
+package notify
 
 import (
 	"fmt"

upstream/upstream.go

 package upstream
 
-//推送类型 101 商品下架 102 商品修改 103 价格变更 104 商品上架 201 订单发货 202 确认收货 301 同意售后退款 302 拒绝售后退款 303 售后退款成功',
-
-const (
-	ProductExpire            = 101
-	ProductModify            = 102
-	ProductPrice             = 103
-	ProductRepost            = 104
-	OrderSendgoods           = 201
-	OrderComfirmReceivegoods = 202
-	OrderSuccess             = 203
-	RefundApply				 = 300
-	RefundAgree              = 301
-	RefundRefuse             = 302
-	RefundSuccess            = 303
-)
-
 const (
 	Jd     = 2
 	Ali    = 6