message/test/main_test.go

package test

import (
	"encoding/json"
	"log"
	"testing"
	"time"

	"app.yhyue.com/moapp/jybase/mongodb"
	"app.yhyue.com/moapp/message/model"
	"github.com/nsqio/go-nsq"
)

func Test_Producer(t *testing.T) {
	Mgo_Log := &mongodb.MongodbSim{
		MongodbAddr: "192.168.3.128:27090",
		Size:        5,
		DbName:      "qfw",
		UserName:    "admin",
		Password:    "123456",
		ReplSet:     "",
	}
	Mgo_Log.InitPool()
	log.Println("初始化 mongodb")
	// Instantiate a producer.
	config := nsq.NewConfig()
	producer, err := nsq.NewProducer("192.168.3.240:4260", config)
	if err != nil {
		log.Println(err)
		return
	}
	/*E_code对应的值：
	 *	jyweb_article_open 打开招投标信息三级页
	 *	jydocs_doc_open 打开文库三级页
	 *	jyapp_wx_register app微信注册
	 *	jyapp_phone_register app手机号注册
	 *	jypc_phone_register pc端手机号注册
	 *	jywx_subscribe_new 微信新用户关注
	 *	jywx_subscribe_invite 已邀请并产生了新用户
	 *	jywx_subscribe_invited 被邀请产生新用户
	 */
	msg := &model.Message{
		E_code:   "jyweb_article_open",
		E_userId: "5d6378301c298a5aac7b5402",
		E_time:   time.Now().Unix(), //1605223065
		E_app:    "jywx_node",
	}
	// Synchronously publish a single message to the specified topic.
	// Messages can also be sent asynchronously and/or in batches.
	b, _ := json.Marshal(msg)
	err = producer.Publish("jy_event", b)
	if err != nil {
		log.Println(err)
		return
	}
	log.Println("消息发布成功")
	//保存消息到日志库
	Mgo_Log.Save("nsq_logs", map[string]interface{}{
		"createtime": time.Now().Unix(),
		"body":       msg,
		"type":       "producer", //producer or consumer
	})
	// Gracefully stop the producer.
	producer.Stop()
}