Kotlin Flow Retry y RetryWhen Extension Functions | Via SagaRock101 | Agosto de 2021

En este artículo, escribiré un artículo sobre reintentar llamadas de red usando extensiones de transmisión de Kotlin

Porqué necesitamos esto

En el caso

• Donde hay conexiones de ancho de banda bajo y conexiones de alta latencia
• Falló la autenticación del token
• Red incierta

Debido a la situación anterior, debemos reintentar las llamadas de red a las API importantes, como el inicio de sesión, el registro, el inicio de sesión (o cualquier otra API según la demanda y la importancia) varias veces. De lo contrario, esto puede afectar la experiencia del usuario.

¿Qué es el tráfico?

• Es un flujo de datos desde el flujo ascendente (entrada) al flujo descendente (salida), que se puede calcular de forma asincrónica.
• En este caso, el flujo ascendente es la respuesta de la red y el flujo descendente es donde mostramos la respuesta en la interfaz de usuario (actividad / fragmento)

¿Qué es una corrutina?

Son hilos ligeros.

¿Qué es la función de pausa?

Estas son funciones que pueden realizar tareas de larga duración y esperar a que se complete sin bloquear

Necesito confiar en

Para Kotlin coroutine android

En el momento de escribir este artículo, la versión de este artículo es 1.5.1

Implementar ‘org.jetbrains.kotlinx: kotlinx-coroutines-android: 1.5.1’

Rever

voluntad Dé un ejemplo simple para mostrar cómo podemos reintentar una tarea normal de larga ejecución. Más adelante se proporcionará un ejemplo para ilustrar cómo podemos usar esta función para reintentar una llamada a la API usando la arquitectura MVVM

ejemplo:

fun startTask(): Flow<Int> 
return flow 
for (i in 1..4) 
val randomInt = (0..2).random()
if (randomInt == 0) 
throw IndexOutOfBoundsException()
 else if (randomInt == 2) 
throw  IOException()

emit(i)

.flowOn(Dispatchers.IO)

La función anterior devuelve un flujo de valores Int de 1 a 4 y arroja una excepción a los enteros 0 y 2, que es aleatorio.En el contexto de Dispatcher.IO, este es un programador de rutina diseñado para descargar tareas de bloqueo a un grupo de subprocesos compartidos

Utilizo flowOn (Dispatchers.IO) para simular escenarios reales, porque generalmente las operaciones de larga duración se realizan en subprocesos en segundo plano.

Suponiendo que el método anterior es una llamada a la API en tiempo real, a veces necesitamos volver a intentar la llamada a la API para obtener una respuesta.

Nosotros podemos usar Rever() Funciones de extensión de transmisión como esta

suspend fun main(args: Array<String>)     startTask()
.retry(3) 
println("retrying...")
delay(2000)
it is IndexOutOfBoundsException
.catch 
print(it.toString())

.collect 
println(it)

• El método anterior es recopilar un flujo de valores int desde el flujo ascendente, pero también existe la ayuda de operadores de reintento y captura
• Esta Rever() Acepta un parámetro opcional de tipo Long, que representa el número de veces que se reintenta el flujo ascendente hasta que se produce una excepción en el flujo ascendente.Si no se proporciona, se utilizará el valor Long.MAX_VALUE predeterminado
• Una función de suspensión con throwable como parámetro y tipo de retorno booleano (si no se proporciona) usará el valor predeterminado verdadero
• Esta captura() Capture la excepción en el flujo ascendente y llame a la acción especificada
• ya que Rever() Es una función de suspensión. Podemos introducir un retraso entre las llamadas de reintento para obtener algo de tiempo de búfer con la ayuda de Retraso (tiempo milisegundos: largo) Esta es también una función de suspensión.

La salida del fragmento de código anterior

retrying...
retrying...
retrying...
1
java.io.IOException
Process finished with exit code 0

Se produjo una serie de excepciones en el flujo ascendente, por lo que el mensaje «reintentando …» significa reintentar el flujo ascendente

No hay excepciones en el flujo ascendente, así que imprima 1

Más tarde, apareció una excepción diferente de la excepción especificada en la expresión lambda en el flujo ascendente, y el número de reintentos estaba vacío, así que use catch para capturarlo e imprimir la excepción capturada

El siguiente código se toma del archivo Error.kt del paquete kotlinx.coroutines.flow

public fun <T> Flow<T>.retry(
retries: Long = Long.MAX_VALUE,
predicate: suspend (cause: Throwable) -> Boolean =  true 
): Flow<T> 
require(retries > 0)  "Expected positive amount of retries, but had $retries" 
return retryWhen  cause, attempt -> attempt < retries && predicate(cause) 

Rever() Intercomunicador Tiempo de reintento () Solo funciona después de verificar si el número de reintentos es mayor que cero; de lo contrario, se lanzará una IllegalArgumentException. Esto cancelará el proceso y mostrará el mensaje «El número esperado de reintentos es positivo, pero hay $ reintentos».

Conecte el ejemplo anterior [predicate] Es una expresión lambda

println("retrying...")
delay(2000)
it is IndexOutOfBoundsException

eso ¿Se pasa el elemento arrojadizo en el predicado: Pausa (Razón: Lanzable) Llamar internamente Rever()

Tiempo de reintento

Salga de la colección de flujos ascendentes hasta que se produzca una excepción en el flujo ascendente y la función de predicado devuelva verdadero.

Modifique el ejemplo anterior para usar Tiempo de reintento ()

startTask()
.retryWhen .catch 
print(it.toString())

.collect 
println(it)

La salida del código anterior está truncada.

retrying...
1
retrying...
retrying...
retrying...
java.lang.IndexOutOfBoundsException

Esta es la primera vez que el flujo ascendente tiene una excepción, por lo que el mensaje «reintentando …»

1 se imprime porque no hay ninguna excepción en sentido ascendente

Inicialmente, el intento será 0 y, dado que 0 <3, devuelve verdadero a la expresión lambda, lo que indica que vuelva a intentarlo en sentido ascendente hasta que el intento llegue a 3

attempt < 3

Veamos como adentro Tiempo de reintento () Ser ejecutado.

A continuación se muestra el código obtenido del archivo Error.kt del paquete kotlinx.coroutines.flow

public fun <T> Flow<T>.retryWhen(predicate: suspend FlowCollector<T>.(cause: Throwable, attempt: Long) -> Boolean): Flow<T> =
flow 
var attempt = 0L
var shallRetry: Boolean
do 
shallRetry = false
val cause = catchImpl(this)
if (cause != null) 
if (predicate(cause, attempt)) 
shallRetry = true
attempt++
 else 
throw cause


 while (shallRetry)

• Tiempo de reintento () Intente utilizar una función de suspensión con dos parámetros, escriba throwable y escriba Long
• El motivo es que se produjo una excepción en el flujo ascendente
• Los intentos son el número de intentos de reintentar el flujo ascendente
• Esta función devuelve la secuencia a la corriente descendente
• Probó una variable interna que se incrementará dentro del ciclo do-while
• Tiene shouldRetry, que también es una variable interna y una bandera booleana, lo que permite que el bucle se ejecute hasta que la función lambda devuelva falso (como Predicado (razonar, intentar))
• catchImp () Reciba la función de la interfaz FlowCollector. Verifique y devuelva excepciones o nulos desde el origen.
• Al comprobar el flujo de aguas arriba anormal catchImp () cada vez Utilice la función de suspensión de recopilación de la interfaz de flujo para activar internamente el flujo ascendente.

El primer intento es 0, shouldRetry es falso

Suponga que hay una excepción en sentido ascendente y luego la excepción (como razón) Pasado a la función lambda (p. Ej. predicado).Conéctese con el ejemplo, esta será una expresión lambda

 cause is IOException

if (predicate(cause, attempt)) 
shallRetry = true
attempt++
 else 
throw cause

@GET("top-headlines")
suspend fun getTopHeadlines(
@Query("country") country: String
): TopHeadlines

suspend fun getNewsHeadLines(country: String): Flow<TopHeadlines> 
return flow 
emit(newsHeadlinesService.getTopHeadlines(country))
.flowOn(Dispatchers.IO)

fun getNewsHeadlines(country: String): LiveData<DataWrapper<TopHeadlines>> 
return liveData 
emit(DataWrapper.loading(null))
remoteSource.getNewsHeadLines(country).retry(3) 
delay(NETWORK_RETRY_DELAY)
return@retry true
.catch 
emit(DataWrapper.error(it.message.toString()))
.collect 
emit(DataWrapper.success(it))

val newsHeadLinesLD = newsHeadLinesMLD.switchMap 
if (it.category.isEmpty())
newsHeadlinesRepo.getNewsHeadlines(it.country)
else 
newsHeadlinesRepo.getNewsHeadlines(it.country, it.category)

viewModel.newsHeadLinesLD.observe(viewLifecycleOwner, Observer {
when (it.status) 
DataWrapper.Status.LOADING -> 
DataWrapper.Status.SUCCESS -> 
DataWrapper.Status.ERROR -> 
)