ftx: handle message in a new struct

pkg/exchange/ftx/stream.go

@@ -23,9 +23,7 @@ func NewStream(key, secret string) *Stream {
 		wsService:      wss,
 	}
 
-	wss.OnMessage(func(message []byte) {
-		logger.Infof("=> message: %+v", string(message))
-	})
+	wss.OnMessage(messageHandler{s.StandardStream}.handleMessage)
 	return s
 }
 
@@ -37,44 +35,11 @@ func (s *Stream) SetPublicOnly() {
 	atomic.StoreInt32(&s.publicOnly, 1)
 }
 
-func (s *Stream) Close() error {
-	return s.wsService.Close()
-}
-
 func (s *Stream) Subscribe(channel types.Channel, symbol string, _ types.SubscribeOptions) {
 	if err := s.wsService.Subscribe(channel, symbol); err != nil {
 		logger.WithError(err).Errorf("subscribe failed, should never happen")
 	}
 }
-
-func (s *Stream) OnTradeUpdate(cb func(trade types.Trade)) {
-	panic("implement me")
-}
-
-func (s *Stream) OnOrderUpdate(cb func(order types.Order)) {
-	panic("implement me")
-}
-
-func (s *Stream) OnBalanceSnapshot(cb func(balances types.BalanceMap)) {
-	panic("implement me")
-}
-
-func (s *Stream) OnBalanceUpdate(cb func(balances types.BalanceMap)) {
-	panic("implement me")
-}
-
-func (s *Stream) OnKLineClosed(cb func(kline types.KLine)) {
-	panic("implement me")
-}
-
-func (s *Stream) OnKLine(cb func(kline types.KLine)) {
-	panic("implement me")
-}
-
-func (s *Stream) OnBookUpdate(cb func(book types.OrderBook)) {
-	panic("implement me")
-}
-
-func (s *Stream) OnBookSnapshot(cb func(book types.OrderBook)) {
-	panic("implement me")
+func (s *Stream) Close() error {
+	return s.wsService.Close()
 }

pkg/exchange/ftx/stream_message_handler.go

@@ -0,0 +1,31 @@
+package ftx
+
+import (
+	"encoding/json"
+
+	"github.com/c9s/bbgo/pkg/types"
+)
+
+type messageHandler struct {
+	types.StandardStream
+}
+
+func (h messageHandler) handleMessage(message []byte) {
+	var r rawResponse
+	if err := json.Unmarshal(message, &r); err != nil {
+		logger.WithError(err).Errorf("failed to unmarshal resp: %s", string(message))
+		return
+	}
+
+	switch r.Type {
+	case subscribedRespType:
+		h.handleSubscribedMessage(r)
+	default:
+		logger.Errorf("unsupported message type: %+v", r.Type)
+	}
+}
+
+// {"type": "subscribed", "channel": "orderbook", "market": "BTC/USDT"}
+func (h messageHandler) handleSubscribedMessage(response rawResponse) {
+	logger.Infof("%s orderbook is subscribed", response.toSubscribedResp().Market)
+}