119 lines
2.9 KiB
Go
119 lines
2.9 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net"
|
|
"sync"
|
|
"time"
|
|
|
|
"google.golang.org/grpc"
|
|
healthgrpc "google.golang.org/grpc/health/grpc_health_v1"
|
|
walletv1 "hyapp.local/api/proto/wallet/v1"
|
|
"hyapp/pkg/grpchealth"
|
|
"hyapp/pkg/grpcshutdown"
|
|
"hyapp/pkg/healthhttp"
|
|
"hyapp/pkg/logx"
|
|
"hyapp/services/wallet-service/internal/config"
|
|
walletservice "hyapp/services/wallet-service/internal/service/wallet"
|
|
mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
|
|
grpcserver "hyapp/services/wallet-service/internal/transport/grpc"
|
|
)
|
|
|
|
// App 装配 wallet-service gRPC 入口。
|
|
type App struct {
|
|
server *grpc.Server
|
|
listener net.Listener
|
|
health *grpchealth.ServingChecker
|
|
healthHTTP *healthhttp.Server
|
|
mysqlRepo *mysqlstorage.Repository
|
|
closeOnce sync.Once
|
|
}
|
|
|
|
// New 初始化 wallet-service。
|
|
func New(cfg config.Config) (*App, error) {
|
|
startupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
// wallet-service 的账户、交易、分录和 outbox 必须共享同一个 MySQL 事务边界。
|
|
repository, err := mysqlstorage.Open(startupCtx, cfg.MySQLDSN)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
listener, err := net.Listen("tcp", cfg.GRPCAddr)
|
|
if err != nil {
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
|
|
server := grpc.NewServer(grpc.UnaryInterceptor(logx.UnaryServerInterceptor("wallet-service")))
|
|
svc := walletservice.New(repository)
|
|
walletv1.RegisterWalletServiceServer(server, grpcserver.NewServer(svc))
|
|
health := grpchealth.NewServingChecker("wallet-service", grpchealth.Dependency{
|
|
Name: "mysql",
|
|
Check: repository.Ping,
|
|
})
|
|
healthgrpc.RegisterHealthServer(server, grpchealth.NewServer(health))
|
|
healthHTTP, err := healthhttp.New(cfg.HealthHTTPAddr, cfg.NodeID, health)
|
|
if err != nil {
|
|
_ = listener.Close()
|
|
_ = repository.Close()
|
|
return nil, err
|
|
}
|
|
|
|
return &App{
|
|
server: server,
|
|
listener: listener,
|
|
health: health,
|
|
healthHTTP: healthHTTP,
|
|
mysqlRepo: repository,
|
|
}, nil
|
|
}
|
|
|
|
// Run 启动 gRPC 服务。
|
|
func (a *App) Run() error {
|
|
a.runHealthHTTP()
|
|
a.health.MarkServing()
|
|
err := a.server.Serve(a.listener)
|
|
a.health.MarkStopped()
|
|
if errors.Is(err, grpc.ErrServerStopped) {
|
|
return nil
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// Close 关闭 gRPC 服务。
|
|
func (a *App) Close() {
|
|
a.closeOnce.Do(func() {
|
|
a.health.MarkDraining()
|
|
grpcshutdown.GracefulStop(a.server, 15*time.Second)
|
|
a.closeHealthHTTP()
|
|
if a.mysqlRepo != nil {
|
|
// MySQL 连接池最后关闭，避免正在 drain 的扣费请求丢失提交能力。
|
|
_ = a.mysqlRepo.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)
|
|
}
|