fixes in websockets

0.11.2

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## 0.11.3
+
+* `Ktor`:
+    * Support of `WebSockets` has been improved
+      * `Client`:
+        * New extensions: `HttpClient#openBaseWebSocketFlow`, `HttpClient#openWebSocketFlow`, `HttpClient#openSecureWebSocketFlow`
+
+## 0.11.2
+
+* `Ktor`:
+  * Support of `WebSockets` has been improved and added fixes inside of clients
+
+## 0.11.1
+
+* `Repos`
+  * `Ktor`
+    * In `configureReadKeyValueRepoRoutes` and `configureReadKeyValuesRepoRoutes` configurators fixed requiring of `reversed` property
+
 ## 0.11.0
 
 * `Versions`

gradle.properties

@@ -14,5 +14,5 @@ crypto_js_version=4.1.1
 # Project data
 
 group=dev.inmo
-version=0.11.0
-android_code_version=124
+version=0.11.3
+android_code_version=127

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

@@ -1,5 +1,6 @@
 package dev.inmo.micro_utils.ktor.client
 
+import dev.inmo.micro_utils.common.Warning
 import dev.inmo.micro_utils.coroutines.runCatchingSafely
 import dev.inmo.micro_utils.coroutines.safely
 import dev.inmo.micro_utils.ktor.common.*
@@ -7,35 +8,25 @@ import io.ktor.client.HttpClient
 import io.ktor.client.plugins.pluginOrNull
 import io.ktor.client.plugins.websocket.*
 import io.ktor.client.request.HttpRequestBuilder
-import io.ktor.websocket.Frame
-import io.ktor.websocket.readBytes
-import io.ktor.websocket.serialization.sendSerializedBase
+import io.ktor.http.URLProtocol
+import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.isActive
-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(
-    url: String,
+@Warning("This feature is internal and should not be used directly. It is can be changed without any notification and warranty on compile-time or other guaranties")
+inline fun <reified T : Any> openBaseWebSocketFlow(
     noinline checkReconnection: suspend (Throwable?) -> Boolean = { true },
-    noinline requestBuilder: HttpRequestBuilder.() -> Unit = {}
+    noinline webSocketSessionRequest: suspend SendChannel<T>.() -> Unit
 ): Flow<T> {
-    pluginOrNull(WebSockets) ?: error("Plugin $WebSockets must be installed for using createStandardWebsocketFlow")
-
-    val correctedUrl = url.asCorrectWebSocketUrl
-
     return channelFlow {
         do {
             val reconnect = runCatchingSafely {
-                ws(correctedUrl, requestBuilder) {
-                    for (received in incoming) {
-                        sendSerialized(received.data)
-                    }
-                }
+                webSocketSessionRequest()
                 checkReconnection(null)
             }.getOrElse { e ->
                 checkReconnection(e).also {
@@ -53,3 +44,60 @@ inline fun <reified T> HttpClient.createStandardWebsocketFlow(
         }
     }
 }
+
+/**
+ * @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 : Any> HttpClient.openWebSocketFlow(
+    url: String,
+    useSecureConnection: Boolean,
+    noinline checkReconnection: suspend (Throwable?) -> Boolean = { true },
+    noinline requestBuilder: HttpRequestBuilder.() -> Unit = {}
+): Flow<T> {
+    pluginOrNull(WebSockets) ?: error("Plugin $WebSockets must be installed for using createStandardWebsocketFlow")
+
+    return openBaseWebSocketFlow<T>(checkReconnection) {
+        val block: suspend DefaultClientWebSocketSession.() -> Unit = {
+            while (isActive) {
+                send(receiveDeserialized<T>())
+            }
+        }
+
+        if (useSecureConnection) {
+            wss(url, requestBuilder, block)
+        } else {
+            ws(url, requestBuilder, block)
+        }
+    }
+}
+
+/**
+ * @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 : Any> HttpClient.openWebSocketFlow(
+    url: String,
+    noinline checkReconnection: suspend (Throwable?) -> Boolean = { true },
+    noinline requestBuilder: HttpRequestBuilder.() -> Unit = {}
+): Flow<T> = openWebSocketFlow(url, false, checkReconnection, requestBuilder)
+
+/**
+ * @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 : Any> HttpClient.openSecureWebSocketFlow(
+    url: String,
+    noinline checkReconnection: suspend (Throwable?) -> Boolean = { true },
+    noinline requestBuilder: HttpRequestBuilder.() -> Unit = {}
+): Flow<T> = openWebSocketFlow(url, true, checkReconnection, requestBuilder)
+
+/**
+ * @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 : Any> HttpClient.createStandardWebsocketFlow(
+    url: String,
+    noinline checkReconnection: suspend (Throwable?) -> Boolean = { true },
+    noinline requestBuilder: HttpRequestBuilder.() -> Unit = {}
+): Flow<T> = openWebSocketFlow(url, checkReconnection, requestBuilder)

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)
             }
         }
     }

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

@@ -13,6 +13,7 @@ import io.ktor.http.content.streamProvider
 import io.ktor.server.application.call
 import io.ktor.server.request.receiveMultipart
 import io.ktor.server.response.respond
+import io.ktor.server.response.respondText
 import io.ktor.server.routing.Route
 import io.ktor.server.routing.post
 import kotlinx.coroutines.*
@@ -111,7 +112,7 @@ class TemporalFilesRoutingConfigurator(
                     temporalFilesMutex.withLock {
                         temporalFilesMap[fileId] = file
                     }
-                    call.respond(fileId.string)
+                    call.respondText(fileId.string)
                     launchSafelyWithoutExceptions { filesFlow.emit(fileId) }
                 } ?: call.respond(HttpStatusCode.BadRequest)
             }

repos/ktor/client/src/commonMain/kotlin/dev/inmo/micro_utils/repos/ktor/client/crud/KtorCRUDRepoClient.kt

@@ -1,6 +1,7 @@
 package dev.inmo.micro_utils.repos.ktor.client.crud
 
 import dev.inmo.micro_utils.ktor.common.*
+import dev.inmo.micro_utils.pagination.PaginationResult
 import dev.inmo.micro_utils.repos.*
 import io.ktor.client.HttpClient
 import io.ktor.http.ContentType
@@ -19,14 +20,14 @@ class KtorCRUDRepoClient<ObjectType, IdType, InputValue> (
         inline operator fun <reified ObjectType, reified IdType, reified InputValue> invoke(
             baseUrl: String,
             httpClient: HttpClient,
-            objectTypeInfo: TypeInfo,
             contentType: ContentType,
             noinline idSerializer: suspend (IdType) -> String
         ) = KtorCRUDRepoClient(
             KtorReadCRUDRepoClient(
                 baseUrl,
                 httpClient,
-                objectTypeInfo,
+                typeInfo<ObjectType>(),
+                typeInfo<PaginationResult<ObjectType>>(),
                 contentType,
                 idSerializer
             ),
@@ -41,33 +42,17 @@ class KtorCRUDRepoClient<ObjectType, IdType, InputValue> (
             baseUrl: String,
             subpart: String,
             httpClient: HttpClient,
-            objectTypeInfo: TypeInfo,
             contentType: ContentType,
             noinline idSerializer: suspend (IdType) -> String
         ) = KtorCRUDRepoClient<ObjectType, IdType, InputValue>(
             buildStandardUrl(baseUrl, subpart),
             httpClient,
-            objectTypeInfo,
             contentType,
             idSerializer
         )
     }
 }
 
-
-inline fun <reified ObjectType, reified IdType, reified InputValue> KtorCRUDRepoClient(
-    baseUrl: String,
-    httpClient: HttpClient,
-    contentType: ContentType,
-    noinline idSerializer: suspend (IdType) -> String
-) = KtorCRUDRepoClient<ObjectType, IdType, InputValue>(
-    baseUrl,
-    httpClient,
-    typeInfo<ObjectType>(),
-    contentType,
-    idSerializer
-)
-
 inline fun <reified ObjectType, reified IdType, reified InputValue> KtorCRUDRepoClient(
     baseUrl: String,
     httpClient: HttpClient,

repos/ktor/client/src/commonMain/kotlin/dev/inmo/micro_utils/repos/ktor/client/crud/KtorReadCRUDRepoClient.kt

@@ -18,6 +18,7 @@ class KtorReadCRUDRepoClient<ObjectType, IdType> (
     private val baseUrl: String,
     private val httpClient: HttpClient,
     private val objectType: TypeInfo,
+    private val paginationObjectType: TypeInfo,
     private val contentType: ContentType,
     private val idSerializer: suspend (IdType) -> String
 ) : ReadCRUDRepo<ObjectType, IdType> {
@@ -25,7 +26,7 @@ class KtorReadCRUDRepoClient<ObjectType, IdType> (
         buildStandardUrl(baseUrl, getByPaginationRouting, pagination.asUrlQueryParts)
     ) {
         contentType(contentType)
-    }.body()
+    }.body(paginationObjectType)
 
     override suspend fun getById(id: IdType): ObjectType? = httpClient.get(
         buildStandardUrl(
@@ -70,6 +71,7 @@ inline fun <reified ObjectType, IdType> KtorReadCRUDRepoClient(
     baseUrl,
     httpClient,
     typeInfo<ObjectType>(),
+    typeInfo<PaginationResult<ObjectType>>(),
     contentType,
     idSerializer
 )

repos/ktor/kv.yml

@@ -1,7 +1,7 @@
-swagger: "2.0"
+openapi: "3.0.0"
 info:
   description: "This is a template for the KeyValue repositories from [microutils](https://github.com/InsanusMokrassar/MicroUtils/tree/master/repos/ktor/server/src/jvmMain/kotlin/dev/inmo/micro_utils/repos/ktor/server/key/value)"
-  version: "0.11.0"
+  version: "0.11.1"
   title: "KeyValue Repo"
   contact:
     email: "ovsyannikov.alexey95@gmail.com"
@@ -11,64 +11,46 @@ tags:
   - name: "Write"
     description: "Operations with `post` request in most cases"
 
-parameters:
+components:
+  parameters:
     KeyInQuery:
       in: "query"
       name: "key"
-    allOf:
-      - $ref: "#/definitions/Key"
-  KeysInBody:
-    in: "body"
-    name: "body"
-    type: array
-    items:
-      $ref: "#/definitions/Key"
-  ValuesInBody:
-    in: "body"
-    name: "body"
-    type: array
-    items:
-      $ref: "#/definitions/Value"
+      schema:
+        $ref: "#/components/schemas/Key"
     PaginationInQueryPage:
       in: "query"
-    type: integer
       name: "ppage"
       description: "Page of pagination"
+      schema:
+        type: integer
       required: false
     PaginationInQuerySize:
       in: "query"
-    type: integer
       name: "psize"
       description: "Size of each page in pagination"
+      schema:
+        type: integer
       required: false
     ReversedInQuery:
       in: "query"
-    type: boolean
       name: "reversed"
       description: "If passed, will tell to reverse the result pages"
+      schema:
+        type: boolean
       required: false
     ValueInQuery:
       in: "query"
       name: "value"
-    allOf:
-      - $ref: "#/definitions/Value"
-  MapInBody:
-    in: "body"
-    name: "body"
-    allOf:
-      - $ref: "#/definitions/Map"
-
-
-definitions:
+      schema:
+        $ref: "#/components/schemas/Value"
+  schemas:
     Key:
       type: integer
       description: "REWRITE THIS TYPE AS KEY IN SWAGGER FILE"
     Value:
       type: integer
       description: "REWRITE THIS TYPE AS VALUE IN SWAGGER FILE"
-  Map:
-    type: object
-    description: "Map of objects"
     PaginationResult:
       type: object
       properties:
@@ -93,13 +75,16 @@ paths:
       tags:
         - "Read"
       parameters:
-        - $ref: "#/parameters/KeyInQuery"
-          required: true
+        - allOf:
+            - $ref: "#/components/parameters/KeyInQuery"
+            - required: true
       responses:
         "200":
           description: "Element by key"
+          content:
+            "*/*":
               schema:
-            $ref: "#/definitions/Value"
+                $ref: "#/components/schemas/Value"
         "204":
           description: "No value by id"
   /values:
@@ -107,49 +92,55 @@ paths:
       tags:
         - "Read"
       parameters:
-        - $ref: "#/parameters/PaginationInQueryPage"
-        - $ref: "#/parameters/PaginationInQuerySize"
-        - $ref: "#/parameters/ReversedInQuery"
+        - $ref: "#/components/parameters/PaginationInQueryPage"
+        - $ref: "#/components/parameters/PaginationInQuerySize"
+        - $ref: "#/components/parameters/ReversedInQuery"
       responses:
         "200":
           description: "Pagination with elements"
+          content:
+            "*/*":
               schema:
                 allOf:
-              - $ref: "#/definitions/PaginationResult"
+                  - $ref: "#/components/schemas/PaginationResult"
                   - properties:
                       results:
                         items:
-                      $ref: "#/definitions/Value"
+                          $ref: "#/components/schemas/Value"
   /keys:
     get:
       tags:
         - "Read"
       parameters:
-        - $ref: "#/parameters/PaginationInQueryPage"
-        - $ref: "#/parameters/PaginationInQuerySize"
-        - $ref: "#/parameters/ReversedInQuery"
-        - $ref: "#/parameters/ValueInQuery"
+        - $ref: "#/components/parameters/PaginationInQueryPage"
+        - $ref: "#/components/parameters/PaginationInQuerySize"
+        - $ref: "#/components/parameters/ReversedInQuery"
+        - $ref: "#/components/parameters/ValueInQuery"
           required: false
       responses:
         "200":
           description: "Pagination with elements"
+          content:
+            "*/*":
               schema:
                 allOf:
-              - $ref: "#/definitions/PaginationResult"
+                  - $ref: "#/components/schemas/PaginationResult"
                   - properties:
                       results:
                         items:
-                      $ref: "#/definitions/Key"
+                          $ref: "#/components/schemas/Key"
   /contains:
     get:
       tags:
         - "Read"
       parameters:
-        - $ref: "#/parameters/KeyInQuery"
+        - $ref: "#/components/parameters/KeyInQuery"
           required: true
       responses:
         "200":
           description: "Object with id availability in repo"
+          content:
+            "*/*":
               schema:
                 type: boolean
   /count:
@@ -159,6 +150,8 @@ paths:
       responses:
         "200":
           description: "Amount of objects in repo"
+          content:
+            "*/*":
               schema:
                 type: integer
 
@@ -167,11 +160,13 @@ paths:
     post:
       tags:
         - "Write"
-      parameters:
-        - allOf:
-            - $ref: "#/parameters/MapInBody"
-            - additionalProperties:
-                $ref: "#/definitions/Value"
+      requestBody:
+        content:
+          "*/*":
+            schema:
+              type: object
+              additionalProperties:
+                $ref: "#/components/schemas/Value"
               description: "Map with new elements to set. Use keys as a keys of this map"
       responses:
         "200":
@@ -180,8 +175,13 @@ paths:
     post:
       tags:
         - "Write"
-      parameters:
-        - $ref: "#/parameters/KeysInBody"
+      requestBody:
+        content:
+          "*/*":
+            schema:
+              type: array
+              items:
+                $ref: "#/components/schemas/Key"
       responses:
         "200":
           description: "Objects with keys from body has been unset"
@@ -189,8 +189,13 @@ paths:
     post:
       tags:
         - "Write"
-      parameters:
-        - $ref: "#/parameters/ValuesInBody"
+      requestBody:
+        content:
+          "*/*":
+            schema:
+              type: array
+              items:
+                $ref: "#/components/schemas/Value"
       responses:
         "200":
           description: "Objects with values from body has been unset"

repos/ktor/server/src/jvmMain/kotlin/dev/inmo/micro_utils/repos/ktor/server/key/value/KtorReadKeyValueRepoRoutes.kt

@@ -51,7 +51,7 @@ inline fun <reified Key, reified Value> Route.configureReadKeyValueRepoRoutes (
 
     get(keysRoute) {
         val pagination = call.request.queryParameters.extractPagination
-        val reversed = call.getQueryParameterOrSendError(reversedParameterName) ?.toBoolean() ?: false
+        val reversed = call.getQueryParameter(reversedParameterName) ?.toBoolean() ?: false
         val value = call.getQueryParameter(valueParameterName) ?.let {
             valueDeserializer(it)
         }

repos/ktor/server/src/jvmMain/kotlin/dev/inmo/micro_utils/repos/ktor/server/key/values/KtorReadKeyValuesRepoRoutes.kt

@@ -42,7 +42,7 @@ inline fun <reified Key, reified Value> Route.configureReadKeyValuesRepoRoutes (
 
     get(keysRoute) {
         val pagination = call.request.queryParameters.extractPagination
-        val reversed = call.getQueryParameterOrSendError(reversedParameterName) ?.toBoolean() ?: false
+        val reversed = call.getQueryParameter(reversedParameterName) ?.toBoolean() ?: false
         val value = call.getQueryParameter(valueParameterName) ?.let {
             valueDeserializer(it)
         }