95 lines
2.2 KiB
Go
95 lines
2.2 KiB
Go
|
package okex
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
|
||
|
|
"git.qtrade.icu/lychiyu/qbtrade/pkg/exchange/okex/okexapi"
|
||
|
|
"git.qtrade.icu/lychiyu/qbtrade/pkg/types"
|
||
|
|
)
|
||
|
|
|
||
|
|
//go:generate callbackgen -type KLineStream -interface
|
||
|
|
type KLineStream struct {
|
||
|
|
types.StandardStream
|
||
|
|
|
||
|
|
kLineEventCallbacks []func(candle KLineEvent)
|
||
|
|
}
|
||
|
|
|
||
|
|
func NewKLineStream() *KLineStream {
|
||
|
|
k := &KLineStream{
|
||
|
|
StandardStream: types.NewStandardStream(),
|
||
|
|
}
|
||
|
|
|
||
|
|
k.SetParser(parseWebSocketEvent)
|
||
|
|
k.SetDispatcher(k.dispatchEvent)
|
||
|
|
k.SetEndpointCreator(func(_ context.Context) (string, error) { return okexapi.PublicBusinessWebSocketURL, nil })
|
||
|
|
k.SetPingInterval(pingInterval)
|
||
|
|
|
||
|
|
// K line channel is public only
|
||
|
|
k.SetPublicOnly()
|
||
|
|
k.OnConnect(k.handleConnect)
|
||
|
|
k.OnKLineEvent(k.handleKLineEvent)
|
||
|
|
|
||
|
|
return k
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *KLineStream) handleConnect() {
|
||
|
|
var subs []WebsocketSubscription
|
||
|
|
for _, subscription := range s.Subscriptions {
|
||
|
|
if subscription.Channel != types.KLineChannel {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
|
||
|
|
sub, err := convertSubscription(subscription)
|
||
|
|
if err != nil {
|
||
|
|
log.WithError(err).Errorf("subscription convert error")
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
|
||
|
|
subs = append(subs, sub)
|
||
|
|
}
|
||
|
|
subscribe(s.Conn, subs)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *KLineStream) Connect(ctx context.Context) error {
|
||
|
|
if len(s.StandardStream.Subscriptions) == 0 {
|
||
|
|
log.Info("no subscriptions in kline")
|
||
|
|
return nil
|
||
|
|
}
|
||
|
|
return s.StandardStream.Connect(ctx)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *KLineStream) handleKLineEvent(k KLineEvent) {
|
||
|
|
for _, event := range k.Events {
|
||
|
|
kline := kLineToGlobal(event, types.Interval(k.Interval), k.Symbol)
|
||
|
|
if kline.Closed {
|
||
|
|
s.EmitKLineClosed(kline)
|
||
|
|
} else {
|
||
|
|
s.EmitKLine(kline)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *KLineStream) dispatchEvent(e interface{}) {
|
||
|
|
switch et := e.(type) {
|
||
|
|
case *WebSocketEvent:
|
||
|
|
if err := et.IsValid(); err != nil {
|
||
|
|
log.Errorf("invalid event: %v", err)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
case *KLineEvent:
|
||
|
|
s.EmitKLineEvent(*et)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *KLineStream) Unsubscribe() {
|
||
|
|
// errors are handled in the syncSubscriptions, so they are skipped here.
|
||
|
|
if len(s.StandardStream.Subscriptions) != 0 {
|
||
|
|
_ = syncSubscriptions(s.StandardStream.Conn, s.StandardStream.Subscriptions, WsEventTypeUnsubscribe)
|
||
|
|
}
|
||
|
|
s.Resubscribe(func(old []types.Subscription) (new []types.Subscription, err error) {
|
||
|
|
// clear the subscriptions
|
||
|
|
return []types.Subscription{}, nil
|
||
|
|
})
|
||
|
|
}
|