Websocket: Remove GenerateMessageID

cmd/documentation/exchange_templates/exchange_websocket_readme.tmpl

@@ -109,7 +109,6 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Unsubscriber:             e.SpotUnsubscribe,
 		GenerateSubscriptions:    e.GenerateDefaultSubscriptionsSpot,
 		Connector:                e.WsConnectSpot,
-		BespokeGenerateMessageID: e.GenerateWebsocketMessageID,
 	}); err != nil {
 		return err
 	}
@@ -125,7 +124,6 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Unsubscriber:             e.FuturesUnsubscribe,
 		GenerateSubscriptions:    func() (subscription.List, error) { return e.GenerateFuturesDefaultSubscriptions(currency.USDT) },
 		Connector:                e.WsFuturesConnect,
-		BespokeGenerateMessageID: e.GenerateWebsocketMessageID,
 	}); err != nil {
 		return err
 	}

docs/ADD_NEW_EXCHANGE.md

@@ -358,6 +358,12 @@ Alternatively you can use `request.WithVerbose(context.Background())` as the `co
 
 Ensure each endpoint is implemented and has an associated test to improve test coverage and increase confidence
 
+#### Message IDs
+
+If the exchange uses unique string message IDs, then your API functions can call e.MessageID() to get a UUID V7 message ID.
+However if the exchange does not support that format then you should override MessageID with an appropriate implementation.
+For Example: Consider `common.Counter` if universal uniqueness is not a requirement and you need an integer id.
+
 #### Authenticated functions
 
 Authenticated request function is created based on the way the exchange documentation specifies. For example, see the [Binance Spot API - Endpoint Security Types](https://developers.binance.com/docs/binance-spot-api-docs/rest-api/endpoint-security-type).

exchange/websocket/README.md

@@ -127,7 +127,6 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Unsubscriber:             e.SpotUnsubscribe,
 		GenerateSubscriptions:    e.GenerateDefaultSubscriptionsSpot,
 		Connector:                e.WsConnectSpot,
-		BespokeGenerateMessageID: e.GenerateWebsocketMessageID,
 	}); err != nil {
 		return err
 	}
@@ -143,7 +142,6 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Unsubscriber:             e.FuturesUnsubscribe,
 		GenerateSubscriptions:    func() (subscription.List, error) { return e.GenerateFuturesDefaultSubscriptions(currency.USDT) },
 		Connector:                e.WsFuturesConnect,
-		BespokeGenerateMessageID: e.GenerateWebsocketMessageID,
 	}); err != nil {
 		return err
 	}

exchange/websocket/connection.go

@@ -5,11 +5,9 @@ import (
 	"compress/flate"
 	"compress/gzip"
 	"context"
-	"crypto/rand"
 	"errors"
 	"fmt"
 	"io"
-	"math/big"
 	"net"
 	"net/http"
 	"net/url"
@@ -37,10 +35,6 @@ type Connection interface {
 	Dial(context.Context, *gws.Dialer, http.Header) error
 	ReadMessage() Response
 	SetupPingHandler(request.EndpointLimit, PingHandler)
-	// GenerateMessageID generates a message ID for the individual connection. If a bespoke function is set
-	// (by using SetupNewConnection) it will use that, otherwise it will use the defaultGenerateMessageID function
-	// defined in websocket_connection.go.
-	GenerateMessageID(highPrecision bool) int64
 	// SendMessageReturnResponse will send a WS message to the connection and wait for response
 	SendMessageReturnResponse(ctx context.Context, epl request.EndpointLimit, signature, request any) ([]byte, error)
 	// SendMessageReturnResponses will send a WS message to the connection and wait for N responses
@@ -87,12 +81,8 @@ type ConnectionSetup struct {
 	// Handler defines the function that will be called when a message is
 	// received from the exchange's websocket server. This function should
 	// handle the incoming message and pass it to the appropriate data handler.
-	Handler func(ctx context.Context, conn Connection, incoming []byte) error
-	// RequestIDGenerator is a function that returns a unique message ID.
-	// This is useful for when an exchange connection requires a unique or
-	// structured message ID for each message sent.
-	RequestIDGenerator func() int64
-	Authenticate       func(ctx context.Context, conn Connection) error
+	Handler      func(ctx context.Context, conn Connection, incoming []byte) error
+	Authenticate func(ctx context.Context, conn Connection) error
 	// MessageFilter defines the criteria used to match messages to a specific connection.
 	// The filter enables precise routing and handling of messages for distinct connection contexts.
 	MessageFilter any
@@ -128,7 +118,6 @@ type connection struct {
 	ResponseMaxLimit     time.Duration
 	Traffic              chan struct{}
 	readMessageErrors    chan error
-	requestIDGenerator   func() int64
 }
 
 // Dial sets proxy urls and then connects to the websocket
@@ -335,33 +324,6 @@ func (c *connection) parseBinaryResponse(resp []byte) ([]byte, error) {
 	return standardMessage, reader.Close()
 }
 
-// GenerateMessageID generates a message ID for the individual connection.
-// If a bespoke function is set (by using SetupNewConnection) it will use that,
-// otherwise it will use the defaultGenerateMessageID function.
-func (c *connection) GenerateMessageID(highPrec bool) int64 {
-	if c.requestIDGenerator != nil {
-		return c.requestIDGenerator()
-	}
-	return c.defaultGenerateMessageID(highPrec)
-}
-
-// defaultGenerateMessageID generates the default message ID
-func (c *connection) defaultGenerateMessageID(highPrec bool) int64 {
-	var minValue int64 = 1e8
-	var maxValue int64 = 2e8
-	if highPrec {
-		maxValue = 2e12
-		minValue = 1e12
-	}
-	// utilization of hard coded positive numbers and default crypto/rand
-	// io.reader will panic on error instead of returning
-	randomNumber, err := rand.Int(rand.Reader, big.NewInt(maxValue-minValue+1))
-	if err != nil {
-		panic(err)
-	}
-	return randomNumber.Int64() + minValue
-}
-
 // Shutdown shuts down and closes specific connection
 func (c *connection) Shutdown() error {
 	if c == nil || c.Connection == nil {

exchange/websocket/manager.go

@@ -305,7 +305,7 @@ func (m *Manager) SetupNewConnection(c *ConnectionSetup) error {
 		return err
 	}
 
-	if c.ResponseCheckTimeout == 0 && c.ResponseMaxLimit == 0 && c.RateLimit == nil && c.URL == "" && c.ConnectionLevelReporter == nil && c.RequestIDGenerator == nil {
+	if c.ResponseCheckTimeout == 0 && c.ResponseMaxLimit == 0 && c.RateLimit == nil && c.URL == "" && c.ConnectionLevelReporter == nil {
 		return fmt.Errorf("%w: %w", errConnSetup, errExchangeConfigEmpty)
 	}
 
@@ -401,7 +401,6 @@ func (m *Manager) getConnectionFromSetup(c *ConnectionSetup) *connection {
 		Match:                match,
 		RateLimit:            c.RateLimit,
 		Reporter:             c.ConnectionLevelReporter,
-		requestIDGenerator:   c.RequestIDGenerator,
 		RateLimitDefinitions: m.rateLimitDefinitions,
 	}
 }

exchange/websocket/manager_test.go

@@ -564,7 +564,7 @@ func TestSendMessageReturnResponse(t *testing.T) {
 		Subscription: testRequestData{
 			Name: "ticker",
 		},
-		RequestID: wc.GenerateMessageID(false),
+		RequestID: 12345,
 	}
 
 	_, err = wc.SendMessageReturnResponse(t.Context(), request.Unset, req.RequestID, req)
@@ -758,37 +758,6 @@ func TestCanUseAuthenticatedWebsocketForWrapper(t *testing.T) {
 	assert.True(t, ws.CanUseAuthenticatedWebsocketForWrapper(), "CanUseAuthenticatedWebsocketForWrapper should return true")
 }
 
-func TestGenerateMessageID(t *testing.T) {
-	t.Parallel()
-	wc := connection{}
-	const spins = 1000
-	ids := make([]int64, spins)
-	for i := range spins {
-		id := wc.GenerateMessageID(true)
-		assert.NotContains(t, ids, id, "GenerateMessageID should not generate the same ID twice")
-		ids[i] = id
-	}
-
-	wc.requestIDGenerator = func() int64 { return 42 }
-	assert.EqualValues(t, 42, wc.GenerateMessageID(true), "GenerateMessageID should use bespokeGenerateMessageID")
-}
-
-// 7002502	       166.7 ns/op	      48 B/op	       3 allocs/op
-func BenchmarkGenerateMessageID_High(b *testing.B) {
-	wc := connection{}
-	for b.Loop() {
-		_ = wc.GenerateMessageID(true)
-	}
-}
-
-// 6536250	       186.1 ns/op	      48 B/op	       3 allocs/op
-func BenchmarkGenerateMessageID_Low(b *testing.B) {
-	wc := connection{}
-	for b.Loop() {
-		_ = wc.GenerateMessageID(false)
-	}
-}
-
 func TestCheckWebsocketURL(t *testing.T) {
 	err := checkWebsocketURL("")
 	assert.ErrorIs(t, err, errInvalidWebsocketURL, "checkWebsocketURL should error correctly on empty string")
@@ -1135,7 +1104,7 @@ func TestLatency(t *testing.T) {
 		Event:        "subscribe",
 		Pairs:        []string{currency.NewPairWithDelimiter("XBT", "USD", "/").String()},
 		Subscription: testRequestData{Name: "ticker"},
-		RequestID:    wc.GenerateMessageID(false),
+		RequestID:    12346,
 	}
 
 	_, err = wc.SendMessageReturnResponse(t.Context(), request.Unset, req.RequestID, req)

exchanges/binance/binance_test.go

@@ -2003,7 +2003,7 @@ func TestSubscribe(t *testing.T) {
 			var req WsPayload
 			require.NoError(tb, json.Unmarshal(msg, &req), "Unmarshal must not error")
 			require.ElementsMatch(tb, req.Params, exp, "Params must have correct channels")
-			return w.WriteMessage(gws.TextMessage, fmt.Appendf(nil, `{"result":null,"id":%d}`, req.ID))
+			return w.WriteMessage(gws.TextMessage, fmt.Appendf(nil, `{"result":null,"id":"%s"}`, req.ID))
 		}
 		e = testexch.MockWsInstance[Exchange](t, mockws.CurryWsMockUpgrader(t, mock))
 	} else {
@@ -2025,7 +2025,7 @@ func TestSubscribeBadResp(t *testing.T) {
 		var req WsPayload
 		err := json.Unmarshal(msg, &req)
 		require.NoError(tb, err, "Unmarshal must not error")
-		return w.WriteMessage(gws.TextMessage, fmt.Appendf(nil, `{"result":{"error":"carrots"},"id":%d}`, req.ID))
+		return w.WriteMessage(gws.TextMessage, fmt.Appendf(nil, `{"result":{"error":"carrots"},"id":"%s"}`, req.ID))
 	}
 	b := testexch.MockWsInstance[Exchange](t, mockws.CurryWsMockUpgrader(t, mock))
 	err := b.Subscribe(channels)

exchanges/binance/binance_types.go

@@ -857,7 +857,7 @@ type WsListStatusData struct {
 type WsPayload struct {
 	Method string   `json:"method"`
 	Params []string `json:"params"`
-	ID     int64    `json:"id"`
+	ID     string   `json:"id"`
 }
 
 // CrossMarginInterestData stores cross margin data for borrowing

exchanges/binance/binance_websocket.go

@@ -166,7 +166,7 @@ func (e *Exchange) wsReadData() {
 }
 
 func (e *Exchange) wsHandleData(respRaw []byte) error {
-	if id, err := jsonparser.GetInt(respRaw, "id"); err == nil {
+	if id, err := jsonparser.GetString(respRaw, "id"); err == nil {
 		if e.Websocket.Match.IncomingWithData(id, respRaw) {
 			return nil
 		}
@@ -582,7 +582,7 @@ func (e *Exchange) manageSubs(ctx context.Context, op string, subs subscription.
 	}
 
 	req := WsPayload{
-		ID:     e.Websocket.Conn.GenerateMessageID(false),
+		ID:     e.MessageID(),
 		Method: op,
 		Params: subs.QualifiedChannels(),
 	}

exchanges/bitfinex/bitfinex_websocket.go

@@ -1716,7 +1716,7 @@ func (e *Exchange) subscribeToChan(ctx context.Context, subs subscription.List)
 
 	// subId is a single round-trip identifier that provides linking sub requests to chanIDs
 	// Although docs only mention subId for wsBookChannel, it works for all chans
-	subID := strconv.FormatInt(e.Websocket.Conn.GenerateMessageID(false), 10)
+	subID := e.MessageID()
 	req["subId"] = subID
 
 	// Add a temporary Key so we can find this Sub when we get the resp without delay or context switch
@@ -1829,7 +1829,7 @@ func (e *Exchange) WsSendAuth(ctx context.Context) error {
 
 // WsNewOrder authenticated new order request
 func (e *Exchange) WsNewOrder(ctx context.Context, data *WsNewOrderRequest) (string, error) {
-	data.CustomID = e.Websocket.AuthConn.GenerateMessageID(false)
+	data.CustomID = e.MessageSequence()
 	req := makeRequestInterface(wsOrderNew, data)
 	resp, err := e.Websocket.AuthConn.SendMessageReturnResponse(ctx, request.Unset, data.CustomID, req)
 	if err != nil {

exchanges/bybit/bybit.go

@@ -30,8 +30,7 @@ import (
 type Exchange struct {
 	exchange.Base
 
-	messageIDSeq common.Counter
-	account      accountTypeHolder
+	account accountTypeHolder
 }
 
 const (

exchanges/bybit/bybit_test.go

@@ -3026,7 +3026,6 @@ type FixtureConnection struct {
 	websocket.Connection
 }
 
-func (d *FixtureConnection) GenerateMessageID(bool) int64                                  { return 1337 }
 func (d *FixtureConnection) SetupPingHandler(request.EndpointLimit, websocket.PingHandler) {}
 func (d *FixtureConnection) Dial(context.Context, *gws.Dialer, http.Header) error          { return d.dialError }
 

exchanges/bybit/bybit_websocket.go

@@ -110,7 +110,7 @@ func (e *Exchange) WebsocketAuthenticateConnection(ctx context.Context, conn web
 		return err
 	}
 	req := Authenticate{
-		RequestID: strconv.FormatInt(conn.GenerateMessageID(false), 10),
+		RequestID: e.MessageID(),
 		Operation: "auth",
 		Args:      []any{creds.Key, intNonce, hex.EncodeToString(hmac)},
 	}
@@ -128,7 +128,7 @@ func (e *Exchange) WebsocketAuthenticateConnection(ctx context.Context, conn web
 	return nil
 }
 
-func (e *Exchange) handleSubscriptions(conn websocket.Connection, operation string, subs subscription.List) (args []SubscriptionArgument, err error) {
+func (e *Exchange) handleSubscriptions(_ websocket.Connection, operation string, subs subscription.List) (args []SubscriptionArgument, err error) {
 	subs, err = subs.ExpandTemplates(e)
 	if err != nil {
 		return
@@ -139,7 +139,7 @@ func (e *Exchange) handleSubscriptions(conn websocket.Connection, operation stri
 			args = append(args, SubscriptionArgument{
 				auth:           b[0].Authenticated,
 				Operation:      operation,
-				RequestID:      strconv.FormatInt(conn.GenerateMessageID(false), 10),
+				RequestID:      e.MessageID(),
 				Arguments:      b.QualifiedChannels(),
 				associatedSubs: b,
 			})
@@ -664,7 +664,7 @@ func hasPotentialDelimiter(a asset.Item) bool {
 
 // TODO: Remove this function when template expansion is across all assets
 func (e *Exchange) submitDirectSubscription(ctx context.Context, conn websocket.Connection, a asset.Item, operation string, channelsToSubscribe subscription.List) error {
-	payloads, err := e.directSubscriptionPayload(conn, a, operation, channelsToSubscribe)
+	payloads, err := e.directSubscriptionPayload(a, operation, channelsToSubscribe)
 	if err != nil {
 		return err
 	}
@@ -702,17 +702,17 @@ func (e *Exchange) submitDirectSubscription(ctx context.Context, conn websocket.
 }
 
 // TODO: Remove this function when template expansion is across all assets
-func (e *Exchange) directSubscriptionPayload(conn websocket.Connection, assetType asset.Item, operation string, channelsToSubscribe subscription.List) ([]SubscriptionArgument, error) {
+func (e *Exchange) directSubscriptionPayload(assetType asset.Item, operation string, channelsToSubscribe subscription.List) ([]SubscriptionArgument, error) {
 	var args []SubscriptionArgument
 	arg := SubscriptionArgument{
 		Operation: operation,
-		RequestID: strconv.FormatInt(conn.GenerateMessageID(false), 10),
+		RequestID: e.MessageID(),
 		Arguments: []string{},
 	}
 	authArg := SubscriptionArgument{
 		auth:      true,
 		Operation: operation,
-		RequestID: strconv.FormatInt(conn.GenerateMessageID(false), 10),
+		RequestID: e.MessageID(),
 		Arguments: []string{},
 	}
 
@@ -757,7 +757,7 @@ func (e *Exchange) directSubscriptionPayload(conn websocket.Connection, assetTyp
 			args = append(args, arg)
 			arg = SubscriptionArgument{
 				Operation: operation,
-				RequestID: strconv.FormatInt(conn.GenerateMessageID(false), 10),
+				RequestID: e.MessageID(),
 				Arguments: []string{},
 			}
 		}

exchanges/bybit/bybit_wrapper.go

@@ -253,7 +253,6 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Handler: func(_ context.Context, conn websocket.Connection, resp []byte) error {
 			return e.wsHandleData(conn, asset.Spot, resp)
 		},
-		RequestIDGenerator: e.messageIDSeq.IncrementAndGet,
 	}); err != nil {
 		return err
 	}
@@ -276,7 +275,6 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Handler: func(_ context.Context, conn websocket.Connection, resp []byte) error {
 			return e.wsHandleData(conn, asset.Options, resp)
 		},
-		RequestIDGenerator: e.messageIDSeq.IncrementAndGet,
 	}); err != nil {
 		return err
 	}
@@ -305,8 +303,7 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Handler: func(_ context.Context, conn websocket.Connection, resp []byte) error {
 			return e.wsHandleData(conn, asset.USDTMarginedFutures, resp)
 		},
-		RequestIDGenerator: e.messageIDSeq.IncrementAndGet,
-		MessageFilter:      asset.USDTMarginedFutures, // Unused but it allows us to differentiate between the two linear futures types.
+		MessageFilter: asset.USDTMarginedFutures, // Unused but it allows us to differentiate between the two linear futures types.
 	}); err != nil {
 		return err
 	}
@@ -335,8 +332,7 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Handler: func(_ context.Context, conn websocket.Connection, resp []byte) error {
 			return e.wsHandleData(conn, asset.USDCMarginedFutures, resp)
 		},
-		RequestIDGenerator: e.messageIDSeq.IncrementAndGet,
-		MessageFilter:      asset.USDCMarginedFutures, // Unused but it allows us to differentiate between the two linear futures types.
+		MessageFilter: asset.USDCMarginedFutures, // Unused but it allows us to differentiate between the two linear futures types.
 	}); err != nil {
 		return err
 	}
@@ -359,7 +355,6 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Handler: func(_ context.Context, conn websocket.Connection, resp []byte) error {
 			return e.wsHandleData(conn, asset.CoinMarginedFutures, resp)
 		},
-		RequestIDGenerator: e.messageIDSeq.IncrementAndGet,
 	}); err != nil {
 		return err
 	}
@@ -381,7 +376,6 @@ func (e *Exchange) Setup(exch *config.Exchange) error {
 		Subscriber:            e.authSubscribe,
 		Unsubscriber:          e.authUnsubscribe,
 		Handler:               e.wsHandleAuthenticatedData,
-		RequestIDGenerator:    e.messageIDSeq.IncrementAndGet,
 		Authenticate:          e.WebsocketAuthenticateConnection,
 	})
 }