GateIO: Abstract WS handler func routing

exchanges/gateio/gateio_types.go

@@ -2001,7 +2001,6 @@ type WSResponse struct {
 	Event     string          `json:"event"`
 	Result    json.RawMessage `json:"result"`
 	RequestID string          `json:"request_id"`
-	assetType asset.Item
 }
 
 // WsTicker websocket ticker information.

exchanges/gateio/gateio_websocket.go

@@ -30,8 +30,9 @@ import (
 )
 
 var (
-	errParsingWSField = errors.New("error parsing WS field")
-	errChannelParts   = errors.New("should contain asset and channel")
+	errParsingWSField      = errors.New("error parsing WS field")
+	errChannelParts        = errors.New("should contain asset and channel")
+	errMsgAssetTypeInvalid = errors.New("message AssetType prefix does not match connection AssetType")
 )
 
 const (
@@ -160,7 +161,7 @@ func (g *Gateio) generateWsSignature(secret, event, channel string, t int64) (st
 }
 
 // wsHandleData handles websocket data
-func (g *Gateio) wsHandleData(ctx context.Context, respRaw []byte) error {
+func (g *Gateio) wsHandleData(ctx context.Context, a asset.Item, respRaw []byte) error {
 	push, err := parseWSHeader(respRaw)
 	if err != nil {
 		return err
@@ -169,18 +170,26 @@ func (g *Gateio) wsHandleData(ctx context.Context, respRaw []byte) error {
 	if channelParts := strings.Split(push.Channel, "."); len(channelParts) != 2 {
 		return fmt.Errorf("%w `channel` (`%s`)", errParsingWSField, push.Channel)
 	} else {
-		// Assign to push so that we can abstract CoinM/USDT as just Futures
-		if push.assetType, err = asset.New(channelParts[0]); err != nil {
+		if a.String() != channelParts[0] {
 			return fmt.Errorf("%w `channel`; %w: `%s`", errParsingWSField, asset.ErrInvalidAsset, channelParts[0])
 		}
 		push.Channel = channelParts[1]
 	}
 
-	switch push.assetType {
-	case asset.Spot:
-		return g.wsHandleSpotData(ctx, push, respRaw)
+	assetHandlers, ok := wsHandlerFuncs[a]
+	if !ok {
+		return fmt.Errorf("%w `channel`; %w: `%s`", errParsingWSField, asset.ErrNotSupported, a)
 	}
-	return fmt.Errorf("%w `channel`; %w: `%s`", errParsingWSField, asset.ErrNotSupported, push.assetType)
+
+	handler, ok := wsHandlerFuncs[a][push.Channel]
+	if !ok {
+		g.Websocket.DataHandler <- stream.UnhandledMessageWarning{
+			Message: g.Name + ": " + stream.UnhandledMessage + ": " + string(respRaw),
+		}
+		return errors.New(stream.UnhandledMessage)
+	}
+
+	return handler(ctx, a, push, respRaw)
 }
 
 // wsHandleSpotData handles spot data

exchanges/gateio/gateio_websocket_delivery_futures.go

@@ -1,19 +1,5 @@
 package gateio
 
-import (
-	"context"
-	"errors"
-	"strconv"
-	"time"
-
-	"github.com/thrasher-corp/gocryptotrader/currency"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/account"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
-)
-
 const (
 	// delivery real trading urls
 	deliveryRealUSDTTradingURL = "wss://fx-ws.gateio.ws/v4/ws/delivery/usdt"
@@ -26,6 +12,7 @@ const (
 
 var fetchedFuturesCurrencyPairSnapshotOrderbook = make(map[string]bool)
 
+/*
 // GenerateDeliveryFuturesDefaultSubscriptions returns delivery futures default subscriptions params.
 func (g *Gateio) GenerateDeliveryFuturesDefaultSubscriptions() (subscription.List, error) {
 	_, err := g.GetCredentials(context.Background())
@@ -174,3 +161,4 @@ func (g *Gateio) generateDeliveryFuturesPayload(ctx context.Context, conn stream
 	}
 	return outbound, nil
 }
+*/

exchanges/gateio/gateio_websocket_futures.go

@@ -4,12 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"strconv"
 	"strings"
 	"time"
 
-	"github.com/thrasher-corp/gocryptotrader/common"
 	"github.com/thrasher-corp/gocryptotrader/currency"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/account"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
@@ -23,6 +21,7 @@ import (
 	"github.com/thrasher-corp/gocryptotrader/exchanges/trade"
 )
 
+/*
 // GenerateFuturesDefaultSubscriptions returns default subscriptions information.
 func (g *Gateio) GenerateFuturesDefaultSubscriptions(a asset.Item) (subscription.List, error) {
 	channelsToSubscribe := defaultFuturesSubscriptions
@@ -73,21 +72,10 @@ func (g *Gateio) FuturesSubscribe(ctx context.Context, conn stream.Connection, c
 func (g *Gateio) FuturesUnsubscribe(ctx context.Context, conn stream.Connection, channelsToUnsubscribe subscription.List) error {
 	return g.handleSubscription(ctx, conn, unsubscribeEvent, channelsToUnsubscribe, g.generateFuturesPayload)
 }
+*/
 
 // WsHandleFuturesData handles futures websocket data
-func (g *Gateio) WsHandleFuturesData(_ context.Context, respRaw []byte, a asset.Item) error {
-	push, err := parseWSHeader(respRaw)
-	if err != nil {
-		return err
-	}
-
-	if push.Event == subscribeEvent || push.Event == unsubscribeEvent {
-		if !g.Websocket.Match.IncomingWithData(push.ID, respRaw) {
-			return fmt.Errorf("couldn't match subscription message with ID: %d", push.ID)
-		}
-		return nil
-	}
-
+func (g *Gateio) wsHandleFuturesData(_ context.Context, a asset.Item, push *WSResponse, respRaw []byte) error {
 	switch push.Channel {
 	case futuresTickersChannel:
 		return g.processFuturesTickers(respRaw, a)

exchanges/gateio/gateio_wrapper.go

@@ -180,114 +180,48 @@ func (g *Gateio) SetDefaults() {
 
 // Setup sets user configuration
 func (g *Gateio) Setup(exch *config.Exchange) error {
-	err := exch.Validate()
-	if err != nil {
+	if err := exch.Validate(); err != nil {
 		return err
 	}
 	if !exch.Enabled {
 		g.SetEnabled(false)
 		return nil
 	}
-	err = g.SetupDefaults(exch)
-	if err != nil {
+	if err = g.SetupDefaults(exch); err != nil {
 		return err
 	}
 
-	err = g.Websocket.Setup(&stream.WebsocketSetup{
+	if err := g.Websocket.Setup(&stream.WebsocketSetup{
 		ExchangeConfig:               exch,
 		Features:                     &g.Features.Supports.WebsocketCapabilities,
 		FillsFeed:                    g.Features.Enabled.FillsFeed,
 		TradeFeed:                    g.Features.Enabled.TradeFeed,
 		GenerateSubscriptions:        g.generateSubscriptions,
 		UseMultiConnectionManagement: true,
 		RateLimitDefinitions:         packageRateLimits,
-	})
-	if err != nil {
-		return err
-	}
-	// Spot connection
-	err = g.Websocket.SetupNewConnection(&stream.ConnectionSetup{
-		URL:                      gateioWebsocketEndpoint,
-		ResponseCheckTimeout:     exch.WebsocketResponseCheckTimeout,
-		ResponseMaxLimit:         exch.WebsocketResponseMaxLimit,
-		Handler:                  g.WsHandleData,
-		Subscriber:               g.Subscribe,
-		Unsubscriber:             g.Unsubscribe,
-		Connector:                g.WsConnect,
-		Authenticate:             g.authenticateSpot,
-		MessageFilter:            asset.Spot,
-		BespokeGenerateMessageID: g.GenerateWebsocketMessageID,
-	})
-	if err != nil {
-		return err
-	}
-	// Futures connection - USDT margined
-	err = g.Websocket.SetupNewConnection(&stream.ConnectionSetup{
-		URL:                  usdtFuturesWebsocketURL,
-		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
-		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
-		Handler: func(ctx context.Context, incoming []byte) error {
-			return g.WsHandleFuturesData(ctx, incoming, asset.USDTMarginedFutures)
-		},
-		Subscriber:               g.FuturesSubscribe,
-		Unsubscriber:             g.FuturesUnsubscribe,
-		Connector:                g.WsConnect,
-		MessageFilter:            asset.USDTMarginedFutures,
-		BespokeGenerateMessageID: g.GenerateWebsocketMessageID,
-	})
-	if err != nil {
+	}); err != nil {
 		return err
 	}
 
-	// Futures connection - BTC margined
-	err = g.Websocket.SetupNewConnection(&stream.ConnectionSetup{
-		URL:                  btcFuturesWebsocketURL,
-		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
-		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
-		Handler: func(ctx context.Context, incoming []byte) error {
-			return g.WsHandleFuturesData(ctx, incoming, asset.CoinMarginedFutures)
-		},
-		Subscriber:               g.FuturesSubscribe,
-		Unsubscriber:             g.FuturesUnsubscribe,
-		Connector:                g.WsConnect,
-		MessageFilter:            asset.CoinMarginedFutures,
-		BespokeGenerateMessageID: g.GenerateWebsocketMessageID,
-	})
-	if err != nil {
-		return err
-	}
-
-	// TODO: Add BTC margined delivery futures.
-	// Futures connection - Delivery - USDT margined
-	err = g.Websocket.SetupNewConnection(&stream.ConnectionSetup{
-		URL:                  deliveryRealUSDTTradingURL,
-		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
-		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
-		Handler: func(ctx context.Context, incoming []byte) error {
-			return g.WsHandleFuturesData(ctx, incoming, asset.DeliveryFutures)
-		},
-		Subscriber:               g.DeliveryFuturesSubscribe,
-		Unsubscriber:             g.DeliveryFuturesUnsubscribe,
-		Connector:                g.WsConnect,
-		MessageFilter:            asset.DeliveryFutures,
-		BespokeGenerateMessageID: g.GenerateWebsocketMessageID,
-	})
-	if err != nil {
-		return err
-	}
-
-	// Futures connection - Options
-	return g.Websocket.SetupNewConnection(&stream.ConnectionSetup{
-		URL:                      optionsWebsocketURL,
-		ResponseCheckTimeout:     exch.WebsocketResponseCheckTimeout,
-		ResponseMaxLimit:         exch.WebsocketResponseMaxLimit,
-		Handler:                  g.WsHandleOptionsData,
-		Subscriber:               g.OptionsSubscribe,
-		Unsubscriber:             g.OptionsUnsubscribe,
-		Connector:                g.WsConnect,
-		MessageFilter:            asset.Options,
-		BespokeGenerateMessageID: g.GenerateWebsocketMessageID,
-	})
+	var errs error
+	for _, a := range g.GetEnabledAssets() {
+		if err = g.Websocket.SetupNewConnection(&stream.ConnectionSetup{
+			URL:                  wsURLs[a],
+			ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
+			ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
+			Handler:              func(ctx context.Context, msg []byte) error { return g.wsHandleData(ctx, msg, a) },
+			Subscriber:           g.Subscribe,
+			Unsubscriber:         g.Unsubscribe,
+			Connector:            g.WsConnect,
+			// TODO: Abstract auth spot
+			Authenticate:             g.authenticateSpot,
+			MessageFilter:            a,
+			BespokeGenerateMessageID: g.GenerateWebsocketMessageID,
+		}); err != nil {
+			errs = common.AppendError(errs, fmt.Errorf("%w: %s", err, a))
+		}
+	}
+	return errs
 }
 
 // UpdateTicker updates and returns the ticker for a currency pair