update ktor utils

This commit is contained in:
InsanusMokrassar 2020-09-26 21:42:05 +06:00
parent fe37f6584e
commit 6f46413a90
13 changed files with 109 additions and 55 deletions

View File

@ -1,8 +1,7 @@
package dev.inmo.micro_utils.ktor.client package dev.inmo.micro_utils.ktor.client
import dev.inmo.micro_utils.coroutines.safely import dev.inmo.micro_utils.coroutines.safely
import dev.inmo.micro_utils.ktor.common.asCorrectWebSocketUrl import dev.inmo.micro_utils.ktor.common.*
import dev.inmo.micro_utils.ktor.common.standardKtorSerialFormat
import io.ktor.client.HttpClient import io.ktor.client.HttpClient
import io.ktor.client.features.websocket.ws import io.ktor.client.features.websocket.ws
import io.ktor.http.cio.websocket.* import io.ktor.http.cio.websocket.*
@ -17,7 +16,7 @@ import kotlinx.serialization.DeserializationStrategy
inline fun <T> HttpClient.createStandardWebsocketFlow( inline fun <T> HttpClient.createStandardWebsocketFlow(
url: String, url: String,
crossinline checkReconnection: (Throwable?) -> Boolean = { true }, crossinline checkReconnection: (Throwable?) -> Boolean = { true },
crossinline conversation: suspend (ByteArray) -> T crossinline conversation: suspend (StandardKtorSerialInputData) -> T
): Flow<T> { ): Flow<T> {
val correctedUrl = url.asCorrectWebSocketUrl val correctedUrl = url.asCorrectWebSocketUrl
@ -25,19 +24,11 @@ inline fun <T> HttpClient.createStandardWebsocketFlow(
val producerScope = this@channelFlow val producerScope = this@channelFlow
do { do {
val reconnect = try { val reconnect = try {
safely( safely ({ throw it }) {
{ ws(correctedUrl) {
throw it for (received in incoming) {
} when (received) {
) { is Frame.Binary -> producerScope.send(conversation(received.readBytes()))
ws(
correctedUrl
) {
while (true) {
when (val received = incoming.receive()) {
is Frame.Binary -> producerScope.send(
conversation(received.readBytes())
)
else -> { else -> {
producerScope.close() producerScope.close()
return@ws return@ws
@ -57,7 +48,7 @@ inline fun <T> HttpClient.createStandardWebsocketFlow(
} while (reconnect) } while (reconnect)
if (!producerScope.isClosedForSend) { if (!producerScope.isClosedForSend) {
safely( safely(
{ /* do nothing */ } { it.printStackTrace() }
) { ) {
producerScope.close() producerScope.close()
} }
@ -77,6 +68,5 @@ inline fun <T> HttpClient.createStandardWebsocketFlow(
url, url,
checkReconnection checkReconnection
) { ) {
standardKtorSerialFormat.decodeFromByteArray(deserializer, it) standardKtorSerialFormat.decodeDefault(deserializer, it)
} }

View File

@ -1,6 +1,6 @@
package dev.inmo.micro_utils.ktor.client package dev.inmo.micro_utils.ktor.client
import dev.inmo.micro_utils.ktor.common.standardKtorSerialFormat import dev.inmo.micro_utils.ktor.common.*
import io.ktor.client.HttpClient import io.ktor.client.HttpClient
import io.ktor.client.request.get import io.ktor.client.request.get
import io.ktor.client.request.post import io.ktor.client.request.post
@ -11,13 +11,14 @@ typealias BodyPair<T> = Pair<SerializationStrategy<T>, T>
suspend fun <ResultType> HttpClient.uniget( suspend fun <ResultType> HttpClient.uniget(
url: String, url: String,
resultDeserializer: DeserializationStrategy<ResultType> resultDeserializer: DeserializationStrategy<ResultType>
) = get<ByteArray>( ) = get<StandardKtorSerialInputData>(
url url
).let { ).let {
standardKtorSerialFormat.decodeFromByteArray(resultDeserializer, it) standardKtorSerialFormat.decodeDefault(resultDeserializer, it)
} }
fun <T> SerializationStrategy<T>.encodeUrlQueryValue(value: T) = standardKtorSerialFormat.encodeToHexString(
fun <T> SerializationStrategy<T>.encodeUrlQueryValue(value: T) = standardKtorSerialFormat.encodeHex(
this, this,
value value
) )
@ -26,8 +27,8 @@ suspend fun <BodyType, ResultType> HttpClient.unipost(
url: String, url: String,
bodyInfo: BodyPair<BodyType>, bodyInfo: BodyPair<BodyType>,
resultDeserializer: DeserializationStrategy<ResultType> resultDeserializer: DeserializationStrategy<ResultType>
) = post<ByteArray>(url) { ) = post<StandardKtorSerialInputData>(url) {
body = standardKtorSerialFormat.encodeToByteArray(bodyInfo.first, bodyInfo.second) body = standardKtorSerialFormat.encodeDefault(bodyInfo.first, bodyInfo.second)
}.let { }.let {
standardKtorSerialFormat.decodeFromByteArray(resultDeserializer, it) standardKtorSerialFormat.decodeDefault(resultDeserializer, it)
} }

View File

@ -21,3 +21,4 @@ fun buildStandardUrl(
subpart: String, subpart: String,
vararg parameters: QueryParam vararg parameters: QueryParam
) = buildStandardUrl(basePart, subpart, parameters.toList()) ) = buildStandardUrl(basePart, subpart, parameters.toList())

View File

@ -1,6 +0,0 @@
package dev.inmo.micro_utils.ktor.common
import kotlinx.serialization.builtins.SetSerializer
import kotlinx.serialization.builtins.serializer
val setIdsSerializer = SetSerializer(String.serializer())

View File

@ -1,6 +1,6 @@
package dev.inmo.micro_utils.ktor.common package dev.inmo.micro_utils.ktor.common
private val schemaRegex = Regex("[^:]*//") private val schemaRegex = Regex("^[^:]*://")
val String.asCorrectWebSocketUrl: String val String.asCorrectWebSocketUrl: String
get() = if (startsWith("ws")) { get() = if (startsWith("ws")) {

View File

@ -11,7 +11,7 @@ val List<QueryParam>.asUrlQuery: String
fun String.includeQueryParams( fun String.includeQueryParams(
queryParams: QueryParams queryParams: QueryParams
): String = "$this${if (contains("?")) "&" else "?"}${queryParams.asUrlQuery}" ): String = "$this${if(queryParams.isNotEmpty()) "${if (contains("?")) "&" else "?"}${queryParams.asUrlQuery}" else ""}"
fun String.includeQueryParams( fun String.includeQueryParams(
queryParams: List<QueryParam> queryParams: List<QueryParam>

View File

@ -1,5 +1,32 @@
package dev.inmo.micro_utils.ktor.common package dev.inmo.micro_utils.ktor.common
import kotlinx.serialization.*
import kotlinx.serialization.cbor.Cbor import kotlinx.serialization.cbor.Cbor
val standardKtorSerialFormat = Cbor typealias StandardKtorSerialFormat = BinaryFormat
typealias StandardKtorSerialInputData = ByteArray
val standardKtorSerialFormat: StandardKtorSerialFormat = Cbor { }
inline fun <T> StandardKtorSerialFormat.decodeDefault(
deserializationStrategy: DeserializationStrategy<T>,
input: StandardKtorSerialInputData
): T = decodeFromByteArray(deserializationStrategy, input)
inline fun <T> StandardKtorSerialFormat.encodeDefault(
serializationStrategy: SerializationStrategy<T>,
data: T
): StandardKtorSerialInputData = encodeToByteArray(serializationStrategy, data)
val cbor = Cbor {}
inline fun <T> StandardKtorSerialFormat.decodeHex(
deserializationStrategy: DeserializationStrategy<T>,
input: String
): T = decodeFromHexString(deserializationStrategy, input)
inline fun <T> StandardKtorSerialFormat.encodeHex(
serializationStrategy: SerializationStrategy<T>,
data: T
): String = encodeToHexString(serializationStrategy, data)

View File

@ -1,7 +0,0 @@
package dev.inmo.micro_utils.ktor.common
const val clientWebsocketHelloMessage = "Start getting of updates"
const val serverWebsocketHelloMessage = "Accepted"
const val serverWebsocketNewMessageMessage = "NewMessage"
const val websocketFinalizationMessage = "Final"

View File

@ -17,6 +17,7 @@ kotlin {
jvmMain { jvmMain {
dependencies { dependencies {
api "io.ktor:ktor-server:$ktor_version" api "io.ktor:ktor-server:$ktor_version"
api "io.ktor:ktor-server-cio:$ktor_version"
api "io.ktor:ktor-server-host-common:$ktor_version" api "io.ktor:ktor-server-host-common:$ktor_version"
api "io.ktor:ktor-websockets:$ktor_version" api "io.ktor:ktor-websockets:$ktor_version"
} }

View File

@ -1,8 +1,7 @@
package dev.inmo.micro_utils.ktor.server package dev.inmo.micro_utils.ktor.server
import dev.inmo.micro_utils.coroutines.safely import dev.inmo.micro_utils.coroutines.safely
import dev.inmo.micro_utils.ktor.common.CorrectCloseException import dev.inmo.micro_utils.ktor.common.*
import dev.inmo.micro_utils.ktor.common.standardKtorSerialFormat
import io.ktor.http.cio.websocket.* import io.ktor.http.cio.websocket.*
import io.ktor.routing.Route import io.ktor.routing.Route
import io.ktor.websocket.webSocket import io.ktor.websocket.webSocket
@ -20,15 +19,16 @@ private suspend fun DefaultWebSocketSession.checkReceivedAndCloseIfExists() {
fun <T> Route.includeWebsocketHandling( fun <T> Route.includeWebsocketHandling(
suburl: String, suburl: String,
flow: Flow<T>, flow: Flow<T>,
converter: (T) -> ByteArray converter: (T) -> StandardKtorSerialInputData
) { ) {
webSocket(suburl) { webSocket(suburl) {
// println("connected")
safely { safely {
flow.collect { flow.collect {
checkReceivedAndCloseIfExists()
send(converter(it)) send(converter(it))
} }
} }
// println("disconnected")
} }
} }
@ -40,5 +40,5 @@ fun <T> Route.includeWebsocketHandling(
suburl, suburl,
flow flow
) { ) {
standardKtorSerialFormat.encodeToByteArray(serializer, it) standardKtorSerialFormat.encodeDefault(serializer, it)
} }

View File

@ -0,0 +1,10 @@
package dev.inmo.micro_utils.ktor.server
import dev.inmo.micro_utils.pagination.*
import io.ktor.http.Parameters
val Parameters.extractPagination: Pagination
get() = SimplePagination(
get("page") ?.toIntOrNull() ?: 0,
get("size") ?.toIntOrNull() ?: defaultMediumPageSize
)

View File

@ -1,8 +1,10 @@
package dev.inmo.micro_utils.ktor.server package dev.inmo.micro_utils.ktor.server
import dev.inmo.micro_utils.ktor.common.standardKtorSerialFormat import dev.inmo.micro_utils.coroutines.safely
import dev.inmo.micro_utils.ktor.common.*
import io.ktor.application.ApplicationCall import io.ktor.application.ApplicationCall
import io.ktor.http.HttpStatusCode import io.ktor.http.HttpStatusCode
import io.ktor.request.receive
import io.ktor.response.respond import io.ktor.response.respond
import io.ktor.response.respondBytes import io.ktor.response.respondBytes
import io.ktor.util.toByteArray import io.ktor.util.toByteArray
@ -12,18 +14,20 @@ suspend fun <T> ApplicationCall.unianswer(
answerSerializer: SerializationStrategy<T>, answerSerializer: SerializationStrategy<T>,
answer: T answer: T
) { ) {
respondBytes( respondBytes (
standardKtorSerialFormat.encodeToByteArray(answerSerializer, answer), standardKtorSerialFormat.encodeDefault(answerSerializer, answer),
standardKtorSerialFormatContentType standardKtorSerialFormatContentType
) )
} }
suspend fun <T> ApplicationCall.uniload( suspend fun <T> ApplicationCall.uniload(
deserializer: DeserializationStrategy<T> deserializer: DeserializationStrategy<T>
) = standardKtorSerialFormat.decodeFromByteArray( ) = safely {
standardKtorSerialFormat.decodeDefault(
deserializer, deserializer,
request.receiveChannel().toByteArray() receive()
) )
}
suspend fun ApplicationCall.getParameterOrSendError( suspend fun ApplicationCall.getParameterOrSendError(
field: String field: String
@ -49,7 +53,7 @@ fun <T> ApplicationCall.decodeUrlQueryValue(
field: String, field: String,
deserializer: DeserializationStrategy<T> deserializer: DeserializationStrategy<T>
) = getQueryParameter(field) ?.let { ) = getQueryParameter(field) ?.let {
standardKtorSerialFormat.decodeFromHexString( standardKtorSerialFormat.decodeHex(
deserializer, deserializer,
it it
) )

View File

@ -0,0 +1,33 @@
package dev.inmo.micro_utils.ktor.server
import io.ktor.application.Application
import io.ktor.server.cio.CIO
import io.ktor.server.engine.*
import kotlin.random.Random
fun <TEngine : ApplicationEngine, TConfiguration : ApplicationEngine.Configuration> createKtorServer(
engine: ApplicationEngineFactory<TEngine, TConfiguration>,
host: String = "localhost",
port: Int = Random.nextInt(1024, 65535),
block: Application.() -> Unit
): TEngine {
val env = applicationEngineEnvironment {
module(block)
connector {
this@connector.host = host
this@connector.port = port
}
}
return embeddedServer(engine, env)
}
/**
* Create server with [CIO] server engine without starting of it
*
* @see ApplicationEngine.start
*/
fun createKtorServer(
host: String = "localhost",
port: Int = Random.nextInt(1024, 65535),
block: Application.() -> Unit
): ApplicationEngine = createKtorServer(CIO, host, port, block)