fixes

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 0.11.2
 
+* `Ktor`:
+  * Support of `WebSockets` has been improved and added fixes inside of clients
+
 ## 0.11.1
 
 * `Repos`

ktor/client/src/commonMain/kotlin/dev/inmo/micro_utils/ktor/client/NewFlowsWebsockets.kt

@@ -19,7 +19,7 @@ import kotlinx.serialization.DeserializationStrategy
  * @param checkReconnection This lambda will be called when it is required to reconnect to websocket to establish
  * connection. Must return true in case if must be reconnected. By default always reconnecting
  */
-inline fun <reified T> HttpClient.createStandardWebsocketFlow(
+inline fun <reified T : Any> HttpClient.createStandardWebsocketFlow(
     url: String,
     noinline checkReconnection: suspend (Throwable?) -> Boolean = { true },
     noinline requestBuilder: HttpRequestBuilder.() -> Unit = {}
@@ -32,8 +32,8 @@ inline fun <reified T> HttpClient.createStandardWebsocketFlow(
         do {
             val reconnect = runCatchingSafely {
                 ws(correctedUrl, requestBuilder) {
-                    for (received in incoming) {
-                        sendSerialized(received.data)
+                    while (isActive) {
+                        send(receiveDeserialized<T>())
                     }
                 }
                 checkReconnection(null)

ktor/server/src/jvmMain/kotlin/dev/inmo/micro_utils/ktor/server/FlowsWebsocket.kt

@@ -12,6 +12,7 @@ import io.ktor.websocket.send
 import kotlinx.coroutines.flow.Flow
 import kotlinx.serialization.SerializationStrategy
 
+@Deprecated("This method will be removed soon")
 fun <T> Route.includeWebsocketHandling(
     suburl: String,
     flow: Flow<T>,

ktor/server/src/jvmMain/kotlin/dev/inmo/micro_utils/ktor/server/NewFlowsWebsocket.kt

@@ -15,7 +15,8 @@ import kotlinx.serialization.SerializationStrategy
 inline fun <reified T : Any> Route.includeWebsocketHandling(
     suburl: String,
     flow: Flow<T>,
-    protocol: URLProtocol? = null
+    protocol: URLProtocol? = null,
+    noinline dataMapper: suspend WebSocketServerSession.(T) -> T? = { it }
 ) {
     application.apply {
         pluginOrNull(WebSockets) ?: install(WebSockets)
@@ -23,7 +24,7 @@ inline fun <reified T : Any> Route.includeWebsocketHandling(
     webSocket(suburl, protocol ?.name) {
         safely {
             flow.collect {
-                sendSerialized(it)
+                sendSerialized(dataMapper(it) ?: return@collect)
             }
         }
     }