sig-pub/cmd/trading/main.go

package main

import (
	"context"
	"fmt"
	"net"
	"sig-pub/api/pb"
	"sig-pub/internal/trading"
	"sig-pub/pkg/client"
	"sig-pub/pkg/config"
	"sig-pub/pkg/grpc/discovery"
	"sig-pub/pkg/grpc/interceptor"
	"sig-pub/pkg/mq"
	"sig-pub/pkg/storage/persist"
	"sig-pub/pkg/utils/exit"
	"sig-pub/pkg/zlog"

	"github.com/hashicorp/consul/api"
	_ "github.com/mostynb/go-grpc-compression/snappy" // 注册grpc snappy compress
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/reflection"
)

type TradingConf struct {
	Register       discovery.Server
	GrpcReflection bool
}

func main() {
	// load config
	conf := config.MustLoadConfig(new(config.Configuration), "config/config.toml")
	tradingConf := config.MustLoadConfig(new(TradingConf), "config/trading.toml")

	// consul
	cc := api.DefaultConfig()
	cc.Address = conf.Consul.Address
	consulClient, err := api.NewClient(cc)
	if err != nil {
		panic(fmt.Errorf("consul client error: %v", err))
	}
	dis := discovery.NewConsulDiscovery(consulClient)
	resolver := dis.Resolver()

	// nats
	if err = mq.InitNats(conf.Nats.Url); err != nil {
		panic(err)
	}

	// database
	db, err := conf.Database.Postgres.NewGormDB()
	if err != nil {
		panic(err)
	}
	rdb := persist.NewDB(db)
	if err := rdb.Init(); err != nil {
		panic(err)
	}
	pgBatchWriter := persist.NewPGBatchWriter()
	if err := pgBatchWriter.Init(context.Background(), conf.Database.Postgres.ConnString()); err != nil {
		panic(err)
	}
	// clickhouse
	// ckBatchWriter := ck.NewClickhouseBatchWriter(conf.Database.Clickhouse)
	// if err := ckBatchWriter.Init(); err != nil {
	// 	panic(err)
	// }
	// ckDB := ck.NewClickhouseDB(conf.Database.Clickhouse)
	// if err := ckDB.Init(); err != nil {
	// 	panic(err)
	// }
	tradingDataPersist := trading.NewTradingDataPersist(rdb, pgBatchWriter, nil, nil)
	if err := tradingDataPersist.Init(); err != nil {
		panic(err)
	}

	// new market grpc client
	marketClient, err := client.NewMarketClient(
		grpc.WithResolvers(resolver),
		grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		panic(err)
	}
	tradeInstanceAside := client.NewTradeInstanceAside(marketClient)

	// new exchange grpc client
	exchangeClient, err := client.NewExchangeClient(
		grpc.WithResolvers(resolver),
		grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		panic(err)
	}

	// services
	tradingService := trading.NewTradingService(tradeInstanceAside, exchangeClient, tradingDataPersist)
	if err := tradingService.Init(); err != nil {
		panic(err)
	}

	// grpc server
	tradingGrpcServer := trading.NewTradingGrpcServer(tradingService)
	if err := tradingGrpcServer.Init(); err != nil {
		panic(err)
	}
	grpcServer := grpc.NewServer(config.GetGrpcOptions(
		conf.Grpc,
		grpc.UnaryInterceptor(interceptor.RecoverInterceptor),
	)...)
	if tradingConf.GrpcReflection {
		// 注册反射服务
		reflection.Register(grpcServer)
	}
	pb.RegisterTradingServiceServer(grpcServer, tradingGrpcServer)
	exit.AddHook(grpcServer.GracefulStop, exit.WithOrderFront())

	// consul 服务注册
	register := tradingConf.Register
	register.Name = pb.TradingService_ServiceDesc.ServiceName
	if deregister, err := dis.Registry(grpcServer, register); err != nil {
		panic(err)
	} else {
		defer deregister()
	}

	// run grpc server
	go func() {
		listen, err := net.Listen("tcp", tradingConf.Register.Addr)
		if err != nil {
			panic(err)
		}
		zlog.Infof("%s grpc server running %s\n", register.Name, listen.Addr().String())
		if err := grpcServer.Serve(listen); err != nil {
			panic(err)
		}
	}()

	exit.Await()
}