Websocket connection handling and subscription management (#297)

* Step one: Sets up  connection handler for websockets to always be connected until a shutdown event is received.
Sets up a vague subscription handler to ensure subscriptions are subscribed

* Adds support for resubscriptions for bitfinex, bitstamp, bitmex and btcc. Adds subscription params for special websocket subscription requirements. Removes subscription monitor from wait group so that it can exist despite a shutdown and continuously check

* Adds channel subscription support to bitmex, btse, coibasepro, coinut, gateio, gemini, hitbtc, huobi, hadax, kraken, okgroup, poloniex and zb

* Implements unsubscribe for bitfinex, btcc, btse, coinbasepro, gateio, gitbtc, huobi, hadax

* ManageSubscriptions now called from WSConnect and made private instead of inside individual exchanges. ManageSubscriptions can now unsubscribe. exchange_websocket_types.go now contains all exchange_websocket.go types to avoid clutter

* Adds it to websocket functionality so managesubscriptions will close when not supported

* Separates functions into testable functions to ensure logic works. Adds tests. Updates websocket setup to include verbosity (inherited from exchange). Adds no connection tolerance to fatal on failed reconnects

* More exchange_websocket tests. Updating to use pointers. Creation of equals func to make comparison easier

* Fixes okex, okcoin tests. Fixes race conditions. Removes pointer usage again.

* Adds subscribe and unsubscribe to wrappers

* Fixes deadlock. Fixes ws verbosity.

* Updates all exchanges to properly support subscription/connection feature. Also reintroduces race conditions....

* Moves connection varialbes to struct from package to allow each websocket to have their own reconnection checks. Neatens up logs

* Fixes lint/critic issues. Fixes tests. Removes unused function.

* Moves websocket ratelimiter to their own const variables. Fixes more race conditions with connecting variable

* Removes redundant subscribe functions. Ensuring only the exchange_websocket.go can manage subscriptions. Fixes debug logs to be verbose wrapped

* Fixes issue with slice copying. Re-adds okgroup default channels

* Adds nolint to append

* Adds comments and adds support for gateio auth request subscriptions

* Adds new test to ensure slices dont point to the same vars

* removes fatals. gofmt goimports

* more gofmts

* Addresses PR comments, removing empty and redundant lines

* Addresses PR comments. Ensures that writing to the websocket is single-threaded by adding a mutex to exchanges. Minimises wrapper code and moves subscription loops to exchange_websocket. Privatises ChannelsToSubscribe, Connecting properties and removeChannelToSubscribe func to prevent unnecessary tampering.

* Removes unused mutex. FMTS and IMPORTS

* Fixes request lock time change

* More specific logs

* Renames ws mutex. Fixes bitmex subscriptions. Increased gateio ratelimiter to 120ms. Removes ratelimiter from bitfinex, bitmex, bitstamp, btcc, btse, coibasepro, hitbtc, huobi, hadax, poloniex and zb

* changes recieved typo due to not being well received

* Fixes parsing issue with Huobi and hadax

* Fixes data race with more locks

* removes defer locks. fixes huobi/hadax verbose output

* Fixes double JSONEncode for coinut. Fixes verbose output for coinut

* gofmt,goimport for coinut

* Fixes issue where multiple connection monitors can spawn

* Removes defer exchange.WebsocketConn.Close() in defer handledata exit as connectionmonitor handles connections instead

* gofmt and go import

* More fmts

exchanges/bitmex/bitmex.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/gorilla/websocket"
@@ -23,6 +24,7 @@ import (
 type Bitmex struct {
 	exchange.Base
 	WebsocketConn *websocket.Conn
+	wsRequestMtx  sync.Mutex
 }
 
 const (
@@ -135,7 +137,9 @@ func (b *Bitmex) SetDefaults() {
 	b.SupportsAutoPairUpdating = true
 	b.WebsocketInit()
 	b.Websocket.Functionality = exchange.WebsocketTradeDataSupported |
-		exchange.WebsocketOrderbookSupported
+		exchange.WebsocketOrderbookSupported |
+		exchange.WebsocketSubscribeSupported |
+		exchange.WebsocketUnsubscribeSupported
 }
 
 // Setup takes in the supplied exchange configuration details and sets params
@@ -174,8 +178,11 @@ func (b *Bitmex) Setup(exch *config.ExchangeConfig) {
 			log.Fatal(err)
 		}
 		err = b.WebsocketSetup(b.WsConnector,
+			b.Subscribe,
+			b.Unsubscribe,
 			exch.Name,
 			exch.Websocket,
+			exch.Verbose,
 			bitmexWSURL,
 			exch.WebsocketURL)
 		if err != nil {

exchanges/bitmex/bitmex_websocket.go

@@ -105,16 +105,7 @@ func (b *Bitmex) WsConnector() error {
 	}
 
 	go b.wsHandleIncomingData()
-
-	err = b.websocketSubscribe()
-	if err != nil {
-		closeError := b.WebsocketConn.Close()
-		if closeError != nil {
-			return fmt.Errorf("bitmex_websocket.go error - Websocket connection could not close %s",
-				closeError)
-		}
-		return err
-	}
+	b.GenerateDefaultSubscriptions()
 
 	if b.AuthenticatedAPISupport {
 		err := b.websocketSendAuth()
@@ -143,11 +134,6 @@ func (b *Bitmex) wsHandleIncomingData() {
 	b.Websocket.Wg.Add(1)
 
 	defer func() {
-		err := b.WebsocketConn.Close()
-		if err != nil {
-			b.Websocket.DataHandler <- fmt.Errorf("bitmex_websocket.go - Unable to close Websocket connection. Error: %s",
-				err)
-		}
 		b.Websocket.Wg.Done()
 	}()
 
@@ -170,7 +156,7 @@ func (b *Bitmex) wsHandleIncomingData() {
 			}
 
 			if common.StringContains(message, "ping") {
-				err = b.WebsocketConn.WriteJSON("pong")
+				err = b.wsSend("pong")
 				if err != nil {
 					b.Websocket.DataHandler <- err
 					continue
@@ -398,26 +384,42 @@ func (b *Bitmex) processOrderbook(data []OrderBookL2, action string, currencyPai
 	return nil
 }
 
-// WebsocketSubscribe subscribes to a websocket channel
-func (b *Bitmex) websocketSubscribe() error {
+// GenerateDefaultSubscriptions Adds default subscriptions to websocket to be handled by ManageSubscriptions()
+func (b *Bitmex) GenerateDefaultSubscriptions() {
 	contracts := b.GetEnabledCurrencies()
+	channels := []string{bitmexWSOrderbookL2, bitmexWSTrade}
+	subscriptions := []exchange.WebsocketChannelSubscription{
+		{
+			Channel: bitmexWSAnnouncement,
+		},
+	}
+	for i := range channels {
+		for j := range contracts {
+			subscriptions = append(subscriptions, exchange.WebsocketChannelSubscription{
+				Channel:  fmt.Sprintf("%v:%v", channels[i], contracts[j].String()),
+				Currency: contracts[j],
+			})
+		}
+	}
+	b.Websocket.SubscribeToChannels(subscriptions)
+}
 
-	// Subscriber
+// Subscribe subscribes to a websocket channel
+func (b *Bitmex) Subscribe(channelToSubscribe exchange.WebsocketChannelSubscription) error {
 	var subscriber WebsocketRequest
 	subscriber.Command = "subscribe"
+	subscriber.Arguments = append(subscriber.Arguments, channelToSubscribe.Channel)
+	return b.wsSend(subscriber)
+}
 
-	// Announcement subscribe
-	subscriber.Arguments = append(subscriber.Arguments, bitmexWSAnnouncement)
-
-	for _, contract := range contracts {
-		// Orderbook and Trade subscribe
-		// NOTE more added here in future
-		subscriber.Arguments = append(subscriber.Arguments,
-			bitmexWSOrderbookL2+":"+contract.String(),
-			bitmexWSTrade+":"+contract.String())
-	}
-
-	return b.WebsocketConn.WriteJSON(subscriber)
+// Unsubscribe sends a websocket message to stop receiving data from the channel
+func (b *Bitmex) Unsubscribe(channelToSubscribe exchange.WebsocketChannelSubscription) error {
+	var subscriber WebsocketRequest
+	subscriber.Command = "unsubscribe"
+	subscriber.Arguments = append(subscriber.Arguments,
+		channelToSubscribe.Params["args"],
+		channelToSubscribe.Channel+":"+channelToSubscribe.Currency.String())
+	return b.wsSend(subscriber)
 }
 
 // WebsocketSendAuth sends an authenticated subscription
@@ -433,5 +435,15 @@ func (b *Bitmex) websocketSendAuth() error {
 	sendAuth.Command = "authKeyExpires"
 	sendAuth.Arguments = append(sendAuth.Arguments, b.APIKey, timestamp,
 		signature)
-	return b.WebsocketConn.WriteJSON(sendAuth)
+	return b.wsSend(sendAuth)
+}
+
+// WsSend sends data to the websocket server
+func (b *Bitmex) wsSend(data interface{}) error {
+	b.wsRequestMtx.Lock()
+	defer b.wsRequestMtx.Unlock()
+	if b.Verbose {
+		log.Debugf("%v sending message to websocket %v", b.Name, data)
+	}
+	return b.WebsocketConn.WriteJSON(data)
 }

exchanges/bitmex/bitmex_wrapper.go

@@ -396,3 +396,17 @@ func (b *Bitmex) GetOrderHistory(getOrdersRequest *exchange.GetOrdersRequest) ([
 
 	return orders, nil
 }
+
+// SubscribeToWebsocketChannels appends to ChannelsToSubscribe
+// which lets websocket.manageSubscriptions handle subscribing
+func (b *Bitmex) SubscribeToWebsocketChannels(channels []exchange.WebsocketChannelSubscription) error {
+	b.Websocket.SubscribeToChannels(channels)
+	return nil
+}
+
+// UnsubscribeToWebsocketChannels removes from ChannelsToSubscribe
+// which lets websocket.manageSubscriptions handle unsubscribing
+func (b *Bitmex) UnsubscribeToWebsocketChannels(channels []exchange.WebsocketChannelSubscription) error {
+	b.Websocket.UnsubscribeToChannels(channels)
+	return nil
+}