Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

common/gateio/stream: add thread-safe counter and overide default GenerateMessageID with connection specific implementation #1615

Merged
merged 8 commits into from
Aug 24, 2024
Merged

common/gateio/stream: add thread-safe counter and overide default GenerateMessageID with connection specific implementation #1615

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
7110e88
add counter and update gateio
Aug 16, 2024
0b729c3
Update exchanges/gateio/gateio.go
shazbert Aug 16, 2024
abde022
thrasher: nits
Aug 19, 2024
9257478
Merge branch 'master' into common_counter
Aug 19, 2024
313aaa1
add test case
Aug 19, 2024
9e0ce0d
linter: fix
Aug 19, 2024
8a31eec
revert change
Aug 19, 2024
138950a
thrasher nits
Aug 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions common/common.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"unicode"

Expand Down Expand Up @@ -671,3 +672,19 @@ func SortStrings[S ~[]E, E fmt.Stringer](x S) S {
})
return n
}

// Counter is a thread-safe counter.
type Counter struct {
n int64 // privatised so you can't use counter as a value type
}

// IncrementAndGet returns the next count after incrementing.
func (c *Counter) IncrementAndGet() int64 {
newID := atomic.AddInt64(&c.n, 1)
// Handle overflow by resetting the counter to 1 if it becomes negative
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Had thoughts of suggesting uint64 instead, but there are many unknowns and I doubt anyone is going to subscribe to WIF more 9,223,372,036,854,775,807 times on GateIO in a single session

if newID < 0 {
atomic.StoreInt64(&c.n, 1)
return 1
}
return newID
}
15 changes: 15 additions & 0 deletions common/common_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -869,3 +869,18 @@ func (a A) String() string {
func TestSortStrings(t *testing.T) {
assert.Equal(t, []A{1, 2, 5, 6}, SortStrings([]A{6, 2, 5, 1}))
}

func TestCounter(t *testing.T) {
t.Parallel()
c := Counter{n: -5}
require.Equal(t, int64(1), c.IncrementAndGet())
require.Equal(t, int64(2), c.IncrementAndGet())
}

// 683185328 1.787 ns/op 0 B/op 0 allocs/op
func BenchmarkCounter(b *testing.B) {
c := Counter{}
for i := 0; i < b.N; i++ {
c.IncrementAndGet()
}
}
1 change: 1 addition & 0 deletions exchanges/gateio/gateio.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,7 @@ var (
// Gateio is the overarching type across this package
type Gateio struct {
exchange.Base
Counter common.Counter
}

// ***************************************** SubAccounts ********************************
Expand Down
2 changes: 1 addition & 1 deletion exchanges/gateio/gateio_websocket.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -819,7 +819,7 @@ func (g *Gateio) generatePayload(event string, channelsToSubscribe subscription.
}

payload := WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Event: event,
Channel: channelsToSubscribe[i].Channel,
Payload: params,
Expand Down
6 changes: 3 additions & 3 deletions exchanges/gateio/gateio_ws_delivery_futures.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ func (g *Gateio) WsDeliveryFuturesConnect() error {
g.Websocket.GetWebsocketURL())
}
pingMessage, err := json.Marshal(WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Time: time.Now().Unix(),
Channel: futuresPingChannel,
})
Expand Down Expand Up @@ -316,7 +316,7 @@ func (g *Gateio) generateDeliveryFuturesPayload(event string, channelsToSubscrib
}
if strings.HasPrefix(channelsToSubscribe[i].Pairs[0].Quote.Upper().String(), "USDT") {
payloads[0] = append(payloads[0], WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Event: event,
Channel: channelsToSubscribe[i].Channel,
Payload: params,
Expand All @@ -325,7 +325,7 @@ func (g *Gateio) generateDeliveryFuturesPayload(event string, channelsToSubscrib
})
} else {
payloads[1] = append(payloads[1], WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Event: event,
Channel: channelsToSubscribe[i].Channel,
Payload: params,
Expand Down
6 changes: 3 additions & 3 deletions exchanges/gateio/gateio_ws_futures.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,7 @@ func (g *Gateio) WsFuturesConnect() error {
g.Websocket.GetWebsocketURL())
}
pingMessage, err := json.Marshal(WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Time: func() int64 {
return time.Now().Unix()
}(),
Expand Down Expand Up @@ -401,7 +401,7 @@ func (g *Gateio) generateFuturesPayload(event string, channelsToSubscribe subscr
}
if strings.HasPrefix(channelsToSubscribe[i].Pairs[0].Quote.Upper().String(), "USDT") {
payloads[0] = append(payloads[0], WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Event: event,
Channel: channelsToSubscribe[i].Channel,
Payload: params,
Expand All @@ -410,7 +410,7 @@ func (g *Gateio) generateFuturesPayload(event string, channelsToSubscribe subscr
})
} else {
payloads[1] = append(payloads[1], WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Event: event,
Channel: channelsToSubscribe[i].Channel,
Payload: params,
Expand Down
4 changes: 2 additions & 2 deletions exchanges/gateio/gateio_ws_option.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,7 @@ func (g *Gateio) WsOptionsConnect() error {
return err
}
pingMessage, err := json.Marshal(WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Time: time.Now().Unix(),
Channel: optionsPingChannel,
})
Expand Down Expand Up @@ -275,7 +275,7 @@ func (g *Gateio) generateOptionsPayload(event string, channelsToSubscribe subscr
params...)
}
payloads[i] = WsInput{
ID: g.Websocket.Conn.GenerateMessageID(false),
ID: g.Counter.IncrementAndGet(),
Event: event,
Channel: channelsToSubscribe[i].Channel,
Payload: params,
Expand Down
Loading