Skip to content

Commit

Permalink
Websocket: Various refactors and test improvements (thrasher-corp#1466)
Browse files Browse the repository at this point in the history
* Websocket: Remove IsInit and simplify SetProxyAddress

IsInit was basically the same as IsConnected.
Any time Connect was called both would be set to true.
Any time we had a disconnect they'd both be set to false
Shutdown() incorrectly didn't setInit(false)

SetProxyAddress simplified to only reconnect a connected Websocket.
Any other state means it hasn't been Connected, or it's about to
reconnect anyway.
There's no handling for IsConnecting previously, either, so I've wrapped
that behind the main mutex.

* Websocket: Expand and Assertify tests

* Websocket: Simplify state transistions

* Websocket: Simplify Connecting/Connected state

* Websocket: Tests and errors for websocket

* Websocket: Make WebsocketNotEnabled a real error

This allows for testing and avoids the repetition.
If each returned error is a error.New() you can never use errors.Is()

* Websocket: Add more testable errors

* Websocket: Improve GenerateMessageID test

Testing just the last id doesn't feel very robust

* Websocket: Protect Setup() from races

* Websocket: Use atomics instead of mutex

This was spurred by looking at the setState call in trafficMonitor and
the effect on blocking and efficiency.
With the new atomic types in Go 1.19, and the small types in use here,
atomics should be safe for our usage. bools should be truly atomic,
and uint32 is atomic when the accepted value range is less than one byte/uint8 since
that can be written atomicly by concurrent processors.
Maybe that's not even a factor any more, however we don't even have to worry enough to check.

* Websocket: Fix and simplify traffic monitor

trafficMonitor had a check throttle at the end of the for loop to stop it just gobbling the (blocking) trafficAlert channel non-stop.
That makes sense, except that nothing is sent to the trafficAlert channel if there's no listener.
So that means that it's out by one second on the trafficAlert, because any traffic received during the pause is doesn't try to send a traffic alert.

The unstopped timer is deliberately leaked for later GC when shutdown.
It won't delay/block anything, and it's a trivial memory leak during an infrequent event.

Deliberately Choosing to recreate the timer each time instead of using Stop, drain and reset

* Websocket: Split traficMonitor test on behaviours

* Websocket: Remove trafficMonitor connected status

trafficMonitor does not need to set the connection to be connected.
Connect() does that. Anything after that should result in a full
shutdown and restart. It can't and shouldn't become connected
unexpectedly, and this is most likely a race anyway.

Also dropped trafficCheckInterval to 100ms to mitigate races of traffic
alerts being buffered for too long.

* Websocket: Set disconnected earlier in Shutdown

This caused a possible race where state is still connected, but we start
to trigger interested actors via ShutdownC and Wait.
They may check state and then call Shutdown again, such as
trafficMonitor

* Websocket: Wait 5s for slow tests to pass traffic draining

Keep getting failures upstream on test rigs.
Think they can be very contended, so this pushes the boundary right out
to 5s
  • Loading branch information
gbjk authored Feb 23, 2024
1 parent 40193bb commit 52c6b3b
Show file tree
Hide file tree
Showing 66 changed files with 573 additions and 861 deletions.
2 changes: 1 addition & 1 deletion cmd/exchange_template/wrapper_file.tmpl
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ func ({{.Variable}} *{{.CapitalName}}) SetDefaults() {
exchange.RestSpot: {{.Name}}APIURL,
// exchange.WebsocketSpot: {{.Name}}WSAPIURL,
})
{{.Variable}}.Websocket = stream.New()
{{.Variable}}.Websocket = stream.NewWebsocket()
{{.Variable}}.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
{{.Variable}}.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
{{.Variable}}.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 1 addition & 1 deletion engine/websocketroutine_manager_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -293,7 +293,7 @@ func TestRegisterWebsocketDataHandlerWithFunctionality(t *testing.T) {
t.Fatal("unexpected data handlers registered")
}

mock := stream.New()
mock := stream.NewWebsocket()
mock.ToRoutine = make(chan interface{})
m.state = readyState
err = m.websocketDataReceiver(mock)
Expand Down
2 changes: 1 addition & 1 deletion exchanges/binance/binance_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ var (
// WsConnect initiates a websocket connection
func (b *Binance) WsConnect() error {
if !b.Websocket.IsEnabled() || !b.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}

var dialer websocket.Dialer
Expand Down
2 changes: 1 addition & 1 deletion exchanges/binance/binance_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -238,7 +238,7 @@ func (b *Binance) SetDefaults() {
log.Errorln(log.ExchangeSys, err)
}

b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
b.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
b.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
}
Expand Down
2 changes: 1 addition & 1 deletion exchanges/binanceus/binanceus_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ var (
// WsConnect initiates a websocket connection
func (bi *Binanceus) WsConnect() error {
if !bi.Websocket.IsEnabled() || !bi.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
dialer.HandshakeTimeout = bi.Config.HTTPTimeout
Expand Down
2 changes: 1 addition & 1 deletion exchanges/binanceus/binanceus_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,7 @@ func (bi *Binanceus) SetDefaults() {
"%s setting default endpoints error %v",
bi.Name, err)
}
bi.Websocket = stream.New()
bi.Websocket = stream.NewWebsocket()
bi.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
bi.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
bi.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bitfinex/bitfinex_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1128,7 +1128,7 @@ func TestGetDepositAddress(t *testing.T) {
// TestWsAuth dials websocket, sends login request.
func TestWsAuth(t *testing.T) {
if !b.Websocket.IsEnabled() {
t.Skip(stream.WebsocketNotEnabled)
t.Skip(stream.ErrWebsocketNotEnabled.Error())
}
sharedtestvalues.SkipTestIfCredentialsUnset(t, b)
if !b.API.AuthenticatedWebsocketSupport {
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bitfinex/bitfinex_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ var cMtx sync.Mutex
// WsConnect starts a new websocket connection
func (b *Bitfinex) WsConnect() error {
if !b.Websocket.IsEnabled() || !b.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
err := b.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bitfinex/bitfinex_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -198,7 +198,7 @@ func (b *Bitfinex) SetDefaults() {
if err != nil {
log.Errorln(log.ExchangeSys, err)
}
b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
b.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
b.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
b.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
3 changes: 1 addition & 2 deletions exchanges/bithumb/bithumb_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ package bithumb

import (
"encoding/json"
"errors"
"fmt"
"net/http"
"time"
Expand All @@ -29,7 +28,7 @@ var (
// WsConnect initiates a websocket connection
func (b *Bithumb) WsConnect() error {
if !b.Websocket.IsEnabled() || !b.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}

var dialer websocket.Dialer
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bithumb/bithumb_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ func (b *Bithumb) SetDefaults() {
log.Errorln(log.ExchangeSys, err)
}

b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
b.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
b.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
}
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bitmex/bitmex_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -789,7 +789,7 @@ func TestGetDepositAddress(t *testing.T) {
func TestWsAuth(t *testing.T) {
t.Parallel()
if !b.Websocket.IsEnabled() && !b.API.AuthenticatedWebsocketSupport || !sharedtestvalues.AreAPICredentialsSet(b) {
t.Skip(stream.WebsocketNotEnabled)
t.Skip(stream.ErrWebsocketNotEnabled.Error())
}
var dialer websocket.Dialer
err := b.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bitmex/bitmex_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ const (
// WsConnect initiates a new websocket connection
func (b *Bitmex) WsConnect() error {
if !b.Websocket.IsEnabled() || !b.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
err := b.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bitmex/bitmex_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -175,7 +175,7 @@ func (b *Bitmex) SetDefaults() {
if err != nil {
log.Errorln(log.ExchangeSys, err)
}
b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
b.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
b.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
b.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bitstamp/bitstamp_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ var (
// WsConnect connects to a websocket feed
func (b *Bitstamp) WsConnect() error {
if !b.Websocket.IsEnabled() || !b.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
err := b.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bitstamp/bitstamp_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ func (b *Bitstamp) SetDefaults() {
if err != nil {
log.Errorln(log.ExchangeSys, err)
}
b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
b.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
b.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
b.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 1 addition & 1 deletion exchanges/btcmarkets/btcmarkets_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ var (
// WsConnect connects to a websocket feed
func (b *BTCMarkets) WsConnect() error {
if !b.Websocket.IsEnabled() || !b.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
err := b.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/btcmarkets/btcmarkets_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ func (b *BTCMarkets) SetDefaults() {
if err != nil {
log.Errorln(log.ExchangeSys, err)
}
b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
b.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
b.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
b.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 1 addition & 1 deletion exchanges/btse/btse_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ const (
// WsConnect connects the websocket client
func (b *BTSE) WsConnect() error {
if !b.Websocket.IsEnabled() || !b.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
err := b.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/btse/btse_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -176,7 +176,7 @@ func (b *BTSE) SetDefaults() {
if err != nil {
log.Errorln(log.ExchangeSys, err)
}
b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
b.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
b.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
b.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 0 additions & 2 deletions exchanges/bybit/bybit.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"github.com/thrasher-corp/gocryptotrader/exchanges/order"
"github.com/thrasher-corp/gocryptotrader/exchanges/orderbook"
"github.com/thrasher-corp/gocryptotrader/exchanges/request"
"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
)

// Bybit is the overarching type across this package
Expand Down Expand Up @@ -90,7 +89,6 @@ var (
errAPIKeyIsNotUnified = errors.New("api key is not unified")
errEndpointAvailableForNormalAPIKeyHolders = errors.New("endpoint available for normal API key holders only")
errInvalidContractLength = errors.New("contract length cannot be less than or equal to zero")
errWebsocketNotEnabled = errors.New(stream.WebsocketNotEnabled)
)

var (
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bybit/bybit_inverse_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ import (
// WsInverseConnect connects to inverse websocket feed
func (by *Bybit) WsInverseConnect() error {
if !by.Websocket.IsEnabled() || !by.IsEnabled() || !by.IsAssetWebsocketSupported(asset.CoinMarginedFutures) {
return errWebsocketNotEnabled
return stream.ErrWebsocketNotEnabled
}
by.Websocket.Conn.SetURL(inversePublic)
var dialer websocket.Dialer
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bybit/bybit_linear_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ import (
// WsLinearConnect connects to linear a websocket feed
func (by *Bybit) WsLinearConnect() error {
if !by.Websocket.IsEnabled() || !by.IsEnabled() || !by.IsAssetWebsocketSupported(asset.LinearContract) {
return errWebsocketNotEnabled
return stream.ErrWebsocketNotEnabled
}
by.Websocket.Conn.SetURL(linearPublic)
var dialer websocket.Dialer
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bybit/bybit_options_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ import (
// WsOptionsConnect connects to options a websocket feed
func (by *Bybit) WsOptionsConnect() error {
if !by.Websocket.IsEnabled() || !by.IsEnabled() || !by.IsAssetWebsocketSupported(asset.Options) {
return errWebsocketNotEnabled
return stream.ErrWebsocketNotEnabled
}
by.Websocket.Conn.SetURL(optionPublic)
var dialer websocket.Dialer
Expand Down
7 changes: 4 additions & 3 deletions exchanges/bybit/bybit_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"github.com/thrasher-corp/gocryptotrader/exchanges/margin"
"github.com/thrasher-corp/gocryptotrader/exchanges/order"
"github.com/thrasher-corp/gocryptotrader/exchanges/sharedtestvalues"
"github.com/thrasher-corp/gocryptotrader/exchanges/stream"
"github.com/thrasher-corp/gocryptotrader/portfolio/withdraw"
)

Expand Down Expand Up @@ -3064,7 +3065,7 @@ func TestWsLinearConnect(t *testing.T) {
t.Skip(skippingWebsocketFunctionsForMockTesting)
}
err := b.WsLinearConnect()
if err != nil && !errors.Is(err, errWebsocketNotEnabled) {
if err != nil && !errors.Is(err, stream.ErrWebsocketNotEnabled) {
t.Error(err)
}
}
Expand All @@ -3074,7 +3075,7 @@ func TestWsInverseConnect(t *testing.T) {
t.Skip(skippingWebsocketFunctionsForMockTesting)
}
err := b.WsInverseConnect()
if err != nil && !errors.Is(err, errWebsocketNotEnabled) {
if err != nil && !errors.Is(err, stream.ErrWebsocketNotEnabled) {
t.Error(err)
}
}
Expand All @@ -3084,7 +3085,7 @@ func TestWsOptionsConnect(t *testing.T) {
t.Skip(skippingWebsocketFunctionsForMockTesting)
}
err := b.WsOptionsConnect()
if err != nil && !errors.Is(err, errWebsocketNotEnabled) {
if err != nil && !errors.Is(err, stream.ErrWebsocketNotEnabled) {
t.Error(err)
}
}
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bybit/bybit_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ const (
// WsConnect connects to a websocket feed
func (by *Bybit) WsConnect() error {
if !by.Websocket.IsEnabled() || !by.IsEnabled() || !by.IsAssetWebsocketSupported(asset.Spot) {
return errWebsocketNotEnabled
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
err := by.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/bybit/bybit_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -216,7 +216,7 @@ func (by *Bybit) SetDefaults() {
log.Errorln(log.ExchangeSys, err)
}

by.Websocket = stream.New()
by.Websocket = stream.NewWebsocket()
by.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
by.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
by.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 1 addition & 1 deletion exchanges/coinbasepro/coinbasepro_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -681,7 +681,7 @@ func TestGetDepositAddress(t *testing.T) {
// TestWsAuth dials websocket, sends login request.
func TestWsAuth(t *testing.T) {
if !c.Websocket.IsEnabled() && !c.API.AuthenticatedWebsocketSupport || !sharedtestvalues.AreAPICredentialsSet(c) {
t.Skip(stream.WebsocketNotEnabled)
t.Skip(stream.ErrWebsocketNotEnabled.Error())
}
var dialer websocket.Dialer
err := c.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/coinbasepro/coinbasepro_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ const (
// WsConnect initiates a websocket connection
func (c *CoinbasePro) WsConnect() error {
if !c.Websocket.IsEnabled() || !c.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
err := c.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/coinbasepro/coinbasepro_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ func (c *CoinbasePro) SetDefaults() {
if err != nil {
log.Errorln(log.ExchangeSys, err)
}
c.Websocket = stream.New()
c.Websocket = stream.NewWebsocket()
c.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
c.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
c.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 1 addition & 1 deletion exchanges/coinut/coinut_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ func setupWSTestAuth(t *testing.T) {
}

if !c.Websocket.IsEnabled() && !c.API.AuthenticatedWebsocketSupport || !sharedtestvalues.AreAPICredentialsSet(c) {
t.Skip(stream.WebsocketNotEnabled)
t.Skip(stream.ErrWebsocketNotEnabled.Error())
}
if sharedtestvalues.AreAPICredentialsSet(c) {
c.Websocket.SetCanUseAuthenticatedEndpoints(true)
Expand Down
2 changes: 1 addition & 1 deletion exchanges/coinut/coinut_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ var (
// WsConnect initiates a websocket connection
func (c *COINUT) WsConnect() error {
if !c.Websocket.IsEnabled() || !c.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
var dialer websocket.Dialer
err := c.Websocket.Conn.Dial(&dialer, http.Header{})
Expand Down
2 changes: 1 addition & 1 deletion exchanges/coinut/coinut_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ func (c *COINUT) SetDefaults() {
if err != nil {
log.Errorln(log.ExchangeSys, err)
}
c.Websocket = stream.New()
c.Websocket = stream.NewWebsocket()
c.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
c.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
c.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
6 changes: 3 additions & 3 deletions exchanges/exchange_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -198,7 +198,7 @@ func TestSetClientProxyAddress(t *testing.T) {
Name: "rawr",
Requester: requester}

newBase.Websocket = stream.New()
newBase.Websocket = stream.NewWebsocket()
err = newBase.SetClientProxyAddress("")
if err != nil {
t.Error(err)
Expand Down Expand Up @@ -1251,7 +1251,7 @@ func TestSetupDefaults(t *testing.T) {
}

// Test websocket support
b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
b.Features.Supports.Websocket = true
err = b.Websocket.Setup(&stream.WebsocketSetup{
ExchangeConfig: &config.Exchange{
Expand Down Expand Up @@ -1596,7 +1596,7 @@ func TestIsWebsocketEnabled(t *testing.T) {
t.Error("exchange doesn't support websocket")
}

b.Websocket = stream.New()
b.Websocket = stream.NewWebsocket()
err := b.Websocket.Setup(&stream.WebsocketSetup{
ExchangeConfig: &config.Exchange{
Enabled: true,
Expand Down
2 changes: 1 addition & 1 deletion exchanges/gateio/gateio_websocket.go
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,7 @@ var fetchedCurrencyPairSnapshotOrderbook = make(map[string]bool)
// WsConnect initiates a websocket connection
func (g *Gateio) WsConnect() error {
if !g.Websocket.IsEnabled() || !g.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
err := g.CurrencyPairs.IsAssetEnabled(asset.Spot)
if err != nil {
Expand Down
2 changes: 1 addition & 1 deletion exchanges/gateio/gateio_wrapper.go
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,7 @@ func (g *Gateio) SetDefaults() {
if err != nil {
log.Errorln(log.ExchangeSys, err)
}
g.Websocket = stream.New()
g.Websocket = stream.NewWebsocket()
g.WebsocketResponseMaxLimit = exchange.DefaultWebsocketResponseMaxLimit
g.WebsocketResponseCheckTimeout = exchange.DefaultWebsocketResponseCheckTimeout
g.WebsocketOrderbookBufferLimit = exchange.DefaultWebsocketOrderbookBufferLimit
Expand Down
2 changes: 1 addition & 1 deletion exchanges/gateio/gateio_ws_delivery_futures.go
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ var fetchedFuturesCurrencyPairSnapshotOrderbook = make(map[string]bool)
// WsDeliveryFuturesConnect initiates a websocket connection for delivery futures account
func (g *Gateio) WsDeliveryFuturesConnect() error {
if !g.Websocket.IsEnabled() || !g.IsEnabled() {
return errors.New(stream.WebsocketNotEnabled)
return stream.ErrWebsocketNotEnabled
}
err := g.CurrencyPairs.IsAssetEnabled(asset.DeliveryFutures)
if err != nil {
Expand Down
Loading

0 comments on commit 52c6b3b

Please sign in to comment.