mirror of
https://github.com/c9s/bbgo.git
synced 2024-11-23 07:15:15 +00:00
123 lines
2.7 KiB
Go
123 lines
2.7 KiB
Go
package ftx
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/gorilla/websocket"
|
|
|
|
"github.com/c9s/bbgo/pkg/service"
|
|
"github.com/c9s/bbgo/pkg/types"
|
|
)
|
|
|
|
const endpoint = "wss://ftx.com/ws/"
|
|
|
|
type Stream struct {
|
|
*types.StandardStream
|
|
|
|
ws *service.WebsocketClientBase
|
|
|
|
// publicOnly can only be configured before connecting
|
|
publicOnly int32
|
|
|
|
key string
|
|
secret string
|
|
|
|
// subscriptions are only accessed in single goroutine environment, so I don't use mutex to protect them
|
|
subscriptions []websocketRequest
|
|
}
|
|
|
|
func NewStream(key, secret string) *Stream {
|
|
s := &Stream{
|
|
key: key,
|
|
secret: secret,
|
|
StandardStream: &types.StandardStream{},
|
|
ws: service.NewWebsocketClientBase(endpoint, 3*time.Second),
|
|
}
|
|
|
|
s.ws.OnMessage((&messageHandler{StandardStream: s.StandardStream}).handleMessage)
|
|
s.ws.OnConnected(func(conn *websocket.Conn) {
|
|
subs := []websocketRequest{newLoginRequest(s.key, s.secret, time.Now())}
|
|
subs = append(subs, s.subscriptions...)
|
|
for _, sub := range subs {
|
|
if err := conn.WriteJSON(sub); err != nil {
|
|
s.ws.EmitError(fmt.Errorf("failed to send subscription: %+v", sub))
|
|
}
|
|
}
|
|
})
|
|
|
|
return s
|
|
}
|
|
|
|
func (s *Stream) Connect(ctx context.Context) error {
|
|
// If it's not public only, let's do the authentication.
|
|
if atomic.LoadInt32(&s.publicOnly) == 0 {
|
|
s.subscribePrivateEvents()
|
|
}
|
|
|
|
if err := s.ws.Connect(ctx); err != nil {
|
|
return err
|
|
}
|
|
|
|
go func() {
|
|
// https://docs.ftx.com/?javascript#request-process
|
|
tk := time.NewTicker(15 * time.Second)
|
|
defer tk.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
if err := ctx.Err(); err != nil {
|
|
logger.WithError(err).Errorf("websocket ping goroutine is terminated")
|
|
}
|
|
case <-tk.C:
|
|
if err := s.ws.Conn().WriteJSON(websocketRequest{
|
|
Operation: ping,
|
|
}); err != nil {
|
|
logger.WithError(err).Warnf("failed to ping, try in next tick")
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
func (s *Stream) subscribePrivateEvents() {
|
|
s.addSubscription(websocketRequest{
|
|
Operation: subscribe,
|
|
Channel: privateOrdersChannel,
|
|
})
|
|
s.addSubscription(websocketRequest{
|
|
Operation: subscribe,
|
|
Channel: privateTradesChannel,
|
|
})
|
|
}
|
|
|
|
func (s *Stream) addSubscription(request websocketRequest) {
|
|
s.subscriptions = append(s.subscriptions, request)
|
|
}
|
|
|
|
func (s *Stream) SetPublicOnly() {
|
|
atomic.StoreInt32(&s.publicOnly, 1)
|
|
}
|
|
|
|
func (s *Stream) Subscribe(channel types.Channel, symbol string, _ types.SubscribeOptions) {
|
|
if channel != types.BookChannel {
|
|
panic("only support book channel now")
|
|
}
|
|
s.addSubscription(websocketRequest{
|
|
Operation: subscribe,
|
|
Channel: orderBookChannel,
|
|
Market: TrimUpperString(symbol),
|
|
})
|
|
}
|
|
|
|
func (s *Stream) Close() error {
|
|
s.subscriptions = nil
|
|
if s.ws != nil {
|
|
return s.ws.Conn().Close()
|
|
}
|
|
return nil
|
|
}
|