Websocket: Simplify the connectionMonitor

exchanges/stream/websocket.go

@@ -34,7 +34,6 @@ var (
 
 // Private websocket errors
 var (
-	errAlreadyRunning                       = errors.New("connection monitor is already running")
 	errExchangeConfigIsNil                  = errors.New("exchange config is nil")
 	errExchangeConfigEmpty                  = errors.New("exchange config is empty")
 	errWebsocketIsNil                       = errors.New("websocket is nil")
@@ -292,9 +291,7 @@ func (w *Websocket) Connect() error {
 	}
 	w.setState(connected)
 
-	if !w.IsConnectionMonitorRunning() {
-		go w.connectionMonitor()
-	}
+	go w.connectionMonitor()
 
 	subs, err := w.GenerateSubs() // regenerate state on new connection
 	if err != nil {
@@ -374,7 +371,6 @@ func (w *Websocket) dataMonitor() {
 
 // connectionMonitor ensures that the WS keeps connecting
 func (w *Websocket) connectionMonitor() {
-	w.setConnectionMonitorRunning(true)
 	delay := w.connectionMonitorDelay
 	timer := time.NewTimer(delay)
 	for {
@@ -394,7 +390,7 @@ func (w *Websocket) connectionMonitor() {
 				log.Debugf(log.WebsocketMgr, "%v websocket: connection monitor exiting", w.exchangeName)
 			}
 			timer.Stop()
-			w.setConnectionMonitorRunning(false)
+			w.connectionMonitorRunning.Store(false)
 			return
 		}
 		select {
@@ -628,19 +624,6 @@ func (w *Websocket) IsTrafficMonitorRunning() bool {
 	return w.trafficMonitorRunning.Load()
 }
 
-func (w *Websocket) checkAndSetMonitorRunning() (alreadyRunning bool) {
-	return !w.connectionMonitorRunning.CompareAndSwap(false, true)
-}
-
-func (w *Websocket) setConnectionMonitorRunning(b bool) {
-	w.connectionMonitorRunning.Store(b)
-}
-
-// IsConnectionMonitorRunning returns status of connection monitor
-func (w *Websocket) IsConnectionMonitorRunning() bool {
-	return w.connectionMonitorRunning.Load()
-}
-
 func (w *Websocket) setDataMonitorRunning(b bool) {
 	w.dataMonitorRunning.Store(b)
 }

exchanges/stream/websocket_test.go

@@ -571,13 +571,14 @@ func TestConnectionMonitor(t *testing.T) {
 	ws.exchangeName = "hello"
 	ws.Wg = &sync.WaitGroup{}
 	ws.setEnabled(true)
+	ws.connectionMonitorRunning.Store(true)
 	go ws.connectionMonitor()
 	require.EventuallyWithT(t, func(c *assert.CollectT) {
-		assert.True(c, ws.IsConnectionMonitorRunning(), "IsConnectionMonitorRunning should return true")
+		assert.True(c, ws.connectionMonitorRunning.Load(), "IsConnectionMonitorRunning should return true")
 	}, 5*time.Second, 10*time.Millisecond, "ConnectionMonitor must be running")
 	ws.setEnabled(false)
 	require.EventuallyWithT(t, func(c *assert.CollectT) {
-		assert.False(c, ws.IsConnectionMonitorRunning(), "IsConnectionMonitorRunning should return false")
+		assert.False(c, ws.connectionMonitorRunning.Load(), "IsConnectionMonitorRunning should return false")
 	}, 5*time.Second, 10*time.Millisecond, "ConnectionMonitor must not be running")
 }