Abandon resuming after failed reconnection attempt

Author: freyacodes

src/main/kotlin/dev/arbjerg/lavalink/client/LavalinkClient.kt

@@ -1,9 +1,9 @@
 package dev.arbjerg.lavalink.client
 
+import dev.arbjerg.lavalink.client.event.ClientEvent
 import dev.arbjerg.lavalink.client.loadbalancing.ILoadBalancer
 import dev.arbjerg.lavalink.client.loadbalancing.VoiceRegion
 import dev.arbjerg.lavalink.client.loadbalancing.builtin.DefaultLoadBalancer
-import dev.arbjerg.lavalink.client.event.ClientEvent
 import dev.arbjerg.lavalink.client.player.LavalinkPlayer
 import dev.arbjerg.lavalink.internal.ReconnectTask
 import dev.arbjerg.lavalink.protocol.v4.VoiceState
@@ -17,7 +17,6 @@ import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.Executors
 import java.util.concurrent.TimeUnit
-import kotlin.time.toJavaDuration
 
 /**
  * @param userId ID of the bot for authenticating with Discord
@@ -173,16 +172,14 @@ class LavalinkClient(val userId: Long) : Closeable, Disposable {
 
         val session = node.cachedSession
         val canResume = session != null && session.resuming && session.timeoutSeconds > 0
-        if (canResume) {
-            node.resumeTimer = Mono.delay(Duration.ofSeconds(session!!.timeoutSeconds))
-                .subscribe() { transferNodes(node) }
-        } else {
+        if (!canResume) {
+            // If canResume is true, onNodeFirstReconnectFailed(node) may do the transfer
             transferNodes(node)
         }
     }
 
-    internal fun onNodeConnected(node: LavalinkNode) {
-        node.resumeTimer?.dispose()
+    internal fun onNodeFirstReconnectFailed(node: LavalinkNode) {
+        transferNodes(node)
     }
 
     private fun transferNodes(node: LavalinkNode) {

src/main/kotlin/dev/arbjerg/lavalink/client/LavalinkNode.kt

@@ -59,7 +59,6 @@ class LavalinkNode(
     internal val sink: Many<ClientEvent> = Sinks.many().multicast().onBackpressureBuffer()
     val flux: Flux<ClientEvent> = sink.asFlux()
     private val reference: Disposable = flux.subscribe()
-    internal var resumeTimer: Disposable? = null
 
     internal val rest = LavalinkRestClient(this)
     val ws = LavalinkSocket(this)

src/main/kotlin/dev/arbjerg/lavalink/internal/LavalinkSocket.kt

@@ -26,9 +26,10 @@ class LavalinkSocket(private val node: LavalinkNode) : WebSocketListener(), Clos
 
     var mayReconnect = true
     var lastReconnectAttempt = 0L
-    private var reconnectsAttempted = 0
+    @Volatile
+    private var connectionsAttempted = 0
     val reconnectInterval: Int
-        get() = reconnectsAttempted * 2000 - 2000
+        get() = connectionsAttempted * 2000 - 2000
     var open: Boolean = false
         private set
 
@@ -39,8 +40,7 @@ class LavalinkSocket(private val node: LavalinkNode) : WebSocketListener(), Clos
     override fun onOpen(webSocket: WebSocket, response: Response) {
         logger.info("${node.name} has been connected!")
         open = true
-        reconnectsAttempted = 0
-        node.lavalink.onNodeConnected(node)
+        connectionsAttempted = 0
     }
 
     override fun onMessage(webSocket: WebSocket, text: String) {
@@ -80,10 +80,12 @@ class LavalinkSocket(private val node: LavalinkNode) : WebSocketListener(), Clos
                     node.cachedSession = null
                 }
                 if (node.cachedSession == null) {
-                    node.rest.getSession().subscribe { node.cachedSession = null }
+                    node.rest.getSession().subscribe { node.cachedSession = it }
                 }
 
-                node.synchronizeAfterResume()
+                if (resumed) {
+                    node.synchronizeAfterResume()
+                }
 
                 // Move players from older, unavailable nodes to ourselves.
                 node.transferOrphansToSelf()
@@ -154,15 +156,15 @@ class LavalinkSocket(private val node: LavalinkNode) : WebSocketListener(), Clos
     }
 
     override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
-        handleFailureTrhowable(t)
+        handleFailureThrowable(t)
 
         node.available = false
         open = false
 
         node.lavalink.onNodeDisconnected(node)
     }
 
-    private fun handleFailureTrhowable(t: Throwable) {
+    private fun handleFailureThrowable(t: Throwable) {
         when(t) {
             is EOFException -> {
                 logger.debug("Got disconnected from ${node.name}, trying to reconnect", t)
@@ -190,6 +192,10 @@ class LavalinkSocket(private val node: LavalinkNode) : WebSocketListener(), Clos
                 logger.error("Unknown error on ${node.name}", t)
             }
         }
+
+        if (connectionsAttempted == 1 && lastReconnectAttempt > 0) {
+            node.lavalink.onNodeFirstReconnectFailed(node)
+        }
     }
 
     override fun onClosing(webSocket: WebSocket, code: Int, reason: String) {
@@ -213,12 +219,10 @@ class LavalinkSocket(private val node: LavalinkNode) : WebSocketListener(), Clos
                 reason
             )
         }
-
     }
 
     fun attemptReconnect() {
         lastReconnectAttempt = System.currentTimeMillis()
-        reconnectsAttempted++
         connect()
     }
 
@@ -234,13 +238,14 @@ class LavalinkSocket(private val node: LavalinkNode) : WebSocketListener(), Clos
             .addHeader("Client-Name", "Lavalink-Client/${CLIENT_VERSION}")
             .addHeader("User-Id", node.lavalink.userId.toString())
             .apply {
-                if (node.sessionId != null) {
+                if (node.sessionId != null && connectionsAttempted == 0) {
                     addHeader("Session-Id", node.sessionId!!)
                 }
             }
             .build()
 
         mayReconnect = true
+        connectionsAttempted++
         socket = node.httpClient.newWebSocket(request, this)
     }