bbgo_origin/pkg/exchange/okex/stream.go

package okex

import (
	"context"
	"net"
	"strconv"
	"sync"
	"time"

	"github.com/c9s/bbgo/pkg/exchange/okex/okexapi"
	"github.com/c9s/bbgo/pkg/types"
	"github.com/gorilla/websocket"
)

const readTimeout = 15 * time.Second

type WebsocketOp struct {
	Op   string      `json:"op"`
	Args interface{} `json:"args"`
}

type WebsocketLogin struct {
	Key        string `json:"apiKey"`
	Passphrase string `json:"passphrase"`
	Timestamp  string `json:"timestamp"`
	Sign       string `json:"sign"`
}

//go:generate callbackgen -type Stream -interface
type Stream struct {
	types.StandardStream

	Client     *okexapi.RestClient
	Conn       *websocket.Conn
	connLock   sync.Mutex
	connCtx    context.Context
	connCancel context.CancelFunc

	// public callbacks
	candleDataCallbacks   []func(candle Candle)
	bookDataCallbacks     []func(book BookData)
	eventCallbacks        []func(event WebSocketEvent)
	accountCallbacks      []func(account okexapi.Account)
	orderDetailsCallbacks []func(orderDetails []okexapi.OrderDetails)

	lastCandle map[CandleKey]Candle
}

type CandleKey struct {
	InstrumentID string
	Channel      string
}

func NewStream(client *okexapi.RestClient) *Stream {
	stream := &Stream{
		Client: client,
		StandardStream: types.StandardStream{
			ReconnectC: make(chan struct{}, 1),
		},
		lastCandle: make(map[CandleKey]Candle),
	}

	stream.OnCandleData(func(candle Candle) {
		key := CandleKey{Channel: candle.Channel, InstrumentID: candle.InstrumentID}
		kline := candle.KLine()

		// check if we need to close previous kline
		lastCandle, ok := stream.lastCandle[key]
		if ok && candle.StartTime.After(lastCandle.StartTime) {
			lastKline := lastCandle.KLine()
			lastKline.Closed = true
			stream.EmitKLineClosed(lastKline)
		}

		stream.EmitKLine(kline)
		stream.lastCandle[key] = candle
	})

	stream.OnBookData(func(data BookData) {
		book := data.Book()
		switch data.Action {
		case "snapshot":
			stream.EmitBookSnapshot(book)
		case "update":
			stream.EmitBookUpdate(book)
		}
	})

	stream.OnAccount(func(account okexapi.Account) {
		balances := toGlobalBalance(&account)
		stream.EmitBalanceSnapshot(balances)
	})

	stream.OnOrderDetails(func(orderDetails []okexapi.OrderDetails) {
		detailTrades, detailOrders := segmentOrderDetails(orderDetails)

		trades, err := toGlobalTrades(detailTrades)
		if err != nil {
			log.WithError(err).Errorf("error converting order details into trades")
		} else {
			for _, trade := range trades {
				stream.EmitTradeUpdate(trade)
			}
		}

		orders, err := toGlobalOrders(detailOrders)
		if err != nil {
			log.WithError(err).Errorf("error converting order details into orders")
		} else {
			for _, order := range orders {
				stream.EmitOrderUpdate(order)
			}
		}
	})

	stream.OnEvent(func(event WebSocketEvent) {
		switch event.Event {
		case "login":
			if event.Code == "0" {
				var subs = []WebsocketSubscription{
					{Channel: "account"},
					{Channel: "orders", InstrumentType: string(okexapi.InstrumentTypeSpot)},
				}

				log.Infof("subscribing private channels: %+v", subs)
				err := stream.Conn.WriteJSON(WebsocketOp{
					Op:   "subscribe",
					Args: subs,
				})

				if err != nil {
					log.WithError(err).Error("private channel subscribe error")
				}
			}
		}
	})

	stream.OnConnect(func() {
		if stream.PublicOnly {
			var subs []WebsocketSubscription
			for _, subscription := range stream.Subscriptions {
				sub, err := convertSubscription(subscription)
				if err != nil {
					log.WithError(err).Errorf("subscription convert error")
					continue
				}

				subs = append(subs, sub)
			}
			if len(subs) == 0 {
				return
			}

			log.Infof("subscribing channels: %+v", subs)
			err := stream.Conn.WriteJSON(WebsocketOp{
				Op:   "subscribe",
				Args: subs,
			})

			if err != nil {
				log.WithError(err).Error("subscribe error")
			}
		} else {
			// login as private channel
			// sign example:
			// sign=CryptoJS.enc.Base64.Stringify(CryptoJS.HmacSHA256(timestamp +'GET'+'/users/self/verify', secretKey))
			msTimestamp := strconv.FormatFloat(float64(time.Now().UnixNano())/float64(time.Second), 'f', -1, 64)
			payload := msTimestamp + "GET" + "/users/self/verify"
			sign := okexapi.Sign(payload, stream.Client.Secret)
			op := WebsocketOp{
				Op: "login",
				Args: []WebsocketLogin{
					{
						Key:        stream.Client.Key,
						Passphrase: stream.Client.Passphrase,
						Timestamp:  msTimestamp,
						Sign:       sign,
					},
				},
			}

			log.Infof("sending okex login request")
			err := stream.Conn.WriteJSON(op)
			if err != nil {
				log.WithError(err).Errorf("can not send login message")
			}
		}
	})

	return stream
}

func (s *Stream) Close() error {
	return nil
}

func (s *Stream) Connect(ctx context.Context) error {
	err := s.connect(ctx)
	if err != nil {
		return err
	}

	// start one re-connector goroutine with the base context
	go s.Reconnector(ctx)

	s.EmitStart()
	return nil
}

func (s *Stream) Reconnector(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			return

		case <-s.ReconnectC:
			log.Warnf("received reconnect signal, reconnecting...")
			time.Sleep(3 * time.Second)

			if err := s.connect(ctx); err != nil {
				log.WithError(err).Errorf("connect error, try to reconnect again...")
				s.Reconnect()
			}
		}
	}
}

func (s *Stream) connect(ctx context.Context) error {
	// when in public mode, the listen key is an empty string
	var url string
	if s.PublicOnly {
		url = okexapi.PublicWebSocketURL
	} else {
		url = okexapi.PrivateWebSocketURL
	}

	conn, err := s.StandardStream.Dial(url)
	if err != nil {
		return err
	}

	log.Infof("websocket connected: %s", url)

	// should only start one connection one time, so we lock the mutex
	s.connLock.Lock()

	// ensure the previous context is cancelled
	if s.connCancel != nil {
		s.connCancel()
	}

	// create a new context
	s.connCtx, s.connCancel = context.WithCancel(ctx)

	conn.SetReadDeadline(time.Now().Add(readTimeout))
	conn.SetPongHandler(func(string) error {
		conn.SetReadDeadline(time.Now().Add(readTimeout))
		return nil
	})

	s.Conn = conn
	s.connLock.Unlock()

	s.EmitConnect()

	go s.read(s.connCtx)
	go s.ping(s.connCtx)
	return nil
}

func (s *Stream) read(ctx context.Context) {
	defer func() {
		if s.connCancel != nil {
			s.connCancel()
		}
		s.EmitDisconnect()
	}()

	for {
		select {

		case <-ctx.Done():
			return

		default:
			s.connLock.Lock()
			conn := s.Conn
			s.connLock.Unlock()

			if err := conn.SetReadDeadline(time.Now().Add(readTimeout)); err != nil {
				log.WithError(err).Errorf("set read deadline error: %s", err.Error())
			}

			mt, message, err := conn.ReadMessage()
			if err != nil {
				// if it's a network timeout error, we should re-connect
				switch err := err.(type) {

				// if it's a websocket related error
				case *websocket.CloseError:
					if err.Code == websocket.CloseNormalClosure {
						return
					}

					// for unexpected close error, we should re-connect
					// emit reconnect to start a new connection
					s.Reconnect()
					return

				case net.Error:
					log.WithError(err).Error("network error")
					s.Reconnect()
					return

				default:
					log.WithError(err).Error("unexpected connection error")
					s.Reconnect()
					return
				}
			}

			// skip non-text messages
			if mt != websocket.TextMessage {
				continue
			}

			e, err := Parse(string(message))
			if err != nil {
				log.WithError(err).Error("message parse error")
			}

			if e != nil {
				switch et := e.(type) {
				case *WebSocketEvent:
					s.EmitEvent(*et)

				case *BookData:
					// there's "books" for 400 depth and books5 for 5 depth
					if et.channel != "books5" {
						s.EmitBookData(*et)
					}
					s.EmitBookTickerUpdate(et.BookTicker())
				case *Candle:
					s.EmitCandleData(*et)

				case *okexapi.Account:
					s.EmitAccount(*et)

				case []okexapi.OrderDetails:
					s.EmitOrderDetails(et)

				}
			}
		}
	}
}

func (s *Stream) ping(ctx context.Context) {
	pingTicker := time.NewTicker(readTimeout / 2)
	defer pingTicker.Stop()

	for {
		select {

		case <-ctx.Done():
			log.Debug("ping worker stopped")
			return

		case <-pingTicker.C:
			s.connLock.Lock()
			conn := s.Conn
			s.connLock.Unlock()

			if err := conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(3*time.Second)); err != nil {
				log.WithError(err).Error("ping error", err)
				s.Reconnect()
			}
		}
	}
}