ftx: send subscriptions when connected

pkg/exchange/ftx/stream.go

@@ -22,6 +22,10 @@ func NewStream(key, secret string) *Stream {
 		StandardStream: types.StandardStream{},
 		wsService:      wss,
 	}
+
+	wss.OnMessage(func(message []byte) {
+		logger.Infof("=> message: %+v", string(message))
+	})
 	return s
 }
 
@@ -37,8 +41,10 @@ func (s *Stream) Close() error {
 	return s.wsService.Close()
 }
 
-func (s *Stream) Subscribe(channel types.Channel, symbol string, options types.SubscribeOptions) {
-	panic("implement me")
+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)) {

pkg/exchange/ftx/websocket.go

@@ -1,9 +1,14 @@
 package ftx
 
 import (
+	"fmt"
+	"strings"
 	"time"
 
+	"github.com/gorilla/websocket"
+
 	"github.com/c9s/bbgo/pkg/service"
+	"github.com/c9s/bbgo/pkg/types"
 )
 
 type WebsocketService struct {
@@ -11,6 +16,8 @@ type WebsocketService struct {
 
 	key    string
 	secret string
+
+	subscriptions []SubscribeRequest
 }
 
 const endpoint = "wss://ftx.com/ws/"
@@ -21,9 +28,40 @@ func NewWebsocketService(key string, secret string) *WebsocketService {
 		key:                 key,
 		secret:              secret,
 	}
+	s.OnConnected(func(_ *websocket.Conn) {
+		if err := s.sendSubscriptions(); err != nil {
+			s.EmitError(err)
+		}
+	})
 	return s
 }
 
+func (w *WebsocketService) Subscribe(channel types.Channel, symbol string) error {
+	r := SubscribeRequest{
+		Operation: subscribe,
+	}
+	if channel != types.BookChannel {
+		return fmt.Errorf("unsupported channel %+v", channel)
+	}
+	r.Channel = orderbook
+	r.Market = strings.ToUpper(strings.TrimSpace(symbol))
+
+	w.subscriptions = append(w.subscriptions, r)
+	return nil
+}
+
+var errSubscriptionFailed = fmt.Errorf("failed to subscribe")
+
+func (w *WebsocketService) sendSubscriptions() error {
+	conn := w.Conn()
+	for _, s := range w.subscriptions {
+		if err := conn.WriteJSON(s); err != nil {
+			return fmt.Errorf("can't send subscription request %+v: %w", s, errSubscriptionFailed)
+		}
+	}
+	return nil
+}
+
 func (w *WebsocketService) Close() error {
 	return w.Conn().Close()
 }

pkg/exchange/ftx/websocket_messages.go

@@ -0,0 +1,19 @@
+package ftx
+
+type operation string
+
+const subscribe operation = "subscribe"
+const unsubscribe operation = "unsubscribe"
+
+type channel string
+
+const orderbook channel = "orderbook"
+const trades channel = "trades"
+const ticker channel = "ticker"
+
+// {'op': 'subscribe', 'channel': 'trades', 'market': 'BTC-PERP'}
+type SubscribeRequest struct {
+	Operation operation `json:"op"`
+	Channel   channel   `json:"channel"`
+	Market    string    `json:"market"`
+}