Skip to content

Commit

Permalink
Bithumb: Add subscription configuration (#1618)
Browse files Browse the repository at this point in the history
  • Loading branch information
gbjk authored Dec 5, 2024
1 parent b98e82d commit 6b79e4c
Show file tree
Hide file tree
Showing 4 changed files with 90 additions and 38 deletions.
81 changes: 49 additions & 32 deletions exchanges/bithumb/bithumb_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,16 @@ import (
"encoding/json"
"fmt"
"net/http"
"strings"
"text/template"
"time"

"github.com/Masterminds/sprig/v3"
"github.com/gorilla/websocket"
"github.com/thrasher-corp/gocryptotrader/common"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
"github.com/thrasher-corp/gocryptotrader/exchanges/request"
"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
Expand All @@ -24,10 +29,15 @@ const (
)

var (
wsDefaultTickTypes = []string{"30M"} // alternatives "1H", "12H", "24H", "MID"
location *time.Location
location *time.Location
)

var defaultSubscriptions = subscription.List{
{Enabled: true, Asset: asset.Spot, Channel: subscription.TickerChannel, Interval: kline.ThirtyMin}, // alternatives "1H", "12H", "24H"
{Enabled: true, Asset: asset.Spot, Channel: subscription.OrderbookChannel},
{Enabled: true, Asset: asset.Spot, Channel: subscription.AllTradesChannel},
}

// WsConnect initiates a websocket connection
func (b *Bithumb) WsConnect() error {
if !b.Websocket.IsEnabled() || !b.IsEnabled() {
Expand Down Expand Up @@ -171,41 +181,19 @@ func (b *Bithumb) wsHandleData(respRaw []byte) error {

// generateSubscriptions generates the default subscription set
func (b *Bithumb) generateSubscriptions() (subscription.List, error) {
var channels = []string{"ticker", "transaction", "orderbookdepth"}
var subscriptions subscription.List
pairs, err := b.GetEnabledPairs(asset.Spot)
if err != nil {
return nil, err
}
return b.Features.Subscriptions.ExpandTemplates(b)
}

pFmt, err := b.GetPairFormat(asset.Spot, true)
if err != nil {
return nil, err
}
pairs = pairs.Format(pFmt)

for y := range channels {
subscriptions = append(subscriptions, &subscription.Subscription{
Channel: channels[y],
Pairs: pairs,
Asset: asset.Spot,
})
}
return subscriptions, nil
// GetSubscriptionTemplate returns a subscription channel template
func (b *Bithumb) GetSubscriptionTemplate(_ *subscription.Subscription) (*template.Template, error) {
return template.New("master.tmpl").Funcs(sprig.FuncMap()).Funcs(template.FuncMap{"subToReq": subToReq}).Parse(subTplText)
}

// Subscribe subscribes to a set of channels
func (b *Bithumb) Subscribe(channelsToSubscribe subscription.List) error {
func (b *Bithumb) Subscribe(subs subscription.List) error {
var errs error
for _, s := range channelsToSubscribe {
req := &WsSubscribe{
Type: s.Channel,
Symbols: s.Pairs,
}
if s.Channel == "ticker" {
req.TickTypes = wsDefaultTickTypes
}
err := b.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, req)
for _, s := range subs {
err := b.Websocket.Conn.SendJSONMessage(context.TODO(), request.Unset, json.RawMessage(s.QualifiedChannel))
if err == nil {
err = b.Websocket.AddSuccessfulSubscriptions(b.Websocket.Conn, s)
}
Expand All @@ -215,3 +203,32 @@ func (b *Bithumb) Subscribe(channelsToSubscribe subscription.List) error {
}
return errs
}

// subToReq returns the subscription as a map to populate WsSubscribe
func subToReq(s *subscription.Subscription, p currency.Pairs) *WsSubscribe {
req := &WsSubscribe{
Type: s.Channel,
Symbols: common.SortStrings(p),
}
switch s.Channel {
case subscription.TickerChannel:
// As-is
case subscription.OrderbookChannel:
req.Type = "orderbookdepth"
case subscription.AllTradesChannel:
req.Type = "transaction"
default:
panic(fmt.Errorf("%w: %s", subscription.ErrNotSupported, s.Channel))
}
if s.Interval > 0 {
req.TickTypes = []string{strings.ToUpper(s.Interval.Short())}
}
return req
}

const subTplText = `
{{ range $asset, $pairs := $.AssetPairs }}
{{- subToReq $.S $pairs | mustToJson }}
{{- $.AssetSeparator }}
{{- end }}
`
45 changes: 39 additions & 6 deletions exchanges/bithumb/bithumb_websocket_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,17 @@ import (
"errors"
"testing"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/thrasher-corp/gocryptotrader/currency"
exchange "github.com/thrasher-corp/gocryptotrader/exchanges"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/kline"
"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
"github.com/thrasher-corp/gocryptotrader/exchanges/subscription"
"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
testexch "github.com/thrasher-corp/gocryptotrader/internal/testing/exchange"
testsubs "github.com/thrasher-corp/gocryptotrader/internal/testing/subscriptions"
)

var (
Expand Down Expand Up @@ -87,13 +93,40 @@ func TestWsHandleData(t *testing.T) {
}
}

func TestSubToReq(t *testing.T) {
t.Parallel()
p := currency.Pairs{currency.NewPairWithDelimiter("BTC", "KRW", "_"), currency.NewPairWithDelimiter("ETH", "KRW", "_")}
r := subToReq(&subscription.Subscription{Channel: subscription.AllTradesChannel}, p)
assert.Equal(t, "transaction", r.Type)
assert.True(t, p.Equal(r.Symbols))
r = subToReq(&subscription.Subscription{Channel: subscription.OrderbookChannel}, p)
assert.Equal(t, "orderbookdepth", r.Type)
assert.True(t, p.Equal(r.Symbols))
r = subToReq(&subscription.Subscription{Channel: subscription.TickerChannel, Interval: kline.OneHour}, p)
assert.Equal(t, "ticker", r.Type)
assert.True(t, p.Equal(r.Symbols))
assert.Equal(t, []string{"1H"}, r.TickTypes)
assert.PanicsWithError(t,
"subscription channel not supported: myTrades",
func() { subToReq(&subscription.Subscription{Channel: subscription.MyTradesChannel}, p) },
"should panic on invalid channel",
)
}

func TestGenerateSubscriptions(t *testing.T) {
t.Parallel()
sub, err := b.generateSubscriptions()
if err != nil {
t.Fatal(err)
}
if sub == nil {
t.Fatal("unexpected value")
b := new(Bithumb)
require.NoError(t, testexch.Setup(b), "Test instance Setup must not error")
p := currency.Pairs{currency.NewPairWithDelimiter("BTC", "KRW", "_"), currency.NewPairWithDelimiter("ETH", "KRW", "_")}
require.NoError(t, b.CurrencyPairs.StorePairs(asset.Spot, p, false))
require.NoError(t, b.CurrencyPairs.StorePairs(asset.Spot, p, true))
subs, err := b.generateSubscriptions()
require.NoError(t, err)
exp := subscription.List{
{Asset: asset.Spot, Channel: subscription.AllTradesChannel, Pairs: p, QualifiedChannel: `{"type":"transaction","symbols":["BTC_KRW","ETH_KRW"]}`},
{Asset: asset.Spot, Channel: subscription.OrderbookChannel, Pairs: p, QualifiedChannel: `{"type":"orderbookdepth","symbols":["BTC_KRW","ETH_KRW"]}`},
{Asset: asset.Spot, Channel: subscription.TickerChannel, Pairs: p, Interval: kline.ThirtyMin,
QualifiedChannel: `{"type":"ticker","symbols":["BTC_KRW","ETH_KRW"],"tickTypes":["30M"]}`},
}
testsubs.EqualLists(t, exp, subs)
}
1 change: 1 addition & 0 deletions exchanges/bithumb/bithumb_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ func (b *Bithumb) SetDefaults() {
GlobalResultLimit: 1500,
},
},
Subscriptions: defaultSubscriptions.Clone(),
}
b.Requester, err = request.New(b.Name,
common.NewHTTPClientWithTimeout(exchange.DefaultHTTPTimeout),
Expand Down
1 change: 1 addition & 0 deletions exchanges/subscription/subscription.go
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ var (
ErrInvalidState = errors.New("invalid subscription state")
ErrDuplicate = errors.New("duplicate subscription")
ErrUseConstChannelName = errors.New("must use standard channel name constants")
ErrNotSupported = errors.New("subscription channel not supported")
)

// State tracks the status of a subscription channel
Expand Down

0 comments on commit 6b79e4c

Please sign in to comment.