358 lines
11 KiB
Go
358 lines
11 KiB
Go
// Package app wires notice-service modules, health endpoints and background workers.
|
|
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net"
|
|
"sync"
|
|
"time"
|
|
|
|
"google.golang.org/grpc"
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/pkg/grpchealth"
|
|
"hyapp/pkg/healthhttp"
|
|
"hyapp/pkg/logx"
|
|
"hyapp/pkg/rocketmqx"
|
|
"hyapp/pkg/roommq"
|
|
serviceapp "hyapp/pkg/servicekit/app"
|
|
servicegrpc "hyapp/pkg/servicekit/grpcserver"
|
|
servicehealth "hyapp/pkg/servicekit/health"
|
|
servicemq "hyapp/pkg/servicekit/mq"
|
|
"hyapp/pkg/tencentim"
|
|
"hyapp/pkg/usermq"
|
|
"hyapp/pkg/walletmq"
|
|
"hyapp/services/notice-service/internal/config"
|
|
"hyapp/services/notice-service/internal/modules/cpnotice"
|
|
"hyapp/services/notice-service/internal/modules/roomnotice"
|
|
"hyapp/services/notice-service/internal/modules/walletnotice"
|
|
mysqlplatform "hyapp/services/notice-service/internal/platform/mysql"
|
|
)
|
|
|
|
// App owns the notice-service process lifecycle.
|
|
type App struct {
|
|
server *grpc.Server
|
|
listener net.Listener
|
|
health *grpchealth.ServingChecker
|
|
healthHTTP *healthhttp.Server
|
|
store *mysqlplatform.Store
|
|
walletNoticeService *walletnotice.Service
|
|
walletWorkerOptions walletnotice.WalletNoticeWorkerOptions
|
|
walletWorkerEnabled bool
|
|
roomNoticeService *roomnotice.Service
|
|
roomWorkerOptions roomnotice.RoomNoticeWorkerOptions
|
|
roomWorkerEnabled bool
|
|
cpNoticeService *cpnotice.Service
|
|
cpWorkerOptions cpnotice.CPNoticeWorkerOptions
|
|
mqConsumers []*rocketmqx.Consumer
|
|
workers *serviceapp.BackgroundGroup
|
|
closeOnce sync.Once
|
|
}
|
|
|
|
// New initializes MySQL, module repositories and optional Tencent IM publisher.
|
|
func New(cfg config.Config) (*App, error) {
|
|
startupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
store, err := mysqlplatform.Open(startupCtx, cfg.MySQLDSN)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
listener, err := net.Listen("tcp", cfg.GRPCAddr)
|
|
if err != nil {
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
|
|
walletRepo, err := walletnotice.NewMySQLRepository(store.DB, cfg.WalletDatabase)
|
|
if err != nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
roomRepo, err := roomnotice.NewMySQLRepository(store.DB, cfg.RoomDatabase)
|
|
if err != nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
cpRepo := cpnotice.NewMySQLRepository(store.DB)
|
|
var publisher *tencentim.RESTClient
|
|
if cfg.TencentIM.Enabled {
|
|
tencentClient, err := tencentim.NewRESTClient(cfg.TencentIM.RESTConfig())
|
|
if err != nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
publisher = tencentClient
|
|
}
|
|
if cfg.WalletNoticeWorker.Enabled && publisher == nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, errors.New("wallet notice worker requires tencent_im.enabled")
|
|
}
|
|
if cfg.RoomNoticeWorker.Enabled && publisher == nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, errors.New("room notice worker requires tencent_im.enabled")
|
|
}
|
|
if cfg.RocketMQ.WalletOutbox.Enabled && publisher == nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, errors.New("wallet outbox mq consumer requires tencent_im.enabled")
|
|
}
|
|
if cfg.RocketMQ.RoomOutbox.Enabled && publisher == nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, errors.New("room outbox mq consumer requires tencent_im.enabled")
|
|
}
|
|
if cfg.RocketMQ.UserOutbox.Enabled && publisher == nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, errors.New("user outbox mq consumer requires tencent_im.enabled")
|
|
}
|
|
|
|
server := servicegrpc.New("notice-service")
|
|
health, healthHTTP, err := servicehealth.New(server, servicehealth.Config{
|
|
ServiceName: "notice-service",
|
|
HTTPAddr: cfg.HealthHTTPAddr,
|
|
NodeID: cfg.NodeID,
|
|
Dependencies: []grpchealth.Dependency{{
|
|
Name: "mysql",
|
|
Check: store.Ping,
|
|
}},
|
|
})
|
|
if err != nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
|
|
walletSvc := walletnotice.New(walletnotice.Config{NodeID: cfg.NodeID}, walletRepo, publisher)
|
|
roomSvc := roomnotice.New(roomnotice.Config{NodeID: cfg.NodeID}, roomRepo, publisher)
|
|
roomOptions := roomNoticeWorkerOptions(cfg.RoomNoticeWorker)
|
|
cpSvc := cpnotice.New(cpnotice.Config{NodeID: cfg.NodeID, GroupIDPrefix: cfg.TencentIM.GroupIDPrefix}, cpRepo, publisher)
|
|
cpOptions := cpNoticeWorkerOptions(cfg.CPNoticeWorker)
|
|
mqConsumers := make([]*rocketmqx.Consumer, 0, 3)
|
|
if cfg.RocketMQ.WalletOutbox.Enabled {
|
|
consumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.ConsumerGroup, cfg.RocketMQ.WalletOutbox.ConsumerMaxReconsumeTimes))
|
|
if err != nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, walletmq.TagWalletOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
|
walletMessage, err := walletmq.DecodeWalletOutboxMessage(message.Body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = walletSvc.ProcessWalletOutboxMessage(appcode.WithContext(ctx, walletMessage.AppCode), walletMessage, walletNoticeWorkerOptions(cfg.WalletNoticeWorker))
|
|
return err
|
|
}); err != nil {
|
|
_ = consumer.Shutdown()
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
mqConsumers = append(mqConsumers, consumer)
|
|
}
|
|
if cfg.RocketMQ.RoomOutbox.Enabled {
|
|
consumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.RoomOutbox.ConsumerGroup, cfg.RocketMQ.RoomOutbox.ConsumerMaxReconsumeTimes))
|
|
if err != nil {
|
|
shutdownConsumers(mqConsumers)
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
if err := consumer.Subscribe(cfg.RocketMQ.RoomOutbox.Topic, roommq.TagRoomOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
|
envelope, _, err := roommq.DecodeRoomOutboxMessage(message.Body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = roomSvc.ProcessRoomOutboxEnvelope(appcode.WithContext(ctx, envelope.GetAppCode()), envelope, roomOptions)
|
|
return err
|
|
}); err != nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
shutdownConsumers(mqConsumers)
|
|
return nil, err
|
|
}
|
|
mqConsumers = append(mqConsumers, consumer)
|
|
}
|
|
if cfg.RocketMQ.UserOutbox.Enabled {
|
|
consumer, err := rocketmqx.NewConsumer(rocketMQConsumerConfig(cfg.RocketMQ, cfg.RocketMQ.UserOutbox.ConsumerGroup, cfg.RocketMQ.UserOutbox.ConsumerMaxReconsumeTimes))
|
|
if err != nil {
|
|
shutdownConsumers(mqConsumers)
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
return nil, err
|
|
}
|
|
if err := consumer.Subscribe(cfg.RocketMQ.UserOutbox.Topic, usermq.TagUserOutboxEvent, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
|
userMessage, err := usermq.DecodeUserOutboxMessage(message.Body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = cpSvc.ProcessUserOutboxMessage(appcode.WithContext(ctx, userMessage.AppCode), userMessage, cpOptions)
|
|
return err
|
|
}); err != nil {
|
|
_ = listener.Close()
|
|
_ = store.Close()
|
|
shutdownConsumers(mqConsumers)
|
|
return nil, err
|
|
}
|
|
mqConsumers = append(mqConsumers, consumer)
|
|
}
|
|
return &App{
|
|
server: server,
|
|
listener: listener,
|
|
health: health,
|
|
healthHTTP: healthHTTP,
|
|
store: store,
|
|
walletNoticeService: walletSvc,
|
|
walletWorkerOptions: walletNoticeWorkerOptions(cfg.WalletNoticeWorker),
|
|
walletWorkerEnabled: cfg.WalletNoticeWorker.Enabled,
|
|
roomNoticeService: roomSvc,
|
|
roomWorkerOptions: roomOptions,
|
|
roomWorkerEnabled: cfg.RoomNoticeWorker.Enabled,
|
|
cpNoticeService: cpSvc,
|
|
cpWorkerOptions: cpOptions,
|
|
mqConsumers: mqConsumers,
|
|
workers: serviceapp.NewBackground(context.Background()),
|
|
}, nil
|
|
}
|
|
|
|
// Run starts health serving, optional workers and the gRPC health endpoint.
|
|
func (a *App) Run() error {
|
|
a.runHealthHTTP()
|
|
a.health.MarkServing()
|
|
if err := a.startMQ(); err != nil {
|
|
a.health.MarkStopped()
|
|
return err
|
|
}
|
|
defer func() {
|
|
if a.workers != nil {
|
|
a.workers.StopAndWait()
|
|
}
|
|
a.shutdownMQ()
|
|
a.health.MarkStopped()
|
|
}()
|
|
if a.walletWorkerEnabled && a.walletNoticeService != nil {
|
|
a.workers.Go(func(ctx context.Context) {
|
|
a.walletNoticeService.RunWalletBalanceWorker(ctx, a.walletWorkerOptions)
|
|
})
|
|
}
|
|
if a.roomWorkerEnabled && a.roomNoticeService != nil {
|
|
a.workers.Go(func(ctx context.Context) {
|
|
a.roomNoticeService.RunRoomKickWorker(ctx, a.roomWorkerOptions)
|
|
})
|
|
}
|
|
err := a.server.Serve(a.listener)
|
|
if errors.Is(err, grpc.ErrServerStopped) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
// Close drains workers before closing gRPC and MySQL resources.
|
|
func (a *App) Close() {
|
|
a.closeOnce.Do(func() {
|
|
a.health.MarkDraining()
|
|
if a.workers != nil {
|
|
a.workers.StopAndWait()
|
|
}
|
|
a.shutdownMQ()
|
|
servicegrpc.GracefulStop(a.server, 15*time.Second)
|
|
a.closeHealthHTTP()
|
|
if a.store != nil {
|
|
_ = a.store.Close()
|
|
}
|
|
})
|
|
}
|
|
|
|
func (a *App) runHealthHTTP() {
|
|
if a.healthHTTP == nil {
|
|
return
|
|
}
|
|
go func() {
|
|
if err := a.healthHTTP.Run(); err != nil {
|
|
logx.Error(context.Background(), "health_http_run_failed", err)
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (a *App) closeHealthHTTP() {
|
|
if a.healthHTTP == nil {
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
_ = a.healthHTTP.Close(ctx)
|
|
}
|
|
|
|
func walletNoticeWorkerOptions(cfg config.WalletNoticeWorkerConfig) walletnotice.WalletNoticeWorkerOptions {
|
|
return walletnotice.WalletNoticeWorkerOptions{
|
|
PollInterval: cfg.PollInterval,
|
|
BatchSize: cfg.BatchSize,
|
|
LockTTL: cfg.LockTTL,
|
|
PublishTimeout: cfg.PublishTimeout,
|
|
MaxRetryCount: cfg.MaxRetryCount,
|
|
InitialBackoff: cfg.InitialBackoff,
|
|
MaxBackoff: cfg.MaxBackoff,
|
|
}
|
|
}
|
|
|
|
func roomNoticeWorkerOptions(cfg config.WalletNoticeWorkerConfig) roomnotice.RoomNoticeWorkerOptions {
|
|
return roomnotice.RoomNoticeWorkerOptions{
|
|
PollInterval: cfg.PollInterval,
|
|
BatchSize: cfg.BatchSize,
|
|
LockTTL: cfg.LockTTL,
|
|
PublishTimeout: cfg.PublishTimeout,
|
|
MaxRetryCount: cfg.MaxRetryCount,
|
|
InitialBackoff: cfg.InitialBackoff,
|
|
MaxBackoff: cfg.MaxBackoff,
|
|
}
|
|
}
|
|
|
|
func cpNoticeWorkerOptions(cfg config.WalletNoticeWorkerConfig) cpnotice.CPNoticeWorkerOptions {
|
|
return cpnotice.CPNoticeWorkerOptions{
|
|
BatchSize: cfg.BatchSize,
|
|
LockTTL: cfg.LockTTL,
|
|
PublishTimeout: cfg.PublishTimeout,
|
|
MaxRetryCount: cfg.MaxRetryCount,
|
|
InitialBackoff: cfg.InitialBackoff,
|
|
MaxBackoff: cfg.MaxBackoff,
|
|
}
|
|
}
|
|
|
|
func (a *App) startMQ() error {
|
|
return servicemq.StartConsumers(a.mqConsumers)
|
|
}
|
|
|
|
func (a *App) shutdownMQ() {
|
|
servicemq.ShutdownConsumers(a.mqConsumers)
|
|
}
|
|
|
|
func shutdownConsumers(consumers []*rocketmqx.Consumer) {
|
|
servicemq.ShutdownConsumers(consumers)
|
|
}
|
|
|
|
func rocketMQConsumerConfig(cfg config.RocketMQConfig, group string, maxReconsumeTimes int32) rocketmqx.ConsumerConfig {
|
|
return rocketmqx.ConsumerConfig{
|
|
EndpointConfig: rocketmqx.EndpointConfig{
|
|
NameServers: cfg.NameServers,
|
|
NameServerDomain: cfg.NameServerDomain,
|
|
AccessKey: cfg.AccessKey,
|
|
SecretKey: cfg.SecretKey,
|
|
SecurityToken: cfg.SecurityToken,
|
|
Namespace: cfg.Namespace,
|
|
VIPChannel: cfg.VIPChannel,
|
|
},
|
|
GroupName: group,
|
|
MaxReconsumeTimes: maxReconsumeTimes,
|
|
ConsumeRetryDelay: time.Second,
|
|
ConsumePullBatch: 32,
|
|
ConsumePullTimeout: 15 * time.Minute,
|
|
}
|
|
}
|