lychiyu/bbgo_origin
1
0
Fork 0
mirror of https://github.com/c9s/bbgo.git synced 2024-09-20 08:11:08 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

grpc: separate market data message and user data message

Browse Source
This commit is contained in:
c9s 2022-04-13 14:14:25 +08:00
parent ec6a94e485
commit 8e81716d2a
5 changed files with 768 additions and 645 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
pkg/grpc/convert.go
View File

@ -49,8 +49,8 @@ func transPriceVolume(srcPvs types.PriceVolumeSlice) (pvs []*pb.PriceVolume) {
return pvs
}
func transBook(session *bbgo.ExchangeSession, book types.SliceOrderBook, event pb.Event) *pb.SubscribeResponse {
return &pb.SubscribeResponse{
func transBook(session *bbgo.ExchangeSession, book types.SliceOrderBook, event pb.Event) *pb.MarketData {
return &pb.MarketData{
Session: session.Name,
Exchange: session.ExchangeName.String(),
Symbol: book.Symbol,
@ -65,8 +65,8 @@ func transBook(session *bbgo.ExchangeSession, book types.SliceOrderBook, event p
}
}
func transMarketTrade(session *bbgo.ExchangeSession, marketTrade types.Trade) *pb.SubscribeResponse {
return &pb.SubscribeResponse{
func transMarketTrade(session *bbgo.ExchangeSession, marketTrade types.Trade) *pb.MarketData {
return &pb.MarketData{
Session: session.Name,
Exchange: session.ExchangeName.String(),
Symbol: marketTrade.Symbol,
@ -100,27 +100,31 @@ func transSide(side types.SideType) pb.Side {
return pb.Side_SELL
}
func transKLine(session *bbgo.ExchangeSession, kline types.KLine) *pb.SubscribeResponse {
return &pb.SubscribeResponse{
Session: session.Name,
Exchange: kline.Exchange.String(),
Symbol: kline.Symbol,
Channel: pb.Channel_KLINE,
Event: pb.Event_UPDATE,
Kline: &pb.KLine{
Session: session.Name,
Exchange: kline.Exchange.String(),
Symbol: kline.Symbol,
Open: kline.Open.String(),
High: kline.High.String(),
Low: kline.Low.String(),
Close: kline.Close.String(),
Volume: kline.Volume.String(),
QuoteVolume: kline.QuoteVolume.String(),
StartTime: kline.StartTime.UnixMilli(),
EndTime: kline.StartTime.UnixMilli(),
Closed: kline.Closed,
},
func transKLine(session *bbgo.ExchangeSession, kline types.KLine) *pb.KLine {
return &pb.KLine{
Session: session.Name,
Exchange: kline.Exchange.String(),
Symbol: kline.Symbol,
Open: kline.Open.String(),
High: kline.High.String(),
Low: kline.Low.String(),
Close: kline.Close.String(),
Volume: kline.Volume.String(),
QuoteVolume: kline.QuoteVolume.String(),
StartTime: kline.StartTime.UnixMilli(),
EndTime: kline.StartTime.UnixMilli(),
Closed: kline.Closed,
}
}
func transKLineResponse(session *bbgo.ExchangeSession, kline types.KLine) *pb.MarketData {
return &pb.MarketData{
Session: session.Name,
Exchange: kline.Exchange.String(),
Symbol: kline.Symbol,
Channel: pb.Channel_KLINE,
Event: pb.Event_UPDATE,
Kline: transKLine(session, kline),
SubscribedAt: 0,
}
}

21
pkg/grpc/server.go
View File

@ -22,6 +22,11 @@ type Server struct {
Trader *bbgo.Trader
pb.UnimplementedMarketDataServiceServer
pb.UnimplementedUserDataServiceServer
}
func (s *Server) SubscribeUserData(empty *pb.Empty, server pb.UserDataService_SubscribeServer) error {
return nil
}
func (s *Server) Subscribe(request *pb.SubscribeRequest, server pb.MarketDataService_SubscribeServer) error {
@ -73,7 +78,7 @@ func (s *Server) Subscribe(request *pb.SubscribeRequest, server pb.MarketDataSer
}
})
stream.OnKLineClosed(func(kline types.KLine) {
err := server.Send(transKLine(session, kline))
err := server.Send(transKLineResponse(session, kline))
if err != nil {
log.WithError(err).Error("grpc stream send error")
}
@ -132,19 +137,7 @@ func (s *Server) QueryKLines(ctx context.Context, request *pb.QueryKLinesRequest
}
for _, kline := range klines {
response.Klines = append(response.Klines, &pb.KLine{
Exchange: kline.Exchange.String(),
Symbol: kline.Symbol,
Open: kline.Open.String(),
High: kline.High.String(),
Low: kline.Low.String(),
Close: kline.Close.String(),
Volume: kline.Volume.String(),
QuoteVolume: kline.QuoteVolume.String(),
Closed: kline.Closed,
StartTime: kline.StartTime.UnixMilli(),
EndTime: kline.EndTime.UnixMilli(),
})
response.Klines = append(response.Klines, transKLine(session, kline))
}
return response, nil

1245
pkg/pb/bbgo.pb.go
View File

File diff suppressed because it is too large Load Diff

39
pkg/pb/bbgo.proto
View File

@ -5,13 +5,12 @@ package bbgo;
option go_package = "../pb";
service MarketDataService {
rpc Subscribe(SubscribeRequest) returns (stream SubscribeResponse) {}
rpc Subscribe(SubscribeRequest) returns (stream MarketData) {}
rpc QueryKLines(QueryKLinesRequest) returns (QueryKLinesResponse) {}
}
service UserDataService {
// should support streaming
rpc SubscribeUserData(Empty) returns (stream SubscribeResponse) {}
rpc Subscribe(UserDataRequest) returns (stream UserData) {}
}
service TradingService {
@ -30,12 +29,6 @@ enum Event {
SNAPSHOT = 3;
UPDATE = 4;
AUTHENTICATED = 5;
ORDER_SNAPSHOT = 6;
ORDER_UPDATE = 7;
TRADE_SNAPSHOT = 8;
TRADE_UPDATE = 9;
ACCOUNT_SNAPSHOT = 10;
ACCOUNT_UPDATE = 11;
ERROR = 99;
}
@ -43,8 +36,8 @@ enum Channel {
BOOK = 0;
TRADE = 1;
TICKER = 2;
USER = 3;
KLINE = 4;
KLINE = 3;
BALANCE = 4;
}
enum Side {
@ -68,6 +61,20 @@ message Error {
string error_message = 2;
}
message UserDataRequest {
string session = 1;
}
message UserData {
string session = 1;
string exchange = 2;
Channel channel = 3; // trade, order, balance
Event event = 4; // snapshot, update ...
repeated Balance balances = 5;
repeated Trade trades = 6;
repeated Order orders = 7;
}
message SubscribeRequest {
repeated Subscription subscriptions = 1;
}
@ -80,18 +87,16 @@ message Subscription {
string interval = 5; // interval is for kline channel
}
message SubscribeResponse {
message MarketData {
string session = 1;
string exchange = 2;
string symbol = 3;
Channel channel = 4; // book, trade, ticker, user
Event event = 5; // snapshot, update, order_snapshot, ...
Event event = 5; // snapshot or update
Depth depth = 6; // depth: used by book
KLine kline = 7;
repeated Trade trades = 8;
Ticker ticker = 9;
repeated Order orders = 10;
repeated Balance balances = 11;
Ticker ticker = 9; // market ticker
repeated Trade trades = 8; // market trades
int64 subscribed_at = 12;
Error error = 13;
}

54
pkg/pb/bbgo_grpc.pb.go
View File

@ -46,7 +46,7 @@ func (c *marketDataServiceClient) Subscribe(ctx context.Context, in *SubscribeRe
}
type MarketDataService_SubscribeClient interface {
Recv() (*SubscribeResponse, error)
Recv() (*MarketData, error)
grpc.ClientStream
}
@ -54,8 +54,8 @@ type marketDataServiceSubscribeClient struct {
grpc.ClientStream
}
func (x *marketDataServiceSubscribeClient) Recv() (*SubscribeResponse, error) {
m := new(SubscribeResponse)
func (x *marketDataServiceSubscribeClient) Recv() (*MarketData, error) {
m := new(MarketData)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
@ -112,7 +112,7 @@ func _MarketDataService_Subscribe_Handler(srv interface{}, stream grpc.ServerStr
}
type MarketDataService_SubscribeServer interface {
Send(*SubscribeResponse) error
Send(*MarketData) error
grpc.ServerStream
}
@ -120,7 +120,7 @@ type marketDataServiceSubscribeServer struct {
grpc.ServerStream
}
func (x *marketDataServiceSubscribeServer) Send(m *SubscribeResponse) error {
func (x *marketDataServiceSubscribeServer) Send(m *MarketData) error {
return x.ServerStream.SendMsg(m)
}
@ -168,8 +168,7 @@ var MarketDataService_ServiceDesc = grpc.ServiceDesc{
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
type UserDataServiceClient interface {
// should support streaming
SubscribeUserData(ctx context.Context, in *Empty, opts ...grpc.CallOption) (UserDataService_SubscribeUserDataClient, error)
Subscribe(ctx context.Context, in *UserDataRequest, opts ...grpc.CallOption) (UserDataService_SubscribeClient, error)
}
type userDataServiceClient struct {
@ -180,12 +179,12 @@ func NewUserDataServiceClient(cc grpc.ClientConnInterface) UserDataServiceClient
return &userDataServiceClient{cc}
}
func (c *userDataServiceClient) SubscribeUserData(ctx context.Context, in *Empty, opts ...grpc.CallOption) (UserDataService_SubscribeUserDataClient, error) {
stream, err := c.cc.NewStream(ctx, &UserDataService_ServiceDesc.Streams[0], "/bbgo.UserDataService/SubscribeUserData", opts...)
func (c *userDataServiceClient) Subscribe(ctx context.Context, in *UserDataRequest, opts ...grpc.CallOption) (UserDataService_SubscribeClient, error) {
stream, err := c.cc.NewStream(ctx, &UserDataService_ServiceDesc.Streams[0], "/bbgo.UserDataService/Subscribe", opts...)
if err != nil {
return nil, err
}
x := &userDataServiceSubscribeUserDataClient{stream}
x := &userDataServiceSubscribeClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
@ -195,17 +194,17 @@ func (c *userDataServiceClient) SubscribeUserData(ctx context.Context, in *Empty
return x, nil
}
type UserDataService_SubscribeUserDataClient interface {
Recv() (*SubscribeResponse, error)
type UserDataService_SubscribeClient interface {
Recv() (*UserData, error)
grpc.ClientStream
}
type userDataServiceSubscribeUserDataClient struct {
type userDataServiceSubscribeClient struct {
grpc.ClientStream
}
func (x *userDataServiceSubscribeUserDataClient) Recv() (*SubscribeResponse, error) {
m := new(SubscribeResponse)
func (x *userDataServiceSubscribeClient) Recv() (*UserData, error) {
m := new(UserData)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
@ -216,8 +215,7 @@ func (x *userDataServiceSubscribeUserDataClient) Recv() (*SubscribeResponse, err
// All implementations must embed UnimplementedUserDataServiceServer
// for forward compatibility
type UserDataServiceServer interface {
// should support streaming
SubscribeUserData(*Empty, UserDataService_SubscribeUserDataServer) error
Subscribe(*UserDataRequest, UserDataService_SubscribeServer) error
mustEmbedUnimplementedUserDataServiceServer()
}
@ -225,8 +223,8 @@ type UserDataServiceServer interface {
type UnimplementedUserDataServiceServer struct {
}
func (UnimplementedUserDataServiceServer) SubscribeUserData(*Empty, UserDataService_SubscribeUserDataServer) error {
return status.Errorf(codes.Unimplemented, "method SubscribeUserData not implemented")
func (UnimplementedUserDataServiceServer) Subscribe(*UserDataRequest, UserDataService_SubscribeServer) error {
return status.Errorf(codes.Unimplemented, "method Subscribe not implemented")
}
func (UnimplementedUserDataServiceServer) mustEmbedUnimplementedUserDataServiceServer() {}
@ -241,24 +239,24 @@ func RegisterUserDataServiceServer(s grpc.ServiceRegistrar, srv UserDataServiceS
s.RegisterService(&UserDataService_ServiceDesc, srv)
}
func _UserDataService_SubscribeUserData_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(Empty)
func _UserDataService_Subscribe_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(UserDataRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(UserDataServiceServer).SubscribeUserData(m, &userDataServiceSubscribeUserDataServer{stream})
return srv.(UserDataServiceServer).Subscribe(m, &userDataServiceSubscribeServer{stream})
}
type UserDataService_SubscribeUserDataServer interface {
Send(*SubscribeResponse) error
type UserDataService_SubscribeServer interface {
Send(*UserData) error
grpc.ServerStream
}
type userDataServiceSubscribeUserDataServer struct {
type userDataServiceSubscribeServer struct {
grpc.ServerStream
}
func (x *userDataServiceSubscribeUserDataServer) Send(m *SubscribeResponse) error {
func (x *userDataServiceSubscribeServer) Send(m *UserData) error {
return x.ServerStream.SendMsg(m)
}
@ -271,8 +269,8 @@ var UserDataService_ServiceDesc = grpc.ServiceDesc{
Methods: []grpc.MethodDesc{},
Streams: []grpc.StreamDesc{
{
StreamName: "SubscribeUserData",
Handler: _UserDataService_SubscribeUserData_Handler,
StreamName: "Subscribe",
Handler: _UserDataService_Subscribe_Handler,
ServerStreams: true,
},
},