d0zingcat/gocryptotrader
1
0
Fork 0
mirror of https://github.com/d0zingcat/gocryptotrader.git synced 2026-05-13 23:16:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

stream/websocket: Consolidate fields by using exchange config pointer (#809)

Browse Source 
* stream: add exchange config pointer to setup WebsocketSetup struct to reduce and consolidate setting of variables.

* config: reduce stutter

* config: reduce minor stutter

* glorious: nits addr.

* Update exchanges/stream/websocket.go

Co-authored-by: Scott <gloriousCode@users.noreply.github.com>

* websocket: implement fix

* engine/helpers: fix test

* exchanges: fix after merge issues

* exchange_template: fix output

Co-authored-by: Scott <gloriousCode@users.noreply.github.com>
This commit is contained in:
Ryan O'Hara-Reid
2021-10-20 15:45:06 +11:00
committed by GitHub
parent a70224d123
commit 099ffa1a60
53 changed files with 611 additions and 692 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
exchanges/stream/buffer/buffer.go
View File

@@ -6,6 +6,7 @@ import (
"sort"
"time"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
@@ -15,7 +16,7 @@ import (
const packageError = "websocket orderbook buffer error: %w"
var (
errUnsetExchangeName = errors.New("exchange name unset")
errExchangeConfigNil = errors.New("exchange config is nil")
errUnsetDataHandler = errors.New("datahandler unset")
errIssueBufferEnabledButNoLimit = errors.New("buffer enabled but no limit set")
errUpdateIsNil = errors.New("update is nil")
@@ -25,34 +26,35 @@ var (
)
// Setup sets private variables
func (w *Orderbook) Setup(obBufferLimit int,
bufferEnabled,
sortBuffer,
sortBufferByUpdateIDs,
updateEntriesByID,
verbose bool,
publishPeriod time.Duration,
exchangeName string,
dataHandler chan interface{}) error {
if exchangeName == "" {
return fmt.Errorf(packageError, errUnsetExchangeName)
func (w *Orderbook) Setup(cfg *config.Exchange, sortBuffer, sortBufferByUpdateIDs, updateEntriesByID bool, dataHandler chan interface{}) error {
if cfg == nil { // exchange config fields are checked in stream package
// prior to calling this, so further checks are not needed.
return fmt.Errorf(packageError, errExchangeConfigNil)
}
if dataHandler == nil {
return fmt.Errorf(packageError, errUnsetDataHandler)
}
if bufferEnabled && obBufferLimit < 1 {
if cfg.Orderbook.WebsocketBufferEnabled &&
cfg.Orderbook.WebsocketBufferLimit < 1 {
return fmt.Errorf(packageError, errIssueBufferEnabledButNoLimit)
}
w.obBufferLimit = obBufferLimit
w.bufferEnabled = bufferEnabled
w.bufferEnabled = cfg.Orderbook.WebsocketBufferEnabled
w.obBufferLimit = cfg.Orderbook.WebsocketBufferLimit
w.sortBuffer = sortBuffer
w.sortBufferByUpdateIDs = sortBufferByUpdateIDs
w.updateEntriesByID = updateEntriesByID
w.exchangeName = exchangeName
w.exchangeName = cfg.Name
w.dataHandler = dataHandler
w.ob = make(map[currency.Code]map[currency.Code]map[asset.Item]*orderbookHolder)
w.verbose = verbose
w.publishPeriod = publishPeriod
w.verbose = cfg.Verbose
// set default publish period if missing
orderbookPublishPeriod := config.DefaultOrderbookPublishPeriod
if cfg.Orderbook.PublishPeriod != nil {
orderbookPublishPeriod = *cfg.Orderbook.PublishPeriod
}
w.publishPeriod = orderbookPublishPeriod
return nil
}

40
exchanges/stream/buffer/buffer_test.go
View File

@@ -6,21 +6,24 @@ import (
"testing"
"time"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
)
var itemArray = [][]orderbook.Item{
{{Price: 1000, Amount: 1, ID: 1000}},
{{Price: 2000, Amount: 1, ID: 2000}},
{{Price: 3000, Amount: 1, ID: 3000}},
{{Price: 3000, Amount: 2, ID: 4000}},
{{Price: 4000, Amount: 0, ID: 6000}},
{{Price: 5000, Amount: 1, ID: 5000}},
}
var (
itemArray = [][]orderbook.Item{
{{Price: 1000, Amount: 1, ID: 1000}},
{{Price: 2000, Amount: 1, ID: 2000}},
{{Price: 3000, Amount: 1, ID: 3000}},
{{Price: 3000, Amount: 2, ID: 4000}},
{{Price: 4000, Amount: 0, ID: 6000}},
{{Price: 5000, Amount: 1, ID: 5000}},
}
var cp, _ = currency.NewPairFromString("BTCUSD")
cp, _ = currency.NewPairFromString("BTCUSD")
)
const (
exchangeName = "exchangeTest"
@@ -712,22 +715,27 @@ func TestGetOrderbook(t *testing.T) {
func TestSetup(t *testing.T) {
t.Parallel()
w := Orderbook{}
err := w.Setup(0, false, false, false, false, true, 0, "", nil)
if !errors.Is(err, errUnsetExchangeName) {
t.Fatalf("expected error %v but received %v", errUnsetExchangeName, err)
err := w.Setup(nil, false, false, false, nil)
if !errors.Is(err, errExchangeConfigNil) {
t.Fatalf("expected error %v but received %v", errExchangeConfigNil, err)
}
err = w.Setup(0, false, false, false, false, false, 0, "test", nil)
exchangeConfig := &config.Exchange{}
err = w.Setup(exchangeConfig, false, false, false, nil)
if !errors.Is(err, errUnsetDataHandler) {
t.Fatalf("expected error %v but received %v", errUnsetDataHandler, err)
}
err = w.Setup(0, true, false, false, false, true, 0, "test", make(chan interface{}))
exchangeConfig.Orderbook.WebsocketBufferEnabled = true
err = w.Setup(exchangeConfig, false, false, false, make(chan interface{}))
if !errors.Is(err, errIssueBufferEnabledButNoLimit) {
t.Fatalf("expected error %v but received %v", errIssueBufferEnabledButNoLimit, err)
}
err = w.Setup(1337, true, true, true, true, false, 0, "test", make(chan interface{}))
exchangeConfig.Orderbook.WebsocketBufferLimit = 1337
exchangeConfig.Orderbook.WebsocketBufferEnabled = true
exchangeConfig.Name = "test"
err = w.Setup(exchangeConfig, true, true, true, make(chan interface{}))
if err != nil {
t.Fatal(err)
}
@@ -1002,7 +1010,7 @@ func TestUpdateByIDAndAction(t *testing.T) {
func TestFlushOrderbook(t *testing.T) {
t.Parallel()
w := &Orderbook{}
err := w.Setup(5, false, false, false, false, false, 0, "test", make(chan interface{}, 2))
err := w.Setup(&config.Exchange{Name: "test"}, false, false, false, make(chan interface{}, 2))
if err != nil {
t.Fatal(err)
}

106
exchanges/stream/websocket.go
View File

@@ -24,10 +24,26 @@ const (
)
var (
errClosedConnection = errors.New("use of closed network connection")
// ErrSubscriptionFailure defines an error when a subscription fails
ErrSubscriptionFailure = errors.New("subscription failure")
errAlreadyRunning = errors.New("connection monitor is already running")
errAlreadyRunning = errors.New("connection monitor is already running")
errExchangeConfigIsNil = errors.New("exchange config is nil")
errWebsocketIsNil = errors.New("websocket is nil")
errWebsocketSetupIsNil = errors.New("websocket setup is nil")
errWebsocketAlreadyInitialised = errors.New("websocket already initialised")
errWebsocketFeaturesIsUnset = errors.New("websocket features is unset")
errConfigFeaturesIsNil = errors.New("exchange config features is nil")
errDefaultURLIsEmpty = errors.New("default url is empty")
errRunningURLIsEmpty = errors.New("running url cannot be empty")
errInvalidWebsocketURL = errors.New("invalid websocket url")
errExchangeConfigNameUnset = errors.New("exchange config name unset")
errInvalidTrafficTimeout = errors.New("invalid traffic timeout")
errWebsocketSubscriberUnset = errors.New("websocket subscriber function needs to be set")
errWebsocketUnsubscriberUnset = errors.New("websocket unsubscriber functionality allowed but unsubscriber function not set")
errWebsocketConnectorUnset = errors.New("websocket connector function not set")
errWebsocketSubscriptionsGeneratorUnset = errors.New("websocket subscriptions generator function needs to be set")
errClosedConnection = errors.New("use of closed network connection")
)
// New initialises the websocket struct
@@ -44,97 +60,95 @@ func New() *Websocket {
}
}
var (
errSubscriberUnset = errors.New("subscriber function needs to be set")
errGenerateSubsciptionsUnset = errors.New("generate subscriptions function needs to be set")
)
// Setup sets main variables for websocket connection
func (w *Websocket) Setup(s *WebsocketSetup) error {
if w == nil {
return errors.New("websocket is nil")
return errWebsocketIsNil
}
if s == nil {
return errors.New("websocket setup is nil")
return errWebsocketSetupIsNil
}
if !w.Init {
return fmt.Errorf("%s Websocket already initialised",
s.ExchangeName)
return fmt.Errorf("%s %w", w.exchangeName, errWebsocketAlreadyInitialised)
}
w.verbose = s.Verbose
if s.ExchangeConfig == nil {
return errExchangeConfigIsNil
}
if s.ExchangeConfig.Name == "" {
return errExchangeConfigNameUnset
}
w.exchangeName = s.ExchangeConfig.Name
w.verbose = s.ExchangeConfig.Verbose
if s.Features == nil {
return errors.New("websocket features is unset")
return fmt.Errorf("%s %w", w.exchangeName, errWebsocketFeaturesIsUnset)
}
w.features = s.Features
if s.ExchangeConfig.Features == nil {
return fmt.Errorf("%s %w", w.exchangeName, errConfigFeaturesIsNil)
}
w.enabled = s.ExchangeConfig.Features.Enabled.Websocket
if s.Connector == nil {
return fmt.Errorf("%s %w", w.exchangeName, errWebsocketConnectorUnset)
}
w.connector = s.Connector
if s.Subscriber == nil {
return errSubscriberUnset
return fmt.Errorf("%s %w", w.exchangeName, errWebsocketSubscriberUnset)
}
w.Subscriber = s.Subscriber
if w.features.Unsubscribe && s.UnSubscriber == nil {
return errors.New("features have been set yet channel unsubscriber is not set")
if w.features.Unsubscribe && s.Unsubscriber == nil {
return fmt.Errorf("%s %w", w.exchangeName, errWebsocketUnsubscriberUnset)
}
w.Unsubscriber = s.UnSubscriber
w.Unsubscriber = s.Unsubscriber
if s.GenerateSubscriptions == nil {
return errGenerateSubsciptionsUnset
return fmt.Errorf("%s %w", w.exchangeName, errWebsocketSubscriptionsGeneratorUnset)
}
w.GenerateSubs = s.GenerateSubscriptions
w.enabled = s.Enabled
if s.DefaultURL == "" {
return errors.New("default url is empty")
return fmt.Errorf("%s websocket %w", w.exchangeName, errDefaultURLIsEmpty)
}
w.defaultURL = s.DefaultURL
if s.RunningURL == "" {
return errors.New("running URL cannot be nil")
return fmt.Errorf("%s websocket %w", w.exchangeName, errRunningURLIsEmpty)
}
err := w.SetWebsocketURL(s.RunningURL, false, false)
if err != nil {
return err
return fmt.Errorf("%s %w", w.exchangeName, err)
}
if s.RunningURLAuth != "" {
err = w.SetWebsocketURL(s.RunningURLAuth, true, false)
if err != nil {
return err
return fmt.Errorf("%s %w", w.exchangeName, err)
}
}
w.connector = s.Connector
if s.ExchangeName == "" {
return errors.New("exchange name unset")
if s.ExchangeConfig.WebsocketTrafficTimeout < time.Second {
return fmt.Errorf("%s %w cannot be less than %s",
w.exchangeName,
errInvalidTrafficTimeout,
time.Second)
}
w.exchangeName = s.ExchangeName
if s.WebsocketTimeout < time.Second {
return fmt.Errorf("traffic timeout cannot be less than %s", time.Second)
}
w.trafficTimeout = s.WebsocketTimeout
w.trafficTimeout = s.ExchangeConfig.WebsocketTrafficTimeout
w.ShutdownC = make(chan struct{})
w.Wg = new(sync.WaitGroup)
w.SetCanUseAuthenticatedEndpoints(s.AuthenticatedWebsocketAPISupport)
w.SetCanUseAuthenticatedEndpoints(s.ExchangeConfig.API.AuthenticatedWebsocketSupport)
// default publish period if missing
orderbookPublishPeriod := config.DefaultOrderbookPublishPeriod
if s.OrderbookPublishPeriod != nil {
orderbookPublishPeriod = *s.OrderbookPublishPeriod
}
return w.Orderbook.Setup(s.OrderbookBufferLimit,
s.BufferEnabled,
return w.Orderbook.Setup(s.ExchangeConfig,
s.SortBuffer,
s.SortBufferByUpdateIDs,
s.UpdateEntriesByID,
s.Verbose,
orderbookPublishPeriod,
w.exchangeName,
w.DataHandler)
}
@@ -948,7 +962,7 @@ func checkWebsocketURL(s string) error {
return err
}
if u.Scheme != "ws" && u.Scheme != "wss" {
return fmt.Errorf("cannot set invalid websocket URL %s", s)
return fmt.Errorf("cannot set %w %s", errInvalidWebsocketURL, s)
}
return nil
}

162
exchanges/stream/websocket_test.go
View File

@@ -16,6 +16,7 @@ import (
"time"
"github.com/gorilla/websocket"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/currency"
"github.com/thrasher-corp/gocryptotrader/exchanges/protocol"
)
@@ -52,15 +53,21 @@ type testResponse struct {
}
var defaultSetup = &WebsocketSetup{
Enabled: true,
AuthenticatedWebsocketAPISupport: true,
WebsocketTimeout: time.Second * 5,
DefaultURL: "testDefaultURL",
ExchangeName: "exchangeName",
RunningURL: "wss://testRunningURL",
Connector: func() error { return nil },
Subscriber: func(_ []ChannelSubscription) error { return nil },
UnSubscriber: func(_ []ChannelSubscription) error { return nil },
ExchangeConfig: &config.Exchange{
Features: &config.FeaturesConfig{
Enabled: config.FeaturesEnabledConfig{Websocket: true},
},
API: config.APIConfig{
AuthenticatedWebsocketSupport: true,
},
WebsocketTrafficTimeout: time.Second * 5,
Name: "exchangeName",
},
DefaultURL: "testDefaultURL",
RunningURL: "wss://testRunningURL",
Connector: func() error { return nil },
Subscriber: func(_ []ChannelSubscription) error { return nil },
Unsubscriber: func(_ []ChannelSubscription) error { return nil },
GenerateSubscriptions: func() ([]ChannelSubscription, error) {
return []ChannelSubscription{
{Channel: "TestSub"},
@@ -90,75 +97,104 @@ func TestSetup(t *testing.T) {
t.Parallel()
var w *Websocket
err := w.Setup(nil)
if err == nil {
t.Fatal("error cannot be nil")
if !errors.Is(err, errWebsocketIsNil) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketIsNil)
}
w = &Websocket{DataHandler: make(chan interface{})}
err = w.Setup(nil)
if err == nil {
t.Fatal("error cannot be nil")
if !errors.Is(err, errWebsocketSetupIsNil) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketSetupIsNil)
}
w.Init = true
websocketSetup := &WebsocketSetup{}
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
if !errors.Is(err, errWebsocketAlreadyInitialised) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketAlreadyInitialised)
}
w.Init = true
err = w.Setup(websocketSetup)
if !errors.Is(err, errExchangeConfigIsNil) {
t.Fatalf("received: '%v' but expected: '%v'", err, errExchangeConfigIsNil)
}
websocketSetup.ExchangeConfig = &config.Exchange{}
err = w.Setup(websocketSetup)
if !errors.Is(err, errExchangeConfigNameUnset) {
t.Fatalf("received: '%v' but expected: '%v'", err, errExchangeConfigNameUnset)
}
websocketSetup.ExchangeConfig.Name = "testname"
err = w.Setup(websocketSetup)
if !errors.Is(err, errWebsocketFeaturesIsUnset) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketFeaturesIsUnset)
}
websocketSetup.Features = &protocol.Features{}
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
if !errors.Is(err, errConfigFeaturesIsNil) {
t.Fatalf("received: '%v' but expected: '%v'", err, errConfigFeaturesIsNil)
}
websocketSetup.Features.Subscribe = true
websocketSetup.ExchangeConfig.Features = &config.FeaturesConfig{}
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
if !errors.Is(err, errWebsocketConnectorUnset) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketConnectorUnset)
}
websocketSetup.Connector = func() error { return nil }
err = w.Setup(websocketSetup)
if !errors.Is(err, errWebsocketSubscriberUnset) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketSubscriberUnset)
}
websocketSetup.Subscriber = func([]ChannelSubscription) error { return nil }
websocketSetup.Features.Unsubscribe = true
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
if !errors.Is(err, errWebsocketUnsubscriberUnset) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketUnsubscriberUnset)
}
websocketSetup.UnSubscriber = func([]ChannelSubscription) error { return nil }
websocketSetup.Unsubscriber = func([]ChannelSubscription) error { return nil }
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
}
websocketSetup.DefaultURL = "test"
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
}
websocketSetup.RunningURL = "http://www.google.com"
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
}
websocketSetup.RunningURL = "wss://www.google.com"
websocketSetup.RunningURLAuth = "http://www.google.com"
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
}
websocketSetup.RunningURLAuth = "wss://www.google.com"
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
}
websocketSetup.ExchangeName = "testname"
err = w.Setup(websocketSetup)
if err == nil {
t.Fatal("error cannot be nil")
}
websocketSetup.WebsocketTimeout = time.Minute
err = w.Setup(websocketSetup)
if !errors.Is(err, errGenerateSubsciptionsUnset) {
t.Fatalf("received: %v but expected: %v", err, errGenerateSubsciptionsUnset)
if !errors.Is(err, errWebsocketSubscriptionsGeneratorUnset) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketSubscriptionsGeneratorUnset)
}
websocketSetup.GenerateSubscriptions = func() ([]ChannelSubscription, error) { return nil, nil }
err = w.Setup(websocketSetup)
if !errors.Is(err, errDefaultURLIsEmpty) {
t.Fatalf("received: '%v' but expected: '%v'", err, errDefaultURLIsEmpty)
}
websocketSetup.DefaultURL = "test"
err = w.Setup(websocketSetup)
if !errors.Is(err, errRunningURLIsEmpty) {
t.Fatalf("received: '%v' but expected: '%v'", err, errRunningURLIsEmpty)
}
websocketSetup.RunningURL = "http://www.google.com"
err = w.Setup(websocketSetup)
if !errors.Is(err, errInvalidWebsocketURL) {
t.Fatalf("received: '%v' but expected: '%v'", err, errInvalidWebsocketURL)
}
websocketSetup.RunningURL = "wss://www.google.com"
websocketSetup.RunningURLAuth = "http://www.google.com"
err = w.Setup(websocketSetup)
if !errors.Is(err, errInvalidWebsocketURL) {
t.Fatalf("received: '%v' but expected: '%v'", err, errInvalidWebsocketURL)
}
websocketSetup.RunningURLAuth = "wss://www.google.com"
err = w.Setup(websocketSetup)
if !errors.Is(err, errInvalidTrafficTimeout) {
t.Fatalf("received: '%v' but expected: '%v'", err, errInvalidTrafficTimeout)
}
websocketSetup.ExchangeConfig.WebsocketTrafficTimeout = time.Minute
err = w.Setup(websocketSetup)
if !errors.Is(err, nil) {
t.Fatalf("received: %v but expected: %v", err, nil)
}
@@ -294,11 +330,15 @@ func TestWebsocket(t *testing.T) {
t.Parallel()
wsInit := Websocket{}
err := wsInit.Setup(&WebsocketSetup{
ExchangeName: "test",
Enabled: true,
ExchangeConfig: &config.Exchange{
Features: &config.FeaturesConfig{
Enabled: config.FeaturesEnabledConfig{Websocket: true},
},
Name: "test",
},
})
if err != nil && err.Error() != "test Websocket already initialised" {
t.Errorf("Expected 'test Websocket already initialised', received %v", err)
if !errors.Is(err, errWebsocketAlreadyInitialised) {
t.Fatalf("received: '%v' but expected: '%v'", err, errWebsocketAlreadyInitialised)
}
ws := *New()

28
exchanges/stream/websocket_types.go
View File

@@ -5,6 +5,7 @@ import (
"time"
"github.com/gorilla/websocket"
"github.com/thrasher-corp/gocryptotrader/config"
"github.com/thrasher-corp/gocryptotrader/exchanges/protocol"
"github.com/thrasher-corp/gocryptotrader/exchanges/stream/buffer"
)
@@ -86,28 +87,19 @@ type Websocket struct {
// WebsocketSetup defines variables for setting up a websocket connection
type WebsocketSetup struct {
Enabled bool
Verbose bool
AuthenticatedWebsocketAPISupport bool
WebsocketTimeout time.Duration
DefaultURL string
ExchangeName string
RunningURL string
RunningURLAuth string
Connector func() error
Subscriber func([]ChannelSubscription) error
UnSubscriber func([]ChannelSubscription) error
GenerateSubscriptions func() ([]ChannelSubscription, error)
Features *protocol.Features
ExchangeConfig *config.Exchange
DefaultURL string
RunningURL string
RunningURLAuth string
Connector func() error
Subscriber func([]ChannelSubscription) error
Unsubscriber func([]ChannelSubscription) error
GenerateSubscriptions func() ([]ChannelSubscription, error)
Features *protocol.Features
// Local orderbook buffer config values
OrderbookBufferLimit int
BufferEnabled bool
SortBuffer bool
SortBufferByUpdateIDs bool
UpdateEntriesByID bool
// OrderbookPublishPeriod is a pointer for the same reason as it is in `OrderbookConfig`:
// to allow distinguishing between a zeroed out value and a missing one
OrderbookPublishPeriod *time.Duration
}
// WebsocketConnection contains all the data needed to send a message to a WS