MessageCenter/rpc/message.go

// Code generated by goctl. DO NOT EDIT!
// Source: message.proto

package main

import (
	"app.yhyue.com/moapp/MessageCenter/entity"
	"app.yhyue.com/moapp/MessageCenter/rpc/internal/common"
	"app.yhyue.com/moapp/MessageCenter/rpc/internal/config"
	"app.yhyue.com/moapp/MessageCenter/rpc/internal/server"
	"app.yhyue.com/moapp/MessageCenter/rpc/internal/svc"
	"app.yhyue.com/moapp/MessageCenter/rpc/type/message"
	mc "app.yhyue.com/moapp/jybase/common"
	"app.yhyue.com/moapp/jybase/endless"
	"app.yhyue.com/moapp/jybase/mail"
	m "app.yhyue.com/moapp/jybase/mongodb"
	"app.yhyue.com/moapp/jybase/mysql"
	"app.yhyue.com/moapp/jybase/redis"
	"context"
	"flag"
	"fmt"
	"github.com/ClickHouse/clickhouse-go/v2"
	"github.com/go-xorm/xorm"
	_ "github.com/gogf/gf/contrib/drivers/clickhouse/v2"
	"github.com/zeromicro/go-zero/core/conf"
	"github.com/zeromicro/go-zero/core/logx"
	"github.com/zeromicro/go-zero/zrpc"
	"google.golang.org/grpc"
	"log"
	"time"
)

var configFile = flag.String("f", "etc/message.yaml", "the config file")
var logFile = flag.String("lf", "etc/logs.yaml", "the logs file")
var logc entity.Logc
var c config.Config

func main() {
	flag.Parse()
	conf.MustLoad(*configFile, &c)
	go common.SaveTask()
	go func() {
		err := endless.ListenAndServe(":"+mc.InterfaceToStr(c.WebRpcPort), nil, func() {})
		if err != nil {
			log.Println("ListenAndServe: ", err)
		}
	}()
	ctx := svc.NewServiceContext(c)
	srv := server.NewMessageServer(ctx)

	s := zrpc.MustNewServer(c.RpcServerConf, func(grpcServer *grpc.Server) {
		message.RegisterMessageServer(grpcServer, srv)
	})
	defer s.Stop()
	fmt.Printf("Starting rpc server at %s...\n", c.ListenOn)
	s.Start()
}

// 创建orm引擎
func init() {
	conf.MustLoad(*configFile, &config.ConfigJson)
	log.Println("开始初始化数据库。。。。。")
	//初始化clickhouse
	err := connectClickhouse(config.ConfigJson.Clickhouse)
	if err != nil {
		log.Println("初始化clickhouse出错：", err)
	}

	//初始化mysql
	entity.Mysql = &mysql.Mysql{
		Address:      config.ConfigJson.DataSource.Address,
		UserName:     config.ConfigJson.DataSource.UserName,
		PassWord:     config.ConfigJson.DataSource.PassWord,
		DBName:       config.ConfigJson.DataSource.DbName,
		MaxOpenConns: config.ConfigJson.DataSource.MaxOpenConns,
		MaxIdleConns: config.ConfigJson.DataSource.MaxIdleConns,
	}
	entity.Mysql.Init()

	common.AllMsgType, entity.MessageColumn = common.MessageType()
	// 初始化需要展示消息细化分类的groupid 和对应的class信息
	common.InitClassSearchMap()

	//初始化basemysql
	entity.BaseMysql = &mysql.Mysql{
		Address:      config.ConfigJson.BaseSource.Address,
		UserName:     config.ConfigJson.BaseSource.UserName,
		PassWord:     config.ConfigJson.BaseSource.PassWord,
		DBName:       config.ConfigJson.BaseSource.DbName,
		MaxOpenConns: config.ConfigJson.BaseSource.MaxOpenConns,
		MaxIdleConns: config.ConfigJson.BaseSource.MaxIdleConns,
	}
	entity.BaseMysql.Init()
	//初始化 redis
	if config.ConfigJson.RedisAddr != "" {
		redis.InitRedis(config.ConfigJson.RedisAddr)
	}
	// 初始化mongo
	if config.ConfigJson.Mongodb != nil {
		log.Println("初始化 mongodb main")
		entity.MQFW = m.MongodbSim{
			MongodbAddr: config.ConfigJson.Mongodb.Address,
			Size:        config.ConfigJson.Mongodb.Size,
			DbName:      config.ConfigJson.Mongodb.DbName,
			ReplSet:     config.ConfigJson.Mongodb.ReplSet,
		}
		entity.MQFW.InitPool()
	}
	// 初始化发送邮件
	for _, v := range config.ConfigJson.Mail {
		entity.GmailAuth = append(entity.GmailAuth, &mail.GmailAuth{
			SmtpHost: v.Addr,
			SmtpPort: v.Port,
			User:     v.User,
			Pwd:      v.Pwd,
		})
	}
	// 初始化消息保存并发通道
	entity.SaveConcurrencyChan = make(chan int, config.ConfigJson.SaveConcurrency)
	//初始化日志信息
	conf.MustLoad(*logFile, &logc)
	if len(logc.Level) > 0 {
		for _, v := range logc.Level {
			logx.MustSetup(logx.LogConf{
				Mode:     logc.Mode,
				Path:     logc.Path,
				Level:    v,
				KeepDays: logc.KeepDays,
			})
			logx.Info(v, "--日志记录")
		}
	}
	if config.ConfigJson.SurvivalTime != 0 {
		entity.SurvivalTime = config.ConfigJson.SurvivalTime
	}
	entity.Engine, _ = xorm.NewEngine("mysql", config.ConfigJson.TidbEng)
	entity.Engine.ShowSQL(true)
	common.LoadTask()
}

// 创建clickhouse连接
func connectClickhouse(cHouseConfig *config.CHouseConfig) error {
	var (
		ctx = context.Background()
		err error
	)
	entity.ClickhouseConn, err = clickhouse.Open(&clickhouse.Options{
		Addr:         []string{cHouseConfig.Addr},
		DialTimeout:  10 * time.Second,
		MaxIdleConns: cHouseConfig.MaxIdleConns,
		MaxOpenConns: cHouseConfig.MaxOpenConns,
		Auth: clickhouse.Auth{
			Database: cHouseConfig.DbName,
			Username: cHouseConfig.UserName,
			Password: cHouseConfig.Password,
		},
		Debugf: func(format string, v ...interface{}) {
			fmt.Printf(format, v)
		},
	})
	if err != nil {
		return err
	}
	if err := entity.ClickhouseConn.Ping(ctx); err != nil {
		if exception, ok := err.(*clickhouse.Exception); ok {
			fmt.Printf("Exception [%d] %s \n%s\n", exception.Code, exception.Message, exception.StackTrace)
		}
		return err
	}
	return nil
}