Websockets: Add channel.EnsureKeyed() for conv.

exchanges/kraken/kraken_websocket.go

@@ -1204,7 +1204,10 @@ func (k *Kraken) subscribeToChan(c *stream.ChannelSubscription) error {
 		c.Asset = asset.Spot
 	}
 
-	key := stream.DefaultChannelKey{c.Channel, c.Currency, c.Asset}
+	key, ok := c.EnsureKeyed().(stream.DefaultChannelKey)
+	if !ok {
+		return common.GetTypeAssertError("stream.DefaultChannelKey", c.Key, "subscription.Key") // Should be impossible
+	}
 
 	if c.Channel == krakenWsOrderbook {
 		if depth, err := depthFromChan(c); err != nil {

exchanges/stream/websocket.go

@@ -941,15 +941,7 @@ func (w *Websocket) AddSuccessfulSubscriptions(channels ...ChannelSubscription)
 		w.subscriptions = subscriptionMap{}
 	}
 	for i := range channels {
-		key := channels[i].Key
-		if key == nil {
-			key = DefaultChannelKey{
-				Channel:  channels[i].Channel,
-				Asset:    channels[i].Asset,
-				Currency: channels[i].Currency,
-			}
-			channels[i].Key = key
-		}
+		key := channels[i].EnsureKeyed()
 		w.subscriptions[key] = channels[i]
 	}
 	w.subscriptionMutex.Unlock()
@@ -970,10 +962,23 @@ func (w *Websocket) RemoveSuccessfulUnsubscriptions(channels ...ChannelSubscript
 	}
 }
 
+// EnsureKeyed sets the default key on a channel if it doesn't have one
+// Returns key for convenience
+func (c *ChannelSubscription) EnsureKeyed() any {
+	if c.Key == nil {
+		c.Key = DefaultChannelKey{
+			Channel:  c.Channel,
+			Asset:    c.Asset,
+			Currency: c.Currency,
+		}
+	}
+	return c.Key
+}
+
 // Equal two WebsocketChannelSubscription to determine equality
-func (w *ChannelSubscription) Equal(s *ChannelSubscription) bool {
-	return strings.EqualFold(w.Channel, s.Channel) &&
-		w.Currency.Equal(s.Currency)
+func (c *ChannelSubscription) Equal(b *ChannelSubscription) bool {
+	return strings.EqualFold(c.Channel, b.Channel) &&
+		c.Currency.Equal(b.Currency)
 }
 
 // GetSubscription returns a pointer to a copy of the subscription at the key provided