Kraken: Paired subscriptions

.github/workflows/tests.yml

@@ -1,7 +1,7 @@
 on: [push, pull_request]
 name: CI
 env:
-  GO_VERSION: 1.21.x
+  GO_VERSION: 1.22.x
 jobs:
   backend-psql:
     name: GoCryptoTrader back-end

exchanges/kraken/kraken_test.go

@@ -1178,34 +1178,16 @@ func TestWithdrawCancel(t *testing.T) {
 
 // ---------------------------- Websocket tests -----------------------------------------
 
-// setupWs is a helper function to connect both auth and normal websockets
-// It will skip the test if websockets are not enabled
-// It's up to the test to skip if it requires creds, though
-func setupWs(tb testing.TB) {
-	tb.Helper()
-
-	if !k.Websocket.IsEnabled() {
-		tb.Skip("Websocket not enabled")
-	}
-	if k.Websocket.IsConnected() {
-		return
-	}
-
-	if wsConnected {
-		return
-	}
-	wsConnected = true
-
-	// We don't use k.websocket.Connect() because it'd subscribe to channels
-	err := k.WsConnect()
-	assert.NoError(tb, err, "WsConnect should not error")
-}
-
 // TestWsSubscribe tests unauthenticated websocket subscriptions
 // Specifically looking to ensure multiple errors are collected and returned and ws.Subscriptions Added/Removed in cases of:
 // single pass, single fail, mixed fail, multiple pass, all fail
 // No objection to this becoming a fixture test, so long as it integrates through Un/Subscribe roundtrip
 func TestWsSubscribe(t *testing.T) {
+	t.Parallel()
+
+	k := new(Kraken)
+	require.NoError(t, testexch.TestInstance(k), "TestInstance must not error")
+	k.Features.Subscriptions = []*subscription.Subscription{}
 	testexch.SetupWs(t, k)
 
 	err := k.Subscribe([]subscription.Subscription{{Channel: krakenWsTicker, Pairs: currency.Pairs{currency.NewPairWithDelimiter("XBT", "USD", "/")}}})

exchanges/kraken/kraken_websocket.go

@@ -1039,12 +1039,12 @@ func (k *Kraken) GenerateDefaultSubscriptions() ([]subscription.Subscription, er
 
 // Subscribe sends a websocket message to receive data from the channel
 func (k *Kraken) Subscribe(channels []subscription.Subscription) error {
-	return k.ParallelChanOp(channels, k.subscribeToChan, len(channels))
+	return k.ParallelChanOp(channels, k.subscribeToChan, 1)
 }
 
 // Unsubscribe sends a websocket message to stop receiving data from the channel
 func (k *Kraken) Unsubscribe(channels []subscription.Subscription) error {
-	return k.ParallelChanOp(channels, k.unsubscribeFromChan, len(channels))
+	return k.ParallelChanOp(channels, k.unsubscribeFromChan, 1)
 }
 
 // subscribeToChan sends a websocket message to receive data from the channel
@@ -1088,6 +1088,7 @@ func (k *Kraken) subscribeToChan(chans []subscription.Subscription) error {
 	if err = k.getErrResp(respRaw); err != nil {
 		wErr := fmt.Errorf("%w Channel: %s Pair: %s; %w", stream.ErrSubscriptionFailure, c.Channel, c.Pairs, err)
 		k.Websocket.DataHandler <- wErr
+		// Currently all or nothing on pairs; Alternatively parse response and remove failing pairs and retry
 		k.Websocket.RemoveSubscriptions(c)
 		return wErr
 	}

go.mod

@@ -1,6 +1,6 @@
 module github.com/thrasher-corp/gocryptotrader
 
-go 1.20
+go 1.22
 
 require (
 	github.com/buger/jsonparser v1.1.1