bybit: enable multiconnection handling across websocket endpoints

docs/ADD_NEW_EXCHANGE.md

@@ -1176,7 +1176,7 @@ Please test all `pair` commands to disable and enable different assets types to
 - `get` to ensure correct enabled and disabled pairs for a supported asset type.
 - `disableasset` to ensure disabling of entire asset class and associated unsubscriptions.
 - `enableasset` to ensure correct enabling of entire asset class and associated subscriptions.
-- `disable` to ensure correct disabling of pair(s) and and associated unsubscriptions.
+- `disable` to ensure correct disabling of pair(s) and associated unsubscriptions.
 - `enable` to ensure correct enabling of pair(s) and associated subscriptions.
 - `enableall` to ensure correct enabling of all pairs for an asset type and associated subscriptions.
 - `disableall` to ensure correct disabling of all pairs for an asset type and associated unsubscriptions.

engine/engine.go

@@ -791,17 +791,10 @@ func (bot *Engine) LoadExchange(name string) error {
 
 	localWG.Wait()
 	if !bot.Settings.EnableExchangeHTTPRateLimiter {
-		gctlog.Warnf(gctlog.ExchangeSys,
-			"Loaded exchange %s rate limiting has been turned off.\n",
-			exch.GetName(),
-		)
+		gctlog.Warnf(gctlog.ExchangeSys, "Loaded exchange %s rate limiting has been turned off.\n", exch.GetName())
 		err = exch.DisableRateLimiter()
 		if err != nil {
-			gctlog.Errorf(gctlog.ExchangeSys,
-				"Loaded exchange %s rate limiting cannot be turned off: %s.\n",
-				exch.GetName(),
-				err,
-			)
+			gctlog.Errorf(gctlog.ExchangeSys, "Loaded exchange %s rate limiting cannot be turned off: %s.\n", exch.GetName(), err)
 		}
 	}
 
@@ -820,29 +813,18 @@ func (bot *Engine) LoadExchange(name string) error {
 		return err
 	}
 
-	base := exch.GetBase()
-	if base.API.AuthenticatedSupport ||
-		base.API.AuthenticatedWebsocketSupport {
-		assetTypes := base.GetAssetTypes(false)
-		var useAsset asset.Item
-		for a := range assetTypes {
-			err = base.CurrencyPairs.IsAssetEnabled(assetTypes[a])
-			if err != nil {
-				continue
-			}
-			useAsset = assetTypes[a]
-			break
-		}
-		err = exch.ValidateAPICredentials(context.TODO(), useAsset)
+	b := exch.GetBase()
+	if b.API.AuthenticatedSupport || b.API.AuthenticatedWebsocketSupport {
+		err = exch.ValidateAPICredentials(context.TODO(), asset.Spot)
 		if err != nil {
-			gctlog.Warnf(gctlog.ExchangeSys,
-				"%s: Cannot validate credentials, authenticated support has been disabled, Error: %s\n",
-				base.Name,
-				err)
-			base.API.AuthenticatedSupport = false
-			base.API.AuthenticatedWebsocketSupport = false
+			gctlog.Warnf(gctlog.ExchangeSys, "%s: Cannot validate credentials, authenticated support has been disabled, Error: %s", b.Name, err)
+			b.API.AuthenticatedSupport = false
+			b.API.AuthenticatedWebsocketSupport = false
 			exchCfg.API.AuthenticatedSupport = false
 			exchCfg.API.AuthenticatedWebsocketSupport = false
+			if b.Websocket != nil {
+				b.Websocket.SetCanUseAuthenticatedEndpoints(false)
+			}
 		}
 	}
 
@@ -854,10 +836,7 @@ func (bot *Engine) dryRunParamInteraction(param string) {
 		return
 	}
 
-	gctlog.Warnf(gctlog.Global,
-		"Command line argument '-%s' induces dry run mode."+
-			" Set -dryrun=false if you wish to override this.",
-		param)
+	gctlog.Warnf(gctlog.Global, "Command line argument '-%s' induces dry run mode. Set -dryrun=false if you wish to override this.", param)
 
 	if !bot.Settings.EnableDryRun {
 		bot.Settings.EnableDryRun = true

exchanges/bybit/bybit.go

@@ -27,10 +27,10 @@ import (
 // Bybit is the overarching type across this package
 type Bybit struct {
 	exchange.Base
-
 	// AccountType holds information about whether the account to which the api key belongs is a unified margin account or not.
 	// 0: unified, and 1: for normal account
 	AccountType uint8
+	Counter     common.Counter
 }
 
 const (

exchanges/bybit/bybit_inverse_websocket.go

@@ -2,81 +2,43 @@ package bybit
 
 import (
 	"context"
-	"net/http"
+	"errors"
 
-	"github.com/gorilla/websocket"
 	"github.com/thrasher-corp/gocryptotrader/currency"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
 )
 
-// WsInverseConnect connects to inverse websocket feed
-func (by *Bybit) WsInverseConnect() error {
-	if !by.Websocket.IsEnabled() || !by.IsEnabled() || !by.IsAssetWebsocketSupported(asset.CoinMarginedFutures) {
-		return stream.ErrWebsocketNotEnabled
-	}
-	by.Websocket.Conn.SetURL(inversePublic)
-	var dialer websocket.Dialer
-	err := by.Websocket.Conn.Dial(&dialer, http.Header{})
-	if err != nil {
-		return err
-	}
-	by.Websocket.Conn.SetupPingHandler(request.Unset, stream.PingHandler{
-		MessageType: websocket.TextMessage,
-		Message:     []byte(`{"op": "ping"}`),
-		Delay:       bybitWebsocketTimer,
-	})
-
-	by.Websocket.Wg.Add(1)
-	go by.wsReadData(asset.CoinMarginedFutures, by.Websocket.Conn)
-	return nil
-}
-
 // GenerateInverseDefaultSubscriptions generates default subscription
 func (by *Bybit) GenerateInverseDefaultSubscriptions() (subscription.List, error) {
-	var subscriptions subscription.List
-	var channels = []string{chanOrderbook, chanPublicTrade, chanPublicTicker}
 	pairs, err := by.GetEnabledPairs(asset.CoinMarginedFutures)
 	if err != nil {
+		if errors.Is(err, asset.ErrNotEnabled) {
+			return nil, nil
+		}
 		return nil, err
 	}
+
+	var subscriptions subscription.List
 	for z := range pairs {
-		for x := range channels {
-			subscriptions = append(subscriptions,
-				&subscription.Subscription{
-					Channel: channels[x],
-					Pairs:   currency.Pairs{pairs[z]},
-					Asset:   asset.CoinMarginedFutures,
-				})
+		for _, channel := range []string{chanOrderbook, chanPublicTrade, chanPublicTicker} {
+			subscriptions = append(subscriptions, &subscription.Subscription{
+				Channel: channel,
+				Pairs:   currency.Pairs{pairs[z]},
+				Asset:   asset.CoinMarginedFutures,
+			})
 		}
 	}
 	return subscriptions, nil
 }
 
 // InverseSubscribe sends a subscription message to linear public channels.
-func (by *Bybit) InverseSubscribe(channelSubscriptions subscription.List) error {
-	return by.handleInversePayloadSubscription("subscribe", channelSubscriptions)
+func (by *Bybit) InverseSubscribe(ctx context.Context, conn stream.Connection, channelSubscriptions subscription.List) error {
+	return by.handleSubscriptionNonTemplate(ctx, conn, asset.CoinMarginedFutures, "subscribe", channelSubscriptions)
 }
 
 // InverseUnsubscribe sends an unsubscription messages through linear public channels.
-func (by *Bybit) InverseUnsubscribe(channelSubscriptions subscription.List) error {
-	return by.handleInversePayloadSubscription("unsubscribe", channelSubscriptions)
-}
-
-func (by *Bybit) handleInversePayloadSubscription(operation string, channelSubscriptions subscription.List) error {
-	payloads, err := by.handleSubscriptions(operation, channelSubscriptions)
-	if err != nil {
-		return err
-	}
-	for a := range payloads {
-		// The options connection does not send the subscription request id back with the subscription notification payload
-		// therefore the code doesn't wait for the response to check whether the subscription is successful or not.
-		err = by.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, payloads[a])
-		if err != nil {
-			return err
-		}
-	}
-	return nil
+func (by *Bybit) InverseUnsubscribe(ctx context.Context, conn stream.Connection, channelSubscriptions subscription.List) error {
+	return by.handleSubscriptionNonTemplate(ctx, conn, asset.CoinMarginedFutures, "unsubscribe", channelSubscriptions)
 }

exchanges/bybit/bybit_linear_websocket.go

@@ -2,99 +2,43 @@ package bybit
 
 import (
 	"context"
-	"net/http"
+	"errors"
 
-	"github.com/gorilla/websocket"
 	"github.com/thrasher-corp/gocryptotrader/currency"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
-	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
 )
 
-// WsLinearConnect connects to linear a websocket feed
-func (by *Bybit) WsLinearConnect() error {
-	if !by.Websocket.IsEnabled() || !by.IsEnabled() || !by.IsAssetWebsocketSupported(asset.LinearContract) {
-		return stream.ErrWebsocketNotEnabled
-	}
-	by.Websocket.Conn.SetURL(linearPublic)
-	var dialer websocket.Dialer
-	err := by.Websocket.Conn.Dial(&dialer, http.Header{})
+// GenerateLinearDefaultSubscriptions generates default subscription
+func (by *Bybit) GenerateLinearDefaultSubscriptions(a asset.Item) (subscription.List, error) {
+	pairs, err := by.GetEnabledPairs(a)
 	if err != nil {
-		return err
-	}
-	by.Websocket.Conn.SetupPingHandler(request.Unset, stream.PingHandler{
-		MessageType: websocket.TextMessage,
-		Message:     []byte(`{"op": "ping"}`),
-		Delay:       bybitWebsocketTimer,
-	})
-
-	by.Websocket.Wg.Add(1)
-	go by.wsReadData(asset.LinearContract, by.Websocket.Conn)
-	if by.IsWebsocketAuthenticationSupported() {
-		err = by.WsAuth(context.TODO())
-		if err != nil {
-			by.Websocket.DataHandler <- err
-			by.Websocket.SetCanUseAuthenticatedEndpoints(false)
+		if errors.Is(err, asset.ErrNotEnabled) {
+			return nil, nil
 		}
+		return nil, err
 	}
-	return nil
-}
 
-// GenerateLinearDefaultSubscriptions generates default subscription
-func (by *Bybit) GenerateLinearDefaultSubscriptions() (subscription.List, error) {
 	var subscriptions subscription.List
-	var channels = []string{chanOrderbook, chanPublicTrade, chanPublicTicker}
-	pairs, err := by.GetEnabledPairs(asset.USDTMarginedFutures)
-	if err != nil {
-		return nil, err
-	}
-	linearPairMap := map[asset.Item]currency.Pairs{
-		asset.USDTMarginedFutures: pairs,
-	}
-	usdcPairs, err := by.GetEnabledPairs(asset.USDCMarginedFutures)
-	if err != nil {
-		return nil, err
-	}
-	linearPairMap[asset.USDCMarginedFutures] = usdcPairs
-	pairs = append(pairs, usdcPairs...)
-	for a := range linearPairMap {
-		for p := range linearPairMap[a] {
-			for x := range channels {
-				subscriptions = append(subscriptions,
-					&subscription.Subscription{
-						Channel: channels[x],
-						Pairs:   currency.Pairs{pairs[p]},
-						Asset:   a,
-					})
-			}
+	for _, pair := range pairs {
+		for _, channel := range []string{chanOrderbook, chanPublicTrade, chanPublicTicker} {
+			subscriptions = append(subscriptions, &subscription.Subscription{
+				Channel: channel,
+				Pairs:   currency.Pairs{pair},
+				Asset:   a,
+			})
 		}
 	}
 	return subscriptions, nil
 }
 
 // LinearSubscribe sends a subscription message to linear public channels.
-func (by *Bybit) LinearSubscribe(channelSubscriptions subscription.List) error {
-	return by.handleLinearPayloadSubscription("subscribe", channelSubscriptions)
+func (by *Bybit) LinearSubscribe(ctx context.Context, conn stream.Connection, channelSubscriptions subscription.List) error {
+	return by.handleSubscriptionNonTemplate(ctx, conn, asset.USDTMarginedFutures, "subscribe", channelSubscriptions)
 }
 
 // LinearUnsubscribe sends an unsubscription messages through linear public channels.
-func (by *Bybit) LinearUnsubscribe(channelSubscriptions subscription.List) error {
-	return by.handleLinearPayloadSubscription("unsubscribe", channelSubscriptions)
-}
-
-func (by *Bybit) handleLinearPayloadSubscription(operation string, channelSubscriptions subscription.List) error {
-	payloads, err := by.handleSubscriptions(operation, channelSubscriptions)
-	if err != nil {
-		return err
-	}
-	for a := range payloads {
-		// The options connection does not send the subscription request id back with the subscription notification payload
-		// therefore the code doesn't wait for the response to check whether the subscription is successful or not.
-		err = by.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, payloads[a])
-		if err != nil {
-			return err
-		}
-	}
-	return nil
+func (by *Bybit) LinearUnsubscribe(ctx context.Context, conn stream.Connection, channelSubscriptions subscription.List) error {
+	return by.handleSubscriptionNonTemplate(ctx, conn, asset.USDTMarginedFutures, "unsubscribe", channelSubscriptions)
 }