Document exception behaviour

Author: duncte123

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

@@ -112,10 +112,22 @@ class LavalinkClient(val userId: Long) : Closeable, Disposable {
     }
 
     // For the java people
+    /**
+     * Listen to events from all nodes. Please note that uncaught exceptions will cause the listener to stop emitting events.
+     *
+     * @param type the [ClientEvent] to listen for
+     *
+     * @return a [Flux] of [ClientEvent]s
+     */
     fun <T : ClientEvent<*>> on(type: Class<T>): Flux<T> {
         return flux.ofType(type)
     }
 
+    /**
+     * Listen to events from all nodes. Please note that uncaught exceptions will cause the listener to stop emitting events.
+     *
+     * @return a [Flux] of [ClientEvent]s
+     */
     inline fun <reified T : ClientEvent<*>> on() = on(T::class.java)
 
     /**
@@ -138,7 +150,7 @@ class LavalinkClient(val userId: Long) : Closeable, Disposable {
                 try {
                     sink.tryEmitNext(it)
                 } catch (e: Exception) {
-                    sink.tryEmitError(e)
+                    sink.emitError(e, Sinks.EmitFailureHandler.FAIL_FAST)
                 }
             }
     }

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

@@ -82,10 +82,22 @@ class LavalinkNode(
     }
 
     // For the java people
+    /**
+     * Listen to events from the node. Please note that uncaught exceptions will cause the listener to stop emitting events.
+     *
+     * @param type the [ClientEvent] to listen for
+     *
+     * @return a [Flux] of [ClientEvent]s
+     */
     fun <T : ClientEvent<*>> on(type: Class<T>): Flux<T> {
         return flux.ofType(type)
     }
 
+    /**
+     * Listen to events from the node. Please note that uncaught exceptions will cause the listener to stop emitting events.
+     *
+     * @return a [Flux] of [ClientEvent]s
+     */
     inline fun <reified T : ClientEvent<*>> on() = on(T::class.java)
 
     /**

src/main/kotlin/dev/arbjerg/lavalink/client/protocol/Track.kt

@@ -5,6 +5,7 @@ import dev.arbjerg.lavalink.internal.fromJsonElement
 import dev.arbjerg.lavalink.internal.toJackson
 import dev.arbjerg.lavalink.internal.toJsonElement
 import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.jsonObject
 import dev.arbjerg.lavalink.protocol.v4.Track as ProtocolTrack
 
 internal fun ProtocolTrack.toCustom() = Track(this)
@@ -16,9 +17,9 @@ class Track internal constructor(private var internalTrack: ProtocolTrack) {
     val info = internalTrack.info
     val pluginInfo = internalTrack.pluginInfo.toJackson()
 
-    fun setUserData(userData: Any) {
+    fun setUserData(userData: Any?) {
         internalTrack = internalTrack.copyWithUserData(
-            toJsonElement(userData) as JsonObject
+            toJsonElement(userData).jsonObject
         )
     }
 

src/main/kotlin/dev/arbjerg/lavalink/client/protocol/TrackUpdateBuilder.kt

@@ -5,6 +5,7 @@ import dev.arbjerg.lavalink.protocol.v4.Omissible
 import dev.arbjerg.lavalink.protocol.v4.PlayerUpdateTrack
 import dev.arbjerg.lavalink.protocol.v4.toOmissible
 import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.jsonObject
 
 /**
  * Allows you to update the playing track with only the fields that you wish to update.
@@ -42,9 +43,9 @@ class TrackUpdateBuilder {
         )
     }
 
-    fun setUserData(userData: Any) = apply {
+    fun setUserData(userData: Any?) = apply {
         internalUpdate = internalUpdate.copy(
-            userData = (toJsonElement(userData) as JsonObject).toOmissible()
+            userData = (toJsonElement(userData).jsonObject).toOmissible()
         )
     }
 

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

@@ -13,7 +13,11 @@ private val objectMapper = ObjectMapper()
 fun JsonElement.toJackson(): JsonNode = objectMapper.readTree(this.toString())
 fun JsonNode.toKotlin(): JsonObject = toJsonElement(this) as JsonObject
 
-fun toJsonElement(obj: Any): JsonElement {
+fun toJsonElement(obj: Any?): JsonElement {
+    if (obj == null) {
+        return JsonObject(mapOf())
+    }
+
     val jsonString = objectMapper.writeValueAsString(obj)
 
     return json.parseToJsonElement(jsonString)

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

@@ -12,6 +12,7 @@ import okhttp3.Response
 import okhttp3.WebSocket
 import okhttp3.WebSocketListener
 import org.slf4j.LoggerFactory
+import reactor.core.publisher.Sinks
 import java.io.Closeable
 import java.io.EOFException
 import java.net.ConnectException
@@ -113,7 +114,7 @@ class LavalinkSocket(private val node: LavalinkNode) : WebSocketListener(), Clos
         try {
             node.sink.tryEmitNext(event.toClientEvent(node))
         } catch (e: Exception) {
-            node.sink.tryEmitError(e)
+            node.sink.emitError(e, Sinks.EmitFailureHandler.FAIL_FAST)
         }
     }
 

src/test/kotlin/testScript.kt

@@ -28,9 +28,25 @@ fun main() {
             println("Node '${event.node.name}' is ready, session id is '${event.sessionId}'!")
         }
 
+    // Testing for exceptions thrown in the handler
+    var counter = 0
+
     client.on<StatsEvent>()
-        .subscribe { event ->
+        .subscribe({ event ->
+            counter++
             println("Node '${event.node.name}' has stats, current players: ${event.playingPlayers}/${event.players}")
+
+            // WARNING: if you do not catch your exception here, the subscription will be canceled.
+            if (counter == 2) {
+                throw RuntimeException("Counter has reached 2")
+            }
+        }) { err ->
+            err.printStackTrace()
+        }
+
+    client.on<StatsEvent>()
+        .subscribe { event ->
+            println("[event 2] Node '${event.node.name}' has stats, current players: ${event.playingPlayers}/${event.players}")
         }
 
     client.on<EmittedEvent<*>>()
@@ -84,19 +100,19 @@ fun main() {
 
 fun registerNode(client: LavalinkClient) {
     listOf(
-        client.addNode(
+        /*client.addNode(
             "Testnode",
             URI.create("ws://localhost:2333"),
             "youshallnotpass",
             RegionGroup.EUROPE
-        ),
+        ),*/
 
-        /*client.addNode(
+        client.addNode(
             "Mac-mini",
             URI.create("ws://mac-mini.local.duncte123.lgbt:2333/bepis"),
             "youshallnotpass",
             RegionGroup.US
-        )*/
+        )
     )
         .forEach { node ->
             node.on<TrackStartEvent>()