Compare commits

..

No commits in common. "1c52e04cdbbdb0ba6a4738a71d511087ec3b84b9" and "c9c6d4c0c1f4d51f93d66a7d463191d6bd6263d5" have entirely different histories.

3 changed files with 18 additions and 73 deletions

View File

@ -1,12 +1,5 @@
# Changelog # Changelog
## 0.11.3
* `Ktor`:
* Support of `WebSockets` has been improved
* `Client`:
* New extensions: `HttpClient#openBaseWebSocketFlow`, `HttpClient#openWebSocketFlow`, `HttpClient#openSecureWebSocketFlow`
## 0.11.2 ## 0.11.2
* `Ktor`: * `Ktor`:

View File

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

View File

@ -1,6 +1,5 @@
package dev.inmo.micro_utils.ktor.client 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.runCatchingSafely
import dev.inmo.micro_utils.coroutines.safely import dev.inmo.micro_utils.coroutines.safely
import dev.inmo.micro_utils.ktor.common.* import dev.inmo.micro_utils.ktor.common.*
@ -8,25 +7,35 @@ import io.ktor.client.HttpClient
import io.ktor.client.plugins.pluginOrNull import io.ktor.client.plugins.pluginOrNull
import io.ktor.client.plugins.websocket.* import io.ktor.client.plugins.websocket.*
import io.ktor.client.request.HttpRequestBuilder import io.ktor.client.request.HttpRequestBuilder
import io.ktor.http.URLProtocol import io.ktor.websocket.Frame
import kotlinx.coroutines.channels.SendChannel import io.ktor.websocket.readBytes
import io.ktor.websocket.serialization.sendSerializedBase
import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.isActive 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 * @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 * connection. Must return true in case if must be reconnected. By default always reconnecting
*/ */
@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> HttpClient.createStandardWebsocketFlow(
inline fun <reified T : Any> openBaseWebSocketFlow( url: String,
noinline checkReconnection: suspend (Throwable?) -> Boolean = { true }, noinline checkReconnection: suspend (Throwable?) -> Boolean = { true },
noinline webSocketSessionRequest: suspend SendChannel<T>.() -> Unit noinline requestBuilder: HttpRequestBuilder.() -> Unit = {}
): Flow<T> { ): Flow<T> {
pluginOrNull(WebSockets) ?: error("Plugin $WebSockets must be installed for using createStandardWebsocketFlow")
val correctedUrl = url.asCorrectWebSocketUrl
return channelFlow { return channelFlow {
do { do {
val reconnect = runCatchingSafely { val reconnect = runCatchingSafely {
webSocketSessionRequest() ws(correctedUrl, requestBuilder) {
while (isActive) {
send(receiveDeserialized<T>())
}
}
checkReconnection(null) checkReconnection(null)
}.getOrElse { e -> }.getOrElse { e ->
checkReconnection(e).also { checkReconnection(e).also {
@ -44,60 +53,3 @@ inline fun <reified T : Any> openBaseWebSocketFlow(
} }
} }
} }
/**
* @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)