add opportunity for reconnection for websockets
This commit is contained in:
parent
13b6dad2fb
commit
8b74e8534c
@ -4,41 +4,60 @@ import com.insanusmokrassar.postssystem.ktor.*
|
|||||||
import com.insanusmokrassar.postssystem.utils.common.safely
|
import com.insanusmokrassar.postssystem.utils.common.safely
|
||||||
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.client.request.url
|
|
||||||
import io.ktor.http.HttpMethod
|
|
||||||
import io.ktor.http.cio.websocket.*
|
import io.ktor.http.cio.websocket.*
|
||||||
import kotlinx.coroutines.flow.*
|
import kotlinx.coroutines.flow.*
|
||||||
import kotlinx.coroutines.isActive
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @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> createStandardWebsocketFlow(
|
inline fun <reified T> createStandardWebsocketFlow(
|
||||||
client: HttpClient,
|
client: HttpClient,
|
||||||
url: String,
|
url: String,
|
||||||
|
crossinline checkReconnection: (Throwable?) -> Boolean = { true },
|
||||||
crossinline conversation: suspend (ByteArray) -> T
|
crossinline conversation: suspend (ByteArray) -> T
|
||||||
): Flow<T> {
|
): Flow<T> {
|
||||||
val correctedUrl = url.asCorrectWebSocketUrl
|
val correctedUrl = url.asCorrectWebSocketUrl
|
||||||
|
|
||||||
return channelFlow {
|
return channelFlow {
|
||||||
val producerScope = this
|
val producerScope = this
|
||||||
safely(
|
do {
|
||||||
{
|
val reconnect = try {
|
||||||
producerScope.close()
|
safely(
|
||||||
throw it
|
{
|
||||||
}
|
throw it
|
||||||
) {
|
}
|
||||||
client.ws(
|
) {
|
||||||
correctedUrl
|
client.ws(
|
||||||
) {
|
correctedUrl
|
||||||
while (true) {
|
) {
|
||||||
when (val received = incoming.receive()) {
|
while (true) {
|
||||||
is Frame.Binary -> producerScope.send(
|
when (val received = incoming.receive()) {
|
||||||
conversation(received.readBytes())
|
is Frame.Binary -> producerScope.send(
|
||||||
)
|
conversation(received.readBytes())
|
||||||
else -> {
|
)
|
||||||
producerScope.close()
|
else -> {
|
||||||
return@ws
|
producerScope.close()
|
||||||
|
return@ws
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
checkReconnection(null)
|
||||||
|
} catch (e: Throwable) {
|
||||||
|
checkReconnection(e).also {
|
||||||
|
if (!it) {
|
||||||
|
producerScope.close(e)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} while (reconnect)
|
||||||
|
if (!producerScope.isClosedForSend) {
|
||||||
|
safely(
|
||||||
|
{ /* do nothing */ }
|
||||||
|
) {
|
||||||
|
producerScope.close()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -47,7 +47,8 @@ class WebsocketsTest {
|
|||||||
}
|
}
|
||||||
val incomingWebsocketFlow = createStandardWebsocketFlow(
|
val incomingWebsocketFlow = createStandardWebsocketFlow(
|
||||||
client,
|
client,
|
||||||
"$serverUrl/$suburl"
|
"$serverUrl/$suburl",
|
||||||
|
{ false } // always skip reconnection
|
||||||
) {
|
) {
|
||||||
standardKtorSerializer.load(Int.serializer(), it)
|
standardKtorSerializer.load(Int.serializer(), it)
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user