trading sub grpc stream kline

.vscode/launch.json

@@ -41,13 +41,13 @@
             "output": "${workspaceFolder}/run/exchange",
         },
         {
-            "name": "indicator",
+            "name": "trading",
             "type": "go",
             "request": "launch",
             "mode": "auto",
-            "program": "${workspaceFolder}/cmd/indicator",
+            "program": "${workspaceFolder}/cmd/trading",
             "cwd": "${workspaceFolder}",
-            "output": "${workspaceFolder}/run/indicator",
+            "output": "${workspaceFolder}/run/trading",
         },
         {
             "name": "test",

api/exchange.proto

@@ -32,6 +32,7 @@ message StreamKline {
   repeated Kline klines = 2;
   ExchangeType exchange = 3; // 交易所
   string instId = 4;         // 交易产品id
+  int64 streamId = 5;        // subscribe stream id
 }
 
 message ReqExchanges {

cmd/exchange/main.go

@@ -6,7 +6,7 @@ import (
 	"sig-pub/api/pb"
 	"sig-pub/internal/exchange"
 	"sig-pub/internal/exchange/okx"
-	"sig-pub/pkg/aside"
+	"sig-pub/pkg/client"
 	"sig-pub/pkg/config"
 	"sig-pub/pkg/grpc/discovery"
 	"sig-pub/pkg/grpc/interceptor"
@@ -36,11 +36,11 @@ func main() {
 	// consul 配置
 	cc := api.DefaultConfig()
 	cc.Address = conf.Consul.Address
-	client, err := api.NewClient(cc)
+	consulClient, err := api.NewClient(cc)
 	if err != nil {
 		panic(fmt.Errorf("consul client error: %v", err))
 	}
-	dis := discovery.NewConsulDiscovery(client)
+	dis := discovery.NewConsulDiscovery(consulClient)
 	resolver := dis.Resolver()
 
 	// new market grpc client
@@ -53,7 +53,7 @@ func main() {
 		panic(err)
 	}
 	marketClient := pb.NewMarketServiceClient(marketConn)
-	tradeInstanceAside := aside.NewTradeInstanceAside(marketClient)
+	tradeInstanceAside := client.NewTradeInstanceAside(marketClient)
 
 	// kvrocks db
 	kvdb := kvrocks.NewKVRocksDB(conf.Database.Kvrocks)
@@ -89,8 +89,7 @@ func main() {
 	grpcServer := grpc.NewServer(config.GetGrpcOptions(
 		conf.Grpc,
 		grpc.UnaryInterceptor(interceptor.RecoverInterceptor),
-	)...,
+	)...)
-	)
 	if exchangeConf.GrpcReflection {
 		// 注册反射服务
 		reflection.Register(grpcServer)

cmd/trading/main.go

@@ -0,0 +1,101 @@
+package main
+
+import (
+	"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/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()
+
+	// 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)
+	if err := tradingService.Init(); err != nil {
+		panic(err)
+	}
+
+	// grpc server
+	tradingGrpcServer := trading.NewTradingGrpcServer()
+	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 err := dis.Registry(grpcServer, register); err != nil {
+		panic(err)
+	}
+
+	// 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()
+}

cmd/trading/test_exchange_subscribe.go

@@ -14,7 +14,7 @@ import (
 	"google.golang.org/grpc/keepalive"
 )
 
-func main() {
+func testSubscribe() {
 	// 配置 keepalive 参数
 	keepaliveParams := keepalive.ClientParameters{
 		Time:                30 * time.Second, // 发送 ping 的间隔

config/trading.toml

@@ -0,0 +1,7 @@
+
+grpcReflection = true     # 注册grpc反射服务
+
+[register]
+nodeId = 1               # grpc服务节点id, 多实例唯一
+addr = ":8031"           # grpc 服务端口
+attrs = { weight = 10 }  # grpc 服务权重

internal/exchange/exchange_grpc_server.go

@@ -34,7 +34,7 @@ func (svr *ExchangeGrpcServer) Init() (err error) {
 // SubscribeKline 订阅k线stream
 func (svr *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]) (err0 error) {
 	streamId := atomic.AddInt64(&svr.klineStreamId, 1)
-	// subKey = /kline/exchange/instId/interval/confirm
+	// subKey = /kline/exchange/instId/interval/confirm -> /kline/OKX/DOGE-USDT-SWAP/1s/1
 
 	// 接收消息的goroutine
 	recvChan := make(chan *pb.ReqStreamSubscribeKline)
@@ -42,12 +42,12 @@ func (svr *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb
 		for {
 			msg, err := stream.Recv()
 			if err == io.EOF {
-				zlog.Infof("关闭EOF: %v", err)
+				zlog.Infof("recv close EOF")
 				close(recvChan)
 				return
 			}
 			if err != nil {
-				zlog.Infof("接收错误: %v", err)
+				zlog.Infof("recv error: %v", err)
 				close(recvChan)
 				return
 			}
@@ -83,7 +83,7 @@ func (svr *ExchangeGrpcServer) SubscribeKline(stream grpc.BidiStreamingServer[pb
 						}
 						for _, confirm := range confirms {
 							subKey := fmt.Sprintf("/kline/%s/%s/%s/%d", exchange.String(), instId, interval, confirm)
-							zlog.Infof("stream: %d sub: %s", streamId, subKey)
+							zlog.Debugf("stream: id=%d, sub %s", streamId, subKey)
 							switch msg.SubType {
 							case pb.SubscribeType_Subscribe:
 								svr.klineSubscriber.Subscribe(subKey, streamId, stream)

internal/exchange/exchange_service.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"runtime"
 	"sig-pub/api/pb"
-	"sig-pub/pkg/aside"
+	"sig-pub/pkg/client"
 	"sig-pub/pkg/data"
 	"sig-pub/pkg/types"
 	"sig-pub/pkg/utils/collect"
@@ -23,7 +23,7 @@ import (
 // ExchangeService 交易所服务
 type ExchangeService struct {
 	exchangeMap         map[pb.ExchangeType]*Exchange
-	tradeInstanceAside  *aside.TradeInstanceAside
+	tradeInstanceAside  *client.TradeInstanceAside
 	exchangeDataPersist *ExchangeDataPersist
 
 	klinePublisher *Publisher[int64, grpc.BidiStreamingServer[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]]
@@ -31,7 +31,7 @@ type ExchangeService struct {
 
 // exchanges: 支持的数据源交易所
 func NewExchangeService(
-	tradeInstanceAside *aside.TradeInstanceAside,
+	tradeInstanceAside *client.TradeInstanceAside,
 	exchangeDataPersist *ExchangeDataPersist,
 	exchanges ...*Exchange,
 ) *ExchangeService {
@@ -203,8 +203,9 @@ func (svc *ExchangeService) consumerKline(exchange *Exchange, c <-chan *types.Ch
 			if len(kline.Klines) == 0 {
 				continue
 			}
-			subs := svc.klinePublisher.Publisher(pubKey)
+			tids, subs := svc.klinePublisher.Publisher(pubKey)
-			for _, sub := range subs {
+			for i, sub := range subs {
+				kline.StreamId = tids[i]
 				if err := sub.Send(&pb.RspStreamSubscribeKline{Kline: kline}); err != nil {
 					zlog.Error(err)
 				}

internal/exchange/publisher.go

@@ -45,12 +45,15 @@ func (p *Publisher[TID, T]) UnsubscribeAll(tid TID) {
 }
 
 // Publisher 获取匹配的 subscribers
-func (p *Publisher[TID, T]) Publisher(k string) (subs []T) {
+func (p *Publisher[TID, T]) Publisher(k string) (tids []TID, subs []T) {
 	p.m0.LoadRLock(k, func(subIds map[TID]T, ok bool) {
 		if !ok {
 			return
 		}
-		for _, sub := range subIds {
+		tids = make([]TID, 0, len(subIds))
+		subs = make([]T, 0, len(subIds))
+		for tid, sub := range subIds {
+			tids = append(tids, tid)
 			subs = append(subs, sub)
 		}
 	})

internal/market/trade_instance_service.go

@@ -148,7 +148,7 @@ func (s *TradeInstanceService) UpdateInstanceStatus(inst *args.UpdateTradeInstan
 // ListExchangeTradeInstance 获取指定交易所的正常状态的交易产品
 func (s *TradeInstanceService) ListExchangeTradeInstance(exchange pb.ExchangeType) (exchangesInsts []*entity.TradeInstanceExchange, err error) {
 	err = s.db.Select(&exchangesInsts, `
-		select * from t_trade_instance_exchange where exchange = ? and status in ? order by inst_id, exchange order by update_time asc
+		select * from t_trade_instance_exchange where exchange = ? and status in ? order by inst_id
 	`, exchange, []data.Status{data.StatusOk, data.StatusProcessing})
 	if err != nil {
 		return

internal/trading/kline_series.go

@@ -12,11 +12,11 @@ type KlineSeries struct {
 	Interval   types.Interval
 	Ts         int64
 	klines     []*types.Kline
-	klineStore KlineStore
+	klineStore *KlineStore
 }
 
-func NewKlineSeries(ts int64, interval types.Interval, klineStore KlineStore) KlineSeries {
+func NewKlineSeries(ts int64, interval types.Interval, klineStore *KlineStore) *KlineSeries {
-	return KlineSeries{
+	return &KlineSeries{
 		Ts:         ts,
 		Interval:   interval,
 		klineStore: klineStore,
@@ -25,16 +25,21 @@ func NewKlineSeries(ts int64, interval types.Interval, klineStore KlineStore) Kl
 
 // Get
 // [0]当前k线
-func (a KlineSeries) Get(start int16) (kline types.Kline) {
+func (a KlineSeries) Get(start int16) types.Kline {
-	ts := a.Interval.MustAddMul(a.Ts, int64(-start))
+	index := len(a.klines) - 1 - int(start)
+	if index >= 0 && index < len(a.klines)-1 {
+		return *(a.klines[index])
+	}
 
+	// todo query store
+	ts := a.Interval.MustAddMul(a.Ts, int64(-start))
 	for _, k := range a.klines {
 		if k.Ts == ts {
 			return *k
 		}
 	}
-	// todo query tsdb
+
-	return
+	panic("kline not exists")
 }
 
 // Series [start...end]
@@ -48,6 +53,7 @@ func (a KlineSeries) Series(start, end int16) (klines series.Klines) {
 	return
 }
 
+// 检查k线序列完整
 func (s *KlineSeries) Update(kline *types.Kline) []types.Kline {
 	s.klines = append(s.klines, kline)
 	// todo copy(s.klines, s.klines[0:1]) set index=20, ts=kline.ts

internal/trading/kline_store.go

@@ -9,19 +9,26 @@ import (
 
 type KlineStore struct {
 	vmdb  vmts.VictoriaMetricsTSDB
-	store [3]*collect.ConcurrentMap[string, collect.ConcurrentMap[types.Interval, *KlineSeries]] // K线列表: []exchange<instId, interval, klines>
+	store [3]*collect.ConcurrentMap[string, *collect.ConcurrentMap[types.Interval, *KlineSeries]] // K线列表: []exchange<instId, interval, klines>
 }
 
 func NewKlineSeriesStore(vmdb vmts.VictoriaMetricsTSDB) (kss *KlineStore) {
 	kss = &KlineStore{
 		vmdb: vmdb,
 	}
-	kss.store[pb.ExchangeType_OKX] = collect.NewConcurrentMap[string, collect.ConcurrentMap[types.Interval, *KlineSeries]](64, func(s string) string { return s })
+	kss.store[pb.ExchangeType_OKX] = collect.NewConcurrentMap[string, *collect.ConcurrentMap[types.Interval, *KlineSeries]](64, func(s string) string { return s })
 	// kss.klines[pb.ExchangeType_BINANCE] =
 	return
 }
 
-func (s *KlineStore) Update(exchange pb.ExchangeType, kline *types.Kline) (k types.Kline) {
+func (s *KlineStore) Update(exchange pb.ExchangeType, instId string, kline *types.Kline) (k types.Kline) {
-
+	// klineSeries Update
+	intervals := s.store[exchange].ComputeIfAbsent(instId, func(k string) *collect.ConcurrentMap[types.Interval, *KlineSeries] {
+		return collect.NewConcurrentMap[types.Interval, *KlineSeries](8, func(t types.Interval) string { return string(t) })
+	})
+	klineSeries := intervals.ComputeIfAbsent(kline.Interval, func(interval types.Interval) *KlineSeries {
+		return NewKlineSeries(0, interval, s)
+	})
+	klineSeries.Update(kline)
 	return
 }

internal/trading/trading_grpc_server.go

@@ -0,0 +1,17 @@
+package trading
+
+import (
+	"sig-pub/api/pb"
+)
+
+type TradingGrpcServer struct {
+	pb.UnimplementedTradingServiceServer
+}
+
+func NewTradingGrpcServer() *TradingGrpcServer {
+	return &TradingGrpcServer{}
+}
+
+func (svr TradingGrpcServer) Init() (err error) {
+	return
+}

internal/trading/trading_service.go

@@ -0,0 +1,136 @@
+package trading
+
+import (
+	"context"
+	"io"
+	"sig-pub/api/pb"
+	"sig-pub/pkg/client"
+	"sig-pub/pkg/types"
+	"sig-pub/pkg/zlog"
+	"sync"
+	"time"
+
+	"google.golang.org/grpc"
+)
+
+type TradingService struct {
+	marketClientAside *client.TradeInstanceAside
+	exchangeClient    pb.ExchangeServiceClient
+
+	klineStore *KlineStore
+
+	subKlineLock   sync.Mutex
+	subKlineInsts  [3][]string
+	subKlineStream grpc.BidiStreamingClient[pb.ReqStreamSubscribeKline, pb.RspStreamSubscribeKline]
+}
+
+func NewTradingService(
+	marketClientAside *client.TradeInstanceAside,
+	exchangeClient pb.ExchangeServiceClient,
+) *TradingService {
+	return &TradingService{
+		marketClientAside: marketClientAside,
+		exchangeClient:    exchangeClient,
+	}
+}
+
+// 初始化历史k线, 订阅实时k线
+func (svr *TradingService) Init() (err error) {
+	// get instance
+	exchangeTradeInsts, err := svr.marketClientAside.ListExchangeTradeInstance(context.Background(), pb.ExchangeType_OKX)
+	if err != nil {
+		return
+	}
+	for _, exInst := range exchangeTradeInsts {
+		svr.subKlineInsts[exInst.Exchange] = append(svr.subKlineInsts[exInst.Exchange], exInst.InstId)
+	}
+
+	// 订阅k线
+	go svr.subscribeKlines(false)
+	return
+}
+
+func (svr *TradingService) subscribeKlines(reconnect bool) {
+	defer func() {
+		svr.subKlineStream = nil
+		go svr.subscribeKlines(true)
+	}()
+
+	if reconnect {
+		zlog.Infof("subscribeKlines will reconnect after 5s")
+		time.Sleep(5 * time.Second)
+	}
+
+	// svr.subKlineLock.Lock()
+	// defer svr.subKlineLock.Unlock()
+
+	stream, err := svr.exchangeClient.SubscribeKline(context.Background())
+	if err != nil {
+		zlog.Error("subscribeKlines reqeust error: ", err)
+		return
+	}
+
+	if svr.subKlineStream != nil {
+		svr.subKlineStream.CloseSend()
+	}
+
+	// 发送订阅消息
+	exchanges := []pb.ExchangeType{pb.ExchangeType_OKX}
+	for _, exchange := range exchanges {
+		instIds := svr.subKlineInsts[exchange]
+
+		msg := &pb.ReqStreamSubscribeKline{
+			SubType:     pb.SubscribeType_Subscribe,
+			Exchanges:   []pb.ExchangeType{exchange},
+			InstIds:     instIds,
+			Intervals:   SubscribeKlineIntervals,
+			OnlyConfirm: true,
+		}
+		zlog.Debugf("send stream subscribe kline msg: %#v", msg)
+		if err = stream.Send(msg); err != nil {
+			zlog.Errorf("send stream subscribe kline msg error: %v", msg, err)
+			return
+		}
+	}
+
+	// 接收消息的goroutine
+	for {
+		msg, err := stream.Recv()
+		if err == io.EOF {
+			zlog.Debugf("subscribeKlines connection server closeed")
+			return
+		}
+		if err != nil {
+			zlog.Error("subscribeKlines recv error: ", err)
+			return
+		}
+
+		for _, k := range msg.Kline.Klines {
+			kline := new(types.Kline)
+			kline.ParsePBKline(msg.Kline.Exchange, k)
+			zlog.Debugf("recv: streamId=%d, %v, %s, %#v", msg.Kline.StreamId, msg.Kline.Exchange, msg.Kline.InstId, kline)
+
+			svr.klineStore.Update(msg.Kline.Exchange, msg.Kline.InstId, kline)
+		}
+	}
+}
+
+var SubscribeKlineIntervals = []string{
+	string(types.Interval1s),
+	string(types.Interval1m),
+	string(types.Interval3m),
+	string(types.Interval5m),
+	string(types.Interval15m),
+	string(types.Interval30m),
+	string(types.Interval1h),
+	string(types.Interval2h),
+	string(types.Interval4h),
+	string(types.Interval6h),
+	string(types.Interval12h),
+	string(types.Interval1d),
+	string(types.Interval1d),
+	string(types.Interval2d),
+	string(types.Interval3d),
+	string(types.Interval5d),
+	string(types.Interval1w),
+}

pkg/client/market.go

@@ -0,0 +1,30 @@
+package client
+
+import (
+	"sig-pub/api/pb"
+	"sig-pub/pkg/grpc/discovery"
+
+	"google.golang.org/grpc"
+)
+
+// NewMarketClient new MarketServer grpc client
+func NewMarketClient(options ...grpc.DialOption) (marketClient pb.MarketServiceClient, err error) {
+	grpcUrl := discovery.ConsulDialUrl(pb.MarketService_ServiceDesc.ServiceName)
+	grpcConn, err := grpc.NewClient(grpcUrl, options...)
+	if err != nil {
+		panic(err)
+	}
+	marketClient = pb.NewMarketServiceClient(grpcConn)
+	return
+}
+
+// NewExchangeClient new ExchangeServer grpc client
+func NewExchangeClient(options ...grpc.DialOption) (exchangeClient pb.ExchangeServiceClient, err error) {
+	grpcUrl := discovery.ConsulDialUrl(pb.ExchangeService_ServiceDesc.ServiceName)
+	grpcConn, err := grpc.NewClient(grpcUrl, options...)
+	if err != nil {
+		panic(err)
+	}
+	exchangeClient = pb.NewExchangeServiceClient(grpcConn)
+	return
+}

pkg/client/trade_instance_client.go

@@ -1,4 +1,4 @@
-package aside
+package client
 
 import (
 	"context"

pkg/utils/collect/concurrent_map.go

@@ -170,8 +170,8 @@ func (m *ConcurrentMap[K, V]) Size() (size int) {
 
 // ComputeIfAbsent 加载, 如果值不存在使用 mapping(k) 填充并返回
 // mapped 值是否是 mapping(k) 填充的
-func (m *ConcurrentMap[K, V]) ComputeIfAbsent(k K, mapping func(k K) V) (res V, mapped bool) {
+func (m *ConcurrentMap[K, V]) ComputeIfAbsent(k K, mapping func(k K) V) (res V) {
-	res, _, mapped = m.ComputeIfAbsentE(k, func(k K) (V, error) {
+	res, _, _ = m.ComputeIfAbsentE(k, func(k K) (V, error) {
 		return mapping(k), nil
 	})
 	return

pkg/utils/collect/concurrent_map_test.go

@@ -82,12 +82,9 @@ func TestComputeIfAbsent(t *testing.T) {
 			defer wg.Done()
 
 			r := rand.New(rand.NewSource(time.Now().UnixMilli()))
-			v, mapped := cm.ComputeIfAbsent(r.Intn(10), func(k int) *counter {
+			v := cm.ComputeIfAbsent(r.Intn(10), func(k int) *counter {
 				return &counter{}
 			})
-			if !mapped {
-				return
-			}
 			for i := 0; i < add; i++ {
 				v.increment()
 			}