fix keepalive/read race

recws.go

@@ -62,6 +62,7 @@ type RecConn struct {
 	dialErr           error
 	dialer            *websocket.Dialer
 	keepAliveResponse *keepAliveResponse
+	manualReconnect   bool
 
 	*websocket.Conn
 }
@@ -87,6 +88,16 @@ func (rc *RecConn) handleReconnect() {
 
 // CloseAndReconnect will try to reconnect.
 func (rc *RecConn) CloseAndReconnect() {
+	if !rc.getManualReconnect() {
+		rc.Close()
+		rc.handleReconnect()
+		go rc.connect()
+	}
+}
+
+// ManualCloseAndReconnect will try to reconnect.
+func (rc *RecConn) ManualCloseAndReconnect() {
+	rc.setManualReconnect(true)
 	rc.Close()
 	rc.handleReconnect()
 	go rc.connect()
@@ -130,6 +141,19 @@ func (rc *RecConn) Shutdown(writeWait time.Duration) {
 	}
 }
 
+func (rc *RecConn) getManualReconnect() bool {
+	rc.mu.RLock()
+	defer rc.mu.RUnlock()
+
+	return rc.manualReconnect
+}
+
+func (rc *RecConn) setManualReconnect(v bool) {
+	rc.mu.Lock()
+	rc.manualReconnect = v
+	defer rc.mu.Unlock()
+}
+
 // ReadMessage is a helper method for getting a reader
 // using NextReader and reading from that reader to a buffer.
 //
@@ -462,7 +486,7 @@ func (rc *RecConn) keepAlive() {
 			timeoutOffset := time.Millisecond * 500
 			if time.Since(rc.getKeepAliveResponse().getLastResponse()) > rc.getKeepAliveTimeout()+timeoutOffset {
 				rc.log(LogValues{Err: errors.New("keepalive timeout"), Msg: "Reconnect", Url: rc.url})
-				rc.CloseAndReconnect()
+				rc.ManualCloseAndReconnect()
 				return
 			}
 		}
@@ -480,6 +504,8 @@ func (rc *RecConn) connect() {
 		}
 		wsConn, httpResp, err := rc.dialer.Dial(rc.url, rc.reqHeader)
 
+		rc.setManualReconnect(false)
+
 		rc.mu.Lock()
 		rc.Conn = wsConn
 		rc.dialErr = err