Primeros pasos en programación reactiva, I

En esta serie de post acerca de programación reactiva voy a ir contando los pasos que estoy dando para ir practicando con esta forma de programación y más en concreto su uso en aplicaciones HTTP con Micronaut

Básicamente mediante la programación reactiva lo que conseguimos es no bloquear los hilos de ejecución mientras se espera a que se complete una tarea.

Esta forma de programación se puede realizar en ambos lados del diálogo, es decir:

• Podemos hacer que el cliente invoque llamadas al servidor y continue con sus cosas hasta recibir una respuesta de este.

• Podemos usarlo en el servidor de tal forma que este pueda aceptar múltiples peticiones sin bloquear el hilo http, ejecutar la solicitud en otro hilo y responder al cliente cuando se complete.

Casi todos los lenguajes tienen una forma u otra de permitir esta programación, así como diferentes librerías o frameworks. En estos post vamos a usar Micronaut el cual usa RxJava2 (hasta la versión 3 en la que se ha migrado a Reactor) y lo vamos a ver tanto en el cliente como en el servidor

Servicio bloqueante

A continuación podemos ver un controller típico:

@Controller("/numbers")
public class NumberGeneratorController {

    @Get("/list{?size}")
    int[] list(Optional<Integer> size){
        Random rnd = new Random();
        return IntStream.
                range(1,size.orElse(99)+1).
                map( i-> Math.abs(rnd.nextInt())).
                toArray();
    }
}

Simplemente definimos un Controller con un método /list al que se le puede pasar un parámetro opcional, size, y que genera una lista de números de forma aleatoria

Este endoint probablemente no bloquee demasiado el thread y no veamos la diferencia con un reactivo pero es un buen punto de inicio.

Servicio reactivo

La versión reactiva más simple y parecida, sería algo como:

@Controller("/numbers")
public class NumberGeneratorController {

    @Get("/list{?size}")
    Single<int[]> list(Optional<Integer> size){
        Random rnd = new Random();
        return Single.just(
                IntStream.
                range(1,size.orElse(99)+1).
                map( i-> Math.abs(rnd.nextInt())).
                toArray();
        );
    }
}

Simplemente hemos cambiado el tipo de retorno (recubriéndolo con un Single) por lo que devolvemos un objeto de ese tipo que lo único que hace es ejecutar el mismo código. La diferencia entre ambos es que Micronaut, cuando devolvemos un objeto de este tipo, lo que hace es utilizar la librería reactiva para que sea ella la que lo complete. En el caso simple, por el contrario, se está ejecutando en el mismo hilo que hace la petición y si recibimos muchas peticiones, a este u otro endpoint, se terminará bloqueando.

INFO

Lo mejor de este ejemplo es que para el cliente que invoca la petición ambas implementaciones son iguales en cuanto a la forma de invocarlas se refiere.

Descargar Imagen de forma reactiva

Cuando la tarea a ejecutar por el servidor es más pesada, como servir una imagen por ejemplo, la programación reactiva nos permite optimizar su ejecución. Por ejemplo, a continuación se muestra un Controller que permite descargar un Gif. En lugar de leer todo el fichero y enviarlo como un array de bytes en la misma petición lo que va a usar es un Flowable

Un objeto del tipo Flowable nos va a permitir ir enviando bloques de información al cliente según vayamos obteniéndola. Una vez completada la tarea notificaremos al Flowable que ya se ha completado y esto terminará el diálogo con el cliente:

@Controller("/test")
class FileController{

    @Get(value = "/{file}", produces = MediaType.IMAGE_GIF)
    Flowable<byte[]> image(@PathVariable String file) {

      Flowable.create({ emitter ->
          new File(file).withInputStream{ inputStream ->
                int size=1024
              byte[]buff = inputStream.readNBytes(size)
              while( buff.length == size){
                  emitter.onNext(buff)
                  buff = inputStream.readNBytes(size)
              }
              emitter.onNext(buff)
              emitter.onComplete()
          }
      }, BackpressureStrategy.BUFFER)
  }
}

En este ejemplo, el cliente solicita un fichero y lo que devuelve el controller es un objeto Flowable. Cuando el subsistema se encuentre preparado se subscribirá a este y le proporcionará un objeto emitter el cual nos va a servir para ir enviando "chunks" de información:

• abrimos el fichero y vamos leyendo bloques de 1024 bytes

• se los vamos pasando al emitter mediante onNext. El subsistema es el encargado de ir enviándolos a su vez usando la estrategia que hayamos especificado (BUFFER en este caso)

• una vez enviados los últimos bytes notificamos que hemos terminado con onComplete

Mediante esta técnica podemos servir ficheros pesados a multitud de clientes sin bloquear las llamadas. (He empleado este bot en una página donde podías descargar las imágenes de tráfico de la DGT y una simple aplicación era capaz de servir cientos de imágenes de forma fluida)

Resumen

El resumen de este primer post es:

• la programación reactiva nos ayuda a optimizar los recursos

• a un nivel básico no añaden más complejidad

• nuevos jugadores como Single, Flowable y Emitter