La programación reactiva es una de las tendencias hoy en día, en diversos sectores de la industria del desarrollo de software, gracias a iniciativas como ReactiveX, un movimiento que busca llevar el paradigma a varios lenguajes de programación (sean o no funcionales), y que no necesariamente corran en un servidor; como es el caso de las extensiones para JavaScript, el nuevo lenguaje de Apple multiplataforma Swift o Bindings para frameworks específicios como Android.
La programación reactiva es un paradigma orientado al flujos de datos y a la propagación de cambios. La palabra reactiva hace referencia a la reacción, al comportamiento que va a realizar nuestra aplicación en base a un dato, y a partir de los cambios realizados en este dato, cambia también el comportamiento. Esto quiere decir que nos centramos en que los componentes reaccionen y responda de acuerdo a las intreacciones que se realizan con el sistema que se esté desarrollando. Los datos que utilizan las aplicaciones pueden ser muy variados desde un flujo de datos desde un archivo local, de una pagina web, notificaciones del sistema o hasta una serie de eventos tales como entradas del usuario. En la programación reactiva se representan todos estos datos como secuencias de observables y la aplicación puede suscribirse a estas secuencias para recibir notificaciones a medida que llegan nuevos datos.
Sí, en el caso concreto de RxJava, se pueden encontrar conceptos como Eventos, Observadores, pero se pueden contar con secuencias de datos o streams, componer secuencias de forma declarativa las cuales puedan operar sobre los datos y abstraer conceptos como manipulación de thread, sincronizaciones cuando se acceden a los datos, hilos seguros y estructuras de datos para ser usadas en ambientes concurrentes.
Podemos usar Rx extensions, cuando se quiera orquestar aplicaciones basados en eventos y que estos sucedan de forma asíncrona. Por lo general cuando se manejan muchos eventos o llamadas de tipo asincrónicas, el código empieza a perder su legibilidad y mantenerlo se hace complicado, recurriendo a estados globales o uso de algoritmos para manejar las peticiones y el orden de las mismas, además de tener que implementar Callbacks para comunicar si fué exitosa el resultado o si por el contrario, hay fracaso. Al usar Rx extensions estamos haciendo un código mucho mas sencillo que nos permite leer y componer apis que manejen de una forma más práctica las llamadas asíncronas.
(Sí, posiblemente los callbacks sean el nuevo GOTO, que vuelva el BASIC por favor!)
Si la aplicación que estamos desarrollando tiene muy pocas llamadas asíncronas y operaciones basadas en eventos o las mismas no tienen un nivel de complejidad alta podría ser más tormentoso a migrar al uso de la extensión y tal vez su costo - beneficio no lo amerita.
### Ventajas de el uso de extensiones reactivas.
- Facilidad y elegancia al tratar los eventos y los errores.
- Facilidad al diseñar y construir APIS concurrentes.
- Testeabilidad. Como todo lo que sea concurrente, pero existen diversos frameworks y librerías que facilitan esto.
- Curva de aprendizaje de la muerte. No mayor a la de aprender Scala, Haskell... o Lisp O_o.
Los conceptos básicos de un código reactivo son Observables y Subscribers. Un Observable emite items (conocido como stream) y un Subscriber consume estos items.
Hay una manera en cómo se emiten los datos. un Observable puede emitir cualquier cantidad de elementos (se pueden emitir cero items si así lo desea), y cuando esta emisión termina, puede darse por completa de manera exitosa, o la emisión puede finalizar debido a un error. Para cada Subscriber suscrito, el Observable llamará el método Subscriber.onNext(T t), tantas veces como tantos items sean emitidos, seguido de el Subscriber.onCompleted(), que indica la finalización de emisión de items o un Subscriber.OnError() para el caso de emision de errores.
A pesar de lucir como un standard observer pattern, e incluso apoyarse del mismo, el Observable no empieza la emisión de items hasta que no hayan subscriciones al mismo. En otras palabras: si no hay nadie escuchando, no hay por qué decir algo.
RxJava, es librería o implementación de una extensión reactiva escrita en Java, cuya finalidad es componer programas de tipo asíncrono y basados en evento mediante el uso de sequencias de observables.
A continuación, vamos a ver algo de código y el uso del framework de RxJava con un Hello world.
Observable<String> helloWorldObservable = Observable.create(
new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello, world...!");
subscriber.onCompleted();
}
}
);En este caso en particular, nuestro Observable está emitiendo un dato de tipo String y a su vez, notifica que la emisión del stream o el flujo de datos finalizó, por medio del método onCompleted(). Ahora crearemos un Subscriber que consumirá esta data y será notificado de el status de la misma.
Subscriber<String> helloWorldSubscriber = new Subscriber<String>() {
@Override
public void onNext(String s) { System.out.println(s); }
@Override
public void onCompleted() { System.out.println("Stream emission completed!")}
@Override
public void onError(Throwable e) { }
};Ya que tenemos escrito el Observable y el Subscriber, podemos realizar la conexión de la siguiente manera:
helloWorldObservable.subscribe(helloWorldSubscriber);En este caso, la salida que tenemos será la siguiente:
Hello, world...!
Stream emission completed
Cuando se finaliza la subscripción a un Observable, este realiza un llamado a el método onNext(String s) del Subscriber emitiendo el stream y paso siguiente, realiza un llamado al método onCompleted() para comunicar el fin de la emisión de la data. En este punto el desarrollador puede tomar acciones con respecto a su implementación ya que cuenta con la posibilidad de saber que no hay más datos que consumir para este Observable.
Qué hay de malo en este código?. Podría mejorarse si RxJava nos lo permite?, no es un poco exagerado sólo para realizar un Hello world?
En RxJava, contamos con una serie de atajos para realizar emisiones simples como en el caso del Hello world.
Primero, podemos simplificar nuestro Observable, siendo posible ya que la librería cuenta con una serie de common methods, como lo es Observable.just(), el cual emite un item y notifica el la completa emisión del mismo:
Subscriber<String> helloWorldSubscriber = Observable.just("Hello, world...!")