Exchanges: Migrate cool exchanges to multi-connection

contrib/spellcheck/ignore_words.txt

@@ -4,4 +4,5 @@ prevend
 flate
 zar
 insid
-totalin
+totalin
+bu

exchange/websocket/manager.go

@@ -495,7 +495,6 @@ func (m *Manager) connect(ctx context.Context) error {
 			}
 
 			if len(subs) == 0 {
-				// If no subscriptions are generated, we skip the connection
 				if m.verbose {
 					log.Warnf(log.WebsocketMgr, "%s websocket: no subscriptions generated", m.exchangeName)
 				}
@@ -614,7 +613,6 @@ func (m *Manager) createConnectAndSubscribe(ctx context.Context, ws *websocket,
 		}
 		return nil
 	}
-
 	if err := ws.setup.Subscriber(ctx, conn, subs); err != nil {
 		return fmt.Errorf("%w: %w", ErrSubscriptionFailure, err)
 	}
@@ -795,7 +793,6 @@ func (m *Manager) SetWebsocketURL(u string, auth, reconnect bool) error {
 		if defaultVals {
 			u = m.defaultURLAuth
 		}
-
 		err := checkWebsocketURL(u)
 		if err != nil {
 			return err

exchanges/bitfinex/bitfinex_wrapper.go

@@ -161,8 +161,9 @@ func (e *Exchange) SetDefaults() {
 	}
 	e.API.Endpoints = e.NewEndpoints()
 	err = e.API.Endpoints.SetDefaultEndpoints(map[exchange.URL]string{
-		exchange.RestSpot:      bitfinexAPIURLBase,
-		exchange.WebsocketSpot: publicBitfinexWebsocketEndpoint,
+		exchange.RestSpot:                   bitfinexAPIURLBase,
+		exchange.WebsocketSpot:              publicBitfinexWebsocketEndpoint,
+		exchange.WebsocketSpotSupplementary: authenticatedBitfinexWebsocketEndpoint,
 	})
 	if err != nil {
 		log.Errorln(log.ExchangeSys, err)
@@ -188,39 +189,51 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		return err
 	}
 
-	wsEndpoint, err := e.API.Endpoints.GetURL(exchange.WebsocketSpot)
+	err = e.Websocket.Setup(&websocket.ManagerSetup{
+		ExchangeConfig:                         exch,
+		Features:                               &e.Features.Supports.WebsocketCapabilities,
+		UseMultiConnectionManagement:           true,
+		MaxWebsocketSubscriptionsPerConnection: 25, // https://docs.bitfinex.com/docs/requirements-and-limitations
+	})
 	if err != nil {
 		return err
 	}
-
-	err = e.Websocket.Setup(&websocket.ManagerSetup{
-		ExchangeConfig:        exch,
-		DefaultURL:            publicBitfinexWebsocketEndpoint,
-		RunningURL:            wsEndpoint,
-		Connector:             e.WsConnect,
-		Subscriber:            e.Subscribe,
-		Unsubscriber:          e.Unsubscribe,
-		GenerateSubscriptions: e.generateSubscriptions,
-		Features:              &e.Features.Supports.WebsocketCapabilities,
-	})
+	wsPublicURL, err := e.API.Endpoints.GetURL(exchange.WebsocketSpot)
+	if err != nil {
+		return err
+	}
+	wsAuthURL, err := e.API.Endpoints.GetURL(exchange.WebsocketSpotSupplementary)
 	if err != nil {
 		return err
 	}
 
 	err = e.Websocket.SetupNewConnection(&websocket.ConnectionSetup{
-		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
-		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
-		URL:                  publicBitfinexWebsocketEndpoint,
+		Connector:             e.wsConnect,
+		Subscriber:            e.subscribeForConnection,
+		Unsubscriber:          e.unsubscribeForConnection,
+		GenerateSubscriptions: e.generatePublicSubscriptions,
+		Handler:               e.wsHandleData,
+		ResponseCheckTimeout:  exch.WebsocketResponseCheckTimeout,
+		ResponseMaxLimit:      exch.WebsocketResponseMaxLimit,
+		URL:                   wsPublicURL,
+		MessageFilter:         wsPublicURL,
 	})
 	if err != nil {
 		return err
 	}
 
 	return e.Websocket.SetupNewConnection(&websocket.ConnectionSetup{
-		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
-		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
-		URL:                  authenticatedBitfinexWebsocketEndpoint,
-		Authenticated:        true,
+		Connector:                e.wsConnect,
+		Authenticate:             e.wsSendAuthConn,
+		Subscriber:               e.subscribeForConnection,
+		Unsubscriber:             e.unsubscribeForConnection,
+		GenerateSubscriptions:    e.generatePrivateSubscriptions,
+		SubscriptionsNotRequired: true,
+		Handler:                  e.wsHandleData,
+		ResponseCheckTimeout:     exch.WebsocketResponseCheckTimeout,
+		ResponseMaxLimit:         exch.WebsocketResponseMaxLimit,
+		URL:                      wsAuthURL,
+		MessageFilter:            wsAuthURL,
 	})
 }
 
@@ -961,11 +974,6 @@ func (e *Exchange) GetOrderHistory(ctx context.Context, req *order.MultiOrderReq
 	return req.Filter(e.Name, orders), nil
 }
 
-// AuthenticateWebsocket sends an authentication message to the websocket
-func (e *Exchange) AuthenticateWebsocket(ctx context.Context) error {
-	return e.WsSendAuth(ctx)
-}
-
 // appendOptionalDelimiter ensures that a delimiter is present for long character currencies
 func (e *Exchange) appendOptionalDelimiter(p *currency.Pair) {
 	if (len(p.Base.String()) > 3 && !p.Quote.IsEmpty()) ||

exchanges/bitstamp/bitstamp_wrapper.go

@@ -167,7 +167,7 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 	}
 
 	return e.Websocket.SetupNewConnection(&websocket.ConnectionSetup{
-		URL:                  e.Websocket.GetWebsocketURL(),
+		URL:                  wsURL,
 		ResponseCheckTimeout: exch.WebsocketResponseCheckTimeout,
 		ResponseMaxLimit:     exch.WebsocketResponseMaxLimit,
 	})

exchanges/bybit/bybit_test.go

@@ -3041,8 +3041,9 @@ func TestWSHandleData(t *testing.T) {
 
 	keys := slices.Collect(maps.Keys(pushDataMap))
 	slices.Sort(keys)
+	conn := testexch.GetMockConn(t, e, "")
 	for x := range keys {
-		err := e.wsHandleData(t.Context(), nil, asset.Spot, []byte(pushDataMap[keys[x]]))
+		err := e.wsHandleData(t.Context(), conn, asset.Spot, []byte(pushDataMap[keys[x]]))
 		if keys[x] == "unhandled" {
 			assert.ErrorIs(t, err, errUnhandledStreamData, "wsHandleData should error correctly for unhandled topics")
 		} else {
@@ -3254,12 +3255,10 @@ func TestWsTicker(t *testing.T) {
 		asset.Spot, asset.Options, asset.USDTMarginedFutures, asset.USDTMarginedFutures,
 		asset.USDCMarginedFutures, asset.USDCMarginedFutures, asset.CoinMarginedFutures, asset.CoinMarginedFutures,
 	}
-	routingIndex := 0
+	conn := testexch.GetMockConn(t, e, "")
 	testexch.FixtureToDataHandler(t, "testdata/wsTicker.json", func(_ context.Context, r []byte) error {
-		require.Less(t, routingIndex, len(assetRouting), "routingIndex must stay within ticker fixture asset routing bounds")
-		a := assetRouting[routingIndex]
-		routingIndex++
-		return e.wsHandleData(t.Context(), nil, a, r)
+		defer slices.Delete(assetRouting, 0, 1)
+		return e.wsHandleData(t.Context(), conn, assetRouting[0], r)
 	})
 	e.Websocket.DataHandler.Close()
 	expected := 8
@@ -3508,7 +3507,7 @@ func TestFetchTradablePairs(t *testing.T) {
 func TestDeltaUpdateOrderbook(t *testing.T) {
 	t.Parallel()
 	data := []byte(`{"topic":"orderbook.50.WEMIXUSDT","ts":1697573183768,"type":"snapshot","data":{"s":"WEMIXUSDT","b":[["0.9511","260.703"],["0.9677","0"]],"a":[],"u":3119516,"seq":14126848493},"cts":1728966699481}`)
-	err := e.wsHandleData(t.Context(), nil, asset.Spot, data)
+	err := e.wsHandleData(t.Context(), testexch.GetMockConn(t, e, ""), asset.Spot, data)
 	require.NoError(t, err, "wsHandleData must not error")
 	update := []byte(`{"topic":"orderbook.50.WEMIXUSDT","ts":1697573183768,"type":"delta","data":{"s":"WEMIXUSDT","b":[["0.9511","260.703"],["0.9677","0"]],"a":[],"u":3119516,"seq":14126848493},"cts":1728966699481}`)
 	var wsResponse WebsocketResponse

exchanges/bybit/bybit_websocket_requests.go

@@ -8,15 +8,10 @@ import (
 
 	"github.com/gofrs/uuid"
 	"github.com/thrasher-corp/gocryptotrader/encoding/json"
+	exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
 	"github.com/thrasher-corp/gocryptotrader/exchanges/request"
 )
 
-// Websocket request operation types
-const (
-	OutboundTradeConnection  = "PRIVATE_TRADE"
-	InboundPrivateConnection = "PRIVATE"
-)
-
 // WSCreateOrder creates an order through the websocket connection
 func (e *Exchange) WSCreateOrder(ctx context.Context, r *PlaceOrderRequest) (*WebsocketOrderDetails, error) {
 	if err := r.Validate(); err != nil {
@@ -64,13 +59,21 @@ func (e *Exchange) WSCancelOrder(ctx context.Context, r *CancelOrderRequest) (*W
 
 // sendWebsocketTradeRequest sends a trade request to the exchange through the websocket connection
 func (e *Exchange) sendWebsocketTradeRequest(ctx context.Context, op, orderLinkID string, payload any, limit request.EndpointLimit) (*WebsocketOrderDetails, error) {
+	wsTradeURL, err := e.API.Endpoints.GetURL(exchange.WebsocketTrade)
+	if err != nil {
+		return nil, err
+	}
+	wsPrivateURL, err := e.API.Endpoints.GetURL(exchange.WebsocketPrivate)
+	if err != nil {
+		return nil, err
+	}
 	// Get the outbound and inbound connections to send and receive the request. This makes sure both are live before
 	// sending the request.
-	outbound, err := e.Websocket.GetConnection(OutboundTradeConnection)
+	outbound, err := e.Websocket.GetConnection(wsTradeURL)
 	if err != nil {
 		return nil, err
 	}
-	inbound, err := e.Websocket.GetConnection(InboundPrivateConnection)
+	inbound, err := e.Websocket.GetConnection(wsPrivateURL)
 	if err != nil {
 		return nil, err
 	}

exchanges/bybit/bybit_wrapper.go

@@ -303,7 +303,7 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Handler: func(ctx context.Context, conn websocket.Connection, resp []byte) error {
 			return e.wsHandleData(ctx, conn, asset.USDTMarginedFutures, resp)
 		},
-		MessageFilter: asset.USDTMarginedFutures, // Unused but it allows us to differentiate between the two linear futures types.
+		MessageFilter: asset.USDTMarginedFutures, // Required to differentiate linear futures connections sharing the same endpoint URL.
 	}); err != nil {
 		return err
 	}
@@ -331,7 +331,7 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Handler: func(ctx context.Context, conn websocket.Connection, resp []byte) error {
 			return e.wsHandleData(ctx, conn, asset.USDCMarginedFutures, resp)
 		},
-		MessageFilter: asset.USDCMarginedFutures, // Unused but it allows us to differentiate between the two linear futures types.
+		MessageFilter: asset.USDCMarginedFutures, // Required to differentiate linear futures connections sharing the same endpoint URL.
 	}); err != nil {
 		return err
 	}
@@ -372,7 +372,7 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 			return e.wsHandleTradeData(conn, resp)
 		},
 		Authenticate:             e.WebsocketAuthenticateTradeConnection,
-		MessageFilter:            OutboundTradeConnection,
+		MessageFilter:            wsTradeURL,
 		SubscriptionsNotRequired: true,
 	}); err != nil {
 		return err
@@ -395,7 +395,7 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Unsubscriber:          e.authUnsubscribe,
 		Handler:               e.wsHandleAuthenticatedData,
 		Authenticate:          e.WebsocketAuthenticatePrivateConnection,
-		MessageFilter:         InboundPrivateConnection,
+		MessageFilter:         wsPrivateURL,
 	})
 }