Programación Reactiva con Spring

La programación reactiva es un paradigma orientado a manejar flujos de datos de manera asíncrona, lo cual resulta ideal para aplicaciones con alta carga de tráfico, como sistemas de microservicios. Con Spring, la programación reactiva se facilita mediante el módulo Spring WebFlux.

Spring WebFlux es el módulo reactivo de Spring que permite construir aplicaciones no bloqueantes y asíncronas. Está basado en el patrón Reactor, que sigue el estándar de Reactive Streams y utiliza Mono y Flux como las abstracciones principales para manejar uno o varios elementos, respectivamente.

• Mono: Representa 0 o 1 elemento.
• Flux: Representa 0 o N elementos.
• Backpressure: Mecanismo para controlar la velocidad de emisión de los eventos.

Primero, crea un proyecto de Spring Boot en Spring Initializr. Asegúrate de agregar las dependencias de Spring Reactive Web.

<dependency>

    <groupId>org.springframework.boot</groupId>

    <artifactId>spring-boot-starter-webflux</artifactId>

</dependency>

A continuación, implementamos un controlador que responde a peticiones HTTP de manera reactiva usando `Flux` y `Mono`.

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.PathVariable;

import org.springframework.web.bind.annotation.RestController;

import reactor.core.publisher.Flux;

import reactor.core.publisher.Mono;

import java.time.Duration;

@RestController

public class ReactiveController {

    // Endpoint que devuelve un solo valor (Mono)

    @GetMapping("/mono/{id}")

    public Mono<String> getMono(@PathVariable String id) {

        return Mono.just("Valor recibido: " + id);

    }

    // Endpoint que devuelve un flujo continuo de datos (Flux)

    @GetMapping("/flux")

    public Flux<Integer> getFlux() {

        return Flux.range(1, 10)

                   .delayElements(Duration.ofSeconds(1));

    }

}

Una vez que hayas implementado el controlador, ejecuta la aplicación de Spring Boot. Puedes acceder a los endpoints:

• /mono/{id}: Devuelve un único valor. Ejemplo: http://localhost:8080/mono/5
• /flux: Devuelve un flujo de datos continuo que se emite cada segundo. Ejemplo: http://localhost:8080/flux

Como ventajas podemos ver: 

• Eficiencia y Escalabilidad: Ideal para manejar aplicaciones que requieren alta concurrencia y asíncronas.
• Control del Flujo: Con Reactive Streams puedes controlar el ritmo de procesamiento y evitar la sobrecarga del sistema.
• Menos Bloqueo: No hay hilos bloqueados mientras se esperan respuestas, lo que mejora el rendimiento en sistemas de alta demanda.

Definiendo Observable y Observer

En Reactive programming un Observable es un objeto que emite una secuencia (o flujo) de eventos. Representa una colección basada en inserción, que es una colección en la que se insertan eventos cuando se crean.

Un observable emite una secuencia que puede ser vacía, finita o infinita. Cuando la secuencia es finita, se emite un evento completo después del final de la secuencia. En cualquier momento durante la emisión (pero no después de su finalización) se puede emitir un evento de error, deteniendo la emisión y cancelando la emisión del evento completo.

Cuando la secuencia está vacía, solo se emite el evento completo, sin emitir ningún ítem. Con una secuencia infinita, el evento completo nunca se emite.

La emisión se puede transformar, filtrar o combinar con otras emisiones, etc. 

Un observador es un objeto que se suscribe a un observable. Escucha y reacciona a cualquier secuencia de elementos emitida por el Observable.

El Observer no está bloqueado mientras espera nuevos elementos emitidos, por lo que en operaciones simultáneas, no se produce ningún bloqueo. Simplemente se activa cuando se emite un nuevo elemento.

Este es uno de los principios fundamentales de la programación reactiva: en lugar de ejecutar instrucciones una a la vez (siempre esperando a que se complete la instrucción anterior), el observable proporciona un mecanismo para recuperar y transformar datos, y el Observer activa este mecanismo, todos de forma concurrente.

El siguiente pseudocódigo es un ejemplo del método que implementa el Observer que reacciona a los elementos del Observable:

onNext = { it -> doSomething }

Aquí, el método está definido, pero no se invoca nada. Para comenzar a reaccionar, debe suscribirse al Observable:

observable.subscribe(onNext)

Ahora el observador está atento a los elementos y reaccionará a cada elemento nuevo que se emitirá.

Reescribamos este ejemplo en código Java usando las API de RxJava:

public void subscribeToObservable(Observable<T> observable) {

    observable.subscribe(nextItem -> {

        // invoked when Observable emits an item

        // usually you will consume the nextItem here

    });

}

Ahora está claro que para conectar un observable con un observador, debes usar el método de suscripción.

ReactiveX

ReactiveX es un framework para los lenguajes de programación más utilizados: Java, JavaScript, C#, Scala, Clojure, C ++, Ruby, Python, Groovy, JRuby, Kotlin, Swift y más. 

RxJava es un framwork que implementa los conceptos de ReactiveX en Java. Veamos un ejemplo RxJava:

List<Integer> input = Arrays.asList(1, 2, 3, 4, 5);

Observable.from(input).filter(new Func1() {

    @Override

     public Boolean call(Integer x) {

          return x % 2 == 0;

     }

})

o usando lambda : 

Observable.from(input).filter(x -> x % 2 == 0);

El objeto resultante (la instancia de rx.Observable) generará una secuencia de los números pares contenidos en la secuencia de entrada: 2 y 4.

En RxJava, rx.Observable agrega dos semánticas al patrón Observador de Gang of Four (la semántica predeterminada es emitir elementos creados, como una lista con elementos 2,4 en el ejemplo anterior):

• El productor puede notificar al consumidor que no hay más datos disponible.
• El productor puede notificar al consumidor que ha ocurrido un error.

La biblioteca RxJava proporciona un modelo de programación donde podemos trabajar con eventos generados desde UI o llamadas asincrónicas de la misma manera en que operamos con colecciones y streams en Java 8.

La biblioteca RxJava se creó en Netflix como una alternativa más inteligente a Java Futures y devoluciones de llamada. Tanto los futuros como las devoluciones de llamada son fáciles de usar cuando solo hay un nivel de ejecución asincrónica, pero son difíciles de administrar cuando están anidados.

El siguiente ejemplo muestra cómo se maneja el problema de las devoluciones de llamada anidadas en RxJava.

Suponga que necesita llamar a una API remota para autenticar a un usuario, luego a otra para obtener los datos del usuario y a otra API para obtener los contactos de un usuario. Normalmente, tendría que escribir llamadas a API anidadas y hacer complejos callbacks. Pero con RxJava se puede hacer así : 

serviceEndpoint.login().doOnNext(accessToken -> storeCredentials(accessToken)).flatMap(accessToken -> serviceEndpoint.getUser()).flatMap(user -> serviceEndpoint.getUserContact(user.getId()))

El patrón observador en la programación reactiva

El patrón Observer es un patrón de diseño en el que hay dos tipos de objetos: observadores y sujetos observables. Un observador es un objeto que observa los cambios de uno o más sujetos; un sujeto es un objeto que mantiene una lista de sus observadores y les notifica automáticamente cuando cambia de estado.

La definición del patrón Observer del libro "Gang of Four" (Design Patterns: Elements of Reusable Object-Oriented Software por Erich Gamma, Richard Helm, Ralph Johnson y John Vlissides, ISBN 0-201-63361-2) es "Defina una dependencia de uno a varios entre objetos para que cuando un objeto cambie de estado, todos sus dependientes sean notificados y actualizados automáticamente".

Este patrón es el núcleo de la programación reactiva. Se ajusta perfectamente al concepto de programación reactiva al proporcionar las estructuras para implementar el mecanismo produce/react

Java SDK implementa el patrón Observer con la clase java.util.Observable y la interfaz java.util.Observer.

class Subject extends java.util.Observable {

    public void doWorkAndNotify() {

        Object result = doWork();

        notifyObservers(result);

    }

}

class MyObserver implements Observer {

    @Override

    public void update(Observable obs, Object item) {

        doSomethingWith(item)

    }

}

La clase Subject extiende java.utils.Observable y es responsable de producir un objeto y notificar a los observadores tan pronto como se haya producido el elemento.

MyObserver implementa Observer y es responsable de observar al sujeto y consumir todos los elementos que produce el sujeto.

Veamos como interactuan estos objetos: 

MyObserver myObserver = new MyObserver();

Subject subject = new Subject();

subject.addObserver(myObserver);

subject.doWorkAndNotify();

Desafortunadamente, esta implementación se revela demasiado simple cuando comienza a escribir una lógica más compleja. No se utilizan estas implementaciones.

Programación reactiva

La programación reactiva lleva la programación funcional un poco más allá, agregando el concepto de flujo de datos y propagación de cambios de datos.

En programación imperativa, se puede asignar un valor a una variable de la siguiente manera:

x = y + z

Aquí, la suma de y+z se asignará a la variable x , si llamamos esta función, los cambios que ocurran a y y a z no modificaran a x.

En la programación reactiva, el valor de x debe actualizarse siempre que cambien los valores de y o z.

Entonces, si los valores iniciales son y = 1 y z = 1, tendrás

x = y + z = 2.

Si y (o z) cambia su valor, esto no significa que x cambia automáticamente, pero debe implementar un mecanismo para actualizar los valores de x cuando se cambian los valores de y y z.

La programación reactiva funcional es un nuevo paradigma de programación; Erik Meijer lo hizo popular (quien creó la biblioteca Rx para .NET cuando trabajaba en Microsoft) y se basa en dos conceptos:

• El código "reacciona" a los eventos.
• El código maneja los valores a medida que varían en el tiempo, propagando los cambios a cada parte del código que usa esos valores.

La clave para entender la programación reactiva es pensar en ella como si operara en un flujo de datos.

Pero, ¿qué quiero decir con "flujo de datos"? Se refiere a una secuencia de eventos, donde un evento podría ser una entrada del usuario (como un toque en un botón), una respuesta de una solicitud de API (como un feed de Facebook), datos contenidos en una colección o incluso una sola variable.

En la programación reactiva, a menudo hay un componente que actúa como fuente, emitiendo una secuencia de elementos (o un flujo de datos) y algunos otros componentes que observan este flujo de elementos y reaccionan a cada elemento emitido ("reaccionan" al elemento emisión).

Imperativo vs reactivo

Para comprender mejor el contraste entre los 2 mundos, necesitamos explicar la diferencia entre los modelos de ejecución reactiva e imperativa. 

En el enfoque tradicional e imperativo, los frameworks asignan un hilo para manejar una solicitud. Entonces, todo el procesamiento de la solicitud se ejecuta en este hilo de trabajo. Este modelo no escala muy bien. De hecho, para manejar múltiples solicitudes simultáneas, necesita múltiples subprocesos; por lo que la simultaneidad de su aplicación está limitada por el número de subprocesos. Además, estos subprocesos se bloquean tan pronto como su código interactúe con servicios remotos. Por lo tanto, conduce a un uso ineficiente de los recursos, ya que es posible que necesite más subprocesos, y cada subproceso, ya que están asignados a subprocesos del sistema operativo, tiene un costo en términos de memoria y CPU.

Por otro lado, el modelo reactivo se basa en E/S sin bloqueo y en un modelo de ejecución diferente. La E/S sin bloqueo proporciona una forma eficaz de tratar las E/S simultáneas. Una cantidad mínima de subprocesos denominados subprocesos de E/S puede manejar muchas E/S simultáneas. Con un modelo de este tipo, el procesamiento de solicitudes no se delega a un subproceso de trabajo, sino que utiliza estos subprocesos de E/S directamente. Ahorra memoria y CPU, ya que no es necesario crear subprocesos de trabajo para manejar las solicitudes. También mejora la simultaneidad ya que elimina la restricción sobre el número de subprocesos. Finalmente, también mejora el tiempo de respuesta ya que reduce el número de cambios de hilo.

Entonces, con el modelo de ejecución reactiva, las solicitudes se procesan mediante subprocesos de E/S. Pero eso no es todo. Un subproceso de E/S puede manejar varias solicitudes simultáneas. ¿Cómo? Aquí está el truco y una de las diferencias más significativas entre reactivo e imperativo.

Cuando el procesamiento de una solicitud requiere interactuar con un servicio remoto, como una API HTTP o una base de datos, no bloquea la ejecución mientras espera la respuesta. En cambio, programa la operación de E/S y adjunta una continuación, es decir, el código restante de procesamiento de la solicitud. Esta continuación se puede pasar como una devolución de llamada (una función invocada con el resultado de E/S) o utilizar construcciones más avanzadas como programación reactiva o co-rutinas. 

Independientemente de cómo se exprese la programación no bloqueante, lo esencial es la liberación del subproceso de E/S y, en consecuencia, el hecho de que este subproceso pueda utilizarse para procesar otra solicitud. Cuando se completa la E/S, el subproceso de E/S continúa el procesamiento de la solicitud pendiente.

Entonces, a diferencia del modelo imperativo, donde la E/S bloquea la ejecución, los conmutadores reactivos a un diseño basado en la continuación, donde se liberan los subprocesos de E/S y se invoca la continuación cuando se completan las E/S. Como resultado, el subproceso de E/S puede manejar múltiples solicitudes concurrentes, mejorando la concurrencia general de la aplicación.

Pero hay una trampa. Necesitamos una forma de escribir código continuation-passing o no bloqueante. Hay muchas maneras de hacer esto. Por ejemplo en Quarkus propone:

• Mutiny: una biblioteca de programación reactiva intuitiva e impulsada por eventos
• Co-rutinas de Kotlin: una forma de escribir código asincrónico de manera secuencial

Dejo link: https://quarkus.io/guides/getting-started-reactive

Como sería una base de datos reactiva??

Antes que nada voy a aclarara que este post tiene como objetivo solo hacer un ejercicio mental, no tiene como objetivo proponer un estándar o sugerir un cambio. Solo vamos a utilizar la imaginación y disfrutar…

Otra cosa, se que existe RDBC que básicamente ofrece un conjunto de funcionalidades reactivas para utilizar base de datos relacionales y esta bien, pero pero tiene sabor a poco. Dado que este conector tiene como objetivo simular un origen de datos reactivo que en realidad no es tan así …

En fin, una base de datos reactiva debería cumplir el manifiesto reactivo :

• Responsivos
• Resilientes
• Elásticos
• Orientados a Mensajes

No quiero detallar como implementar estos puntos, existen muchas bases de datos nosql que intentan implementarlo y a su modo lo hacen muy bien.

Dado los siguientes puntos y supongamos que los cumpla. Una base reactiva debería retornar siempre un Observer (si utilizamos RxJava, o puede ser un Flux o Mono si utilizamos spring) , por ahora no estoy agregando nada de conocimento esto es igual que R2DBC, vamos a soñar …

Lo que podríamos hacer es trasmitir datos por ejemplo : 

Preparestatement ps = con.createPreparestatement(“select * from Persona”); 

ps.subscribe( rs => System.out.println(rs.getString(“name”); ); 

  

Cada vez que se inserte un registro nuevo se imprimirá el nombre.

Le podríamos poner un delay, por ejemplo :

Preparestatement ps = con.createPreparestatement(“select * from Persona”); 

ps.delay(100);

ps.subscribe( rs => System.out.println(rs.getString(“name”); );

Si tuviéramos que hacer esto, de forma tradicional, deberíamos, revisar la base de datos cada tanto tiempo para ver si insertaron un dato nuevo. En cambio, si hubiera una base reactiva, la base nos retornará  cada 100 nanosegundos el nombre de una persona y esto no debería ser bloqueante, es decir si mientras imprimimos alguien cambia algo, no debería influir en nuestro proceso. Y si llega al final, cada vez que se inserte debería notificarnos. 

Otra cosa muy buena, sería poder suscribir a un cambio (esto desplazaría a los triggers). Deberíamos poder suscribir al cambio de un registro o de todos los registros de una tabla o la estructura de una tabla o etc. Por ejemplo vamos a suscribir un proceso que se llame cuanto eliminen un registro : 

Preparestatement ps = con.createPreparestatement(“select * from Persona p where p.id = 33”); 

ps.subscribe(“delete”, rs => System.out.println(rs.getString(“name”); );

ponele, me estoy suscribiendo y cuando se borre esta fila, la base de datos debería, notificarnos. 

Como se deben haber dado cuenta una base de datos reactiva no solo es un repositorio donde consulto datos, sino que es un sistema el cual conversa con nuestra aplicación y nos notifica de los cambios. Esto favorecería increíblemente las comunicaciones, supongamos que tenemos 2 sistemas que escriben en una base de datos, según lo que venimos razonando podríamos suscribirnos a la inserción de nuevos datos. Ect...

En fin, no conozco nada parecido, alguien conoce algo similar??

Dejo link: 

https://rdbc.io/

https://r2dbc.io/

https://www.reactivemanifesto.org/es

ReactiveX parte 2

Continuando con el post anterior...

En el modelo ReactiveX, Observable le permite tratar flujos de eventos asincrónicos con el mismo tipo de operaciones simples y componibles que se utilizan para colecciones. Lo libera de redes enredadas de devoluciones de llamada (como los callbacks) y, por lo tanto, hace un código sea más legible y menos propenso a errores.

Las técnicas como Java Futures son fáciles de usar para un solo nivel de ejecución asincrónica, pero comienzan a agregar una complejidad no trivial cuando están anidadas.

Es difícil usar Futures para componer de manera óptima flujos de ejecución asincrónicos condicionales (o imposible, ya que las latencias de cada solicitud varían en tiempo de ejecución). Esto se puede hacer, por supuesto, pero rápidamente se vuelve complicado (y por lo tanto propenso a errores) o se bloquea prematuramente en Future.get (), lo que elimina el beneficio de la ejecución asincrónica.

Los Observables de ReactiveX, están destinados a componer flujos y secuencias de datos asincrónicos.

ReactiveX Observables admite no solo la emisión de valores escalares únicos (como lo hacen los futuros), sino también secuencias de valores o incluso flujos infinitos. Observable es una única abstracción que puede usarse para cualquiera de estos casos de uso. Un Observable tiene toda la flexibilidad y elegancia asociadas a el Iterable:

ReactiveX no está sesgado hacia alguna fuente particular de concurrencia o asincronía. Los observables se pueden implementar utilizando grupos de subprocesos, bucles de eventos, I/O sin bloqueo, actores (como de Akka) o cualquier implementación que se adapte a sus necesidades, su estilo o su experiencia. El código del cliente trata todas sus interacciones con Observables como asíncronas, ya sea que su implementación subyacente sea bloqueante o no.

ReactiveX proporciona una colección de operadores con los que puede filtrar, seleccionar, transformar, combinar y componer Observables. Esto permite una ejecución y composición eficientes.

Se Puede pensar en la clase Observable como un equivalente a Iterable. Con un Iterable, el consumidor extrae valores del productor y los bloques de subprocesos hasta que lleguen esos valores. Por el contrario, con un Observable, el productor introduce valores al consumidor cuando los valores están disponibles. Este enfoque es más flexible, porque los valores pueden llegar sincrónicamente o asincrónicamente.

El tipo Observable agrega dos semánticas faltantes al patrón Observador de la Banda de los Cuatro, para que coincida con las que están disponibles en el tipo Iterable:

• la capacidad del productor de indicarle al consumidor que no hay más datos disponibles (un bucle foreach en un Iterable se completa y regresa normalmente en tal caso; un Observable llama al método onCompleted de su observador)
• la capacidad del productor de indicarle al consumidor que se ha producido un error (un Iterable genera una excepción si se produce un error durante la iteración; un Observable llama al método onError de su observador)

Con estas adiciones, ReactiveX armoniza los tipos Iterable y Observable. La única diferencia entre ellos es la dirección en la que fluyen los datos. Esto es muy importante porque ahora cualquier operación que pueda realizar en un Iterable, también puede realizarla en un Observable.

Dejo link : http://reactivex.io

ReactiveX

Hace tiempo que existe este framework y es muy raro que no haya escrito de él antes, tal vez porque se utiliza muchísimo en Angular y no soy un experto en esta tecnología. 

Pero bueno, todo llega y vamos a hablar de ReactiveX que en su pagina web dice que ReactiveX es una combinación de buenas ideas: el patrón Observer, el patrón Iterator y la programación funcional.

Patrones y programación funcional, temas que tocamos todo el tiempo en el blog, que raro que no lo vi antes o si lo vi, le reste importancia. Me paso lo mismo con typescript. 

ReactiveX sigue el patrón Reactor. Que si vamos a la wikipedia podemos leer : 

"El patrón de diseño reactor es un patrón de programación concurrente para manejar los pedidos de servicio entregados de forma concurrente a un manejador de servicio desde una o más entradas. El manejador de servicio demultiplexa los pedidos entrantes y los entrega de forma sincrónica a los manejadores de pedidos asociados."

y

"Este patrón separa completamente al código específico de la aplicación de la implementación del reactor, permitiendo que los componentes de la aplicación puedan ser divididos en módulos reutilizables. Además, la llamada síncrona de los manejadores de pedidos permite concurrencia, de grano grueso, sin agregar la complejidad de múltiples hilos en el sistema."

A la vez ReactiveX utiliza los conceptos de programación reactiva. Pero que sería la programación reactiva?? 

Esto es más fácil explicarlo con un ejemplo, supongamos que tenemos que hacer una aplicación que tiene que calcular una serie de montos, el monto 1 se ingresa, el monto 2 es un porcentaje del monto 1, el monto 3 se ingresa, el monto 4, es la suma del montos. Si programaste aplicaciones seguro que te encontraste con un problema similar. Si ingresa el monto 1 debemos refrescar todos los montos calculados y si ingresan el monto 3 solo el 4. Podemos atacamos este problema poniendo un botón que refresque los valores, el tema es que más allá que es poco práctico, no esta bueno que si modificamos el monto 3, se refresque todo (ya que es innecesario)

Excel o una hoja de calculo, la hace más fácil, cuando modificamos la celda, refresca automáticamente todos los valores que la usan esta celda para su calculo, este es el concepto de reactividad. Y es mucho más eficiente. 

Pero ahora volvamos a reactiveX que básicamente implementa lo que acabo de explicar, con diferentes conceptos : observador y suscripción. 

Por lo tanto ReactiveX es una librería para componer programas asincrónicos y basados en eventos mediante el uso de secuencias observables.

Extiende el patrón de observador para admitir secuencias de datos y/o eventos y agrega operadores que le permiten componer secuencias juntas de manera declarativa, mientras abstrae preocupaciones sobre cosas como subprocesos de bajo nivel, sincronización, seguridad de subprocesos, estructuras de datos concurrentes y no bloqueo de I/O.

ReactiveX es multilenguaje y multiplataforma, talvez la versión más utilizada es la de javascript, ya que viene con Angular y otros frameworks webs. Pero se puede utilizar en diferentes plataformas y lenguajes. 

Les dejo la lista: 

Languages

• Java: RxJava
• JavaScript: RxJS
• C#: Rx.NET
• C#(Unity): UniRx
• Scala: RxScala
• Clojure: RxClojure
• C++: RxCpp
• Lua: RxLua
• Ruby: Rx.rb
• Python: RxPY
• Go: RxGo
• Groovy: RxGroovy
• JRuby: RxJRuby
• Kotlin: RxKotlin
• Swift: RxSwift
• PHP: RxPHP
• Elixir: reaxive
• Dart: RxDart

ReactiveX for platforms and frameworks

• RxNetty
• RxAndroid
• RxCocoa

Y como me quedo muy largo el post, voy a seguir en otros...

Dejo link: 

http://reactivex.io/

https://github.com/ReactiveX

Programación reactiva + bases de datos relacionales = R2DBC

Continuamos con Programación reactiva y bases de datos relacionales.

Al carecer de una API estándar y la falta de disponibilidad de controladores, un equipo de Pivotal comenzó a investigar la idea de una API relacional reactiva que sería ideal para fines de programación reactiva. Y en ese momento nació, R2DBC que significa Conectividad de base de datos relacional reactiva.

Entre las características de R2DBC podemos nombrar: 

R2DBC se basa en la especificación de Reactive Streams, que proporciona una API sin bloqueo totalmente reactiva.

Trabaja con bases de datos relacionales. A diferencia de la naturaleza bloqueante de JDBC, R2DBC le permite trabajar con bases de datos SQL utilizando una API reactiva.

Admite soluciones escalables. Con Reactive Streams, R2DBC le permite pasar del modelo clásico de "un subproceso por conexión" a un enfoque más potente y escalable.

Proporciona una especificación abierta. R2DBC es una especificación abierta y establece una interfaz de proveedor de servicios (SPI) para que los proveedores de controladores implementen y los clientes los consuman.

Actualmente existen las siguientes implementaciones : 

• cloud-spanner-r2dbc: controlador para Google Cloud Spanner
• jasync-sql: contenedor R2DBC para Java & Kotlin Async Database Driver para MySQL y PostgreSQL escrito en Kotlin.
• r2dbc-h2: controlador nativo implementado para H2 como base de datos de prueba.
• r2dbc-mariadb: controlador nativo implementado para MariaDB.
• r2dbc-mssql: controlador nativo implementado para Microsoft SQL Server.
• r2dbc-mysql: controlador nativo implementado para MySQL.
• r2dbc-postgres: controlador nativo implementado para PostgreSQL.

Los estándares existentes, basados ​​en el bloqueo de I/O, cortan la programación reactiva de los usuarios de bases de datos relacionales. R2DBC especifica una nueva API para permitir código reactivo que funciona de manera eficiente con bases de datos relacionales.

R2DBC es una especificación diseñada desde cero para la programación reactiva con bases de datos SQL. Define un SPI sin bloqueo para implementadores de controladores de bases de datos y autores de bibliotecas de clientes. Los controladores R2DBC implementan completamente el protocolo de conexión de la base de datos sobre una capa de I/O sin bloqueo.

R2DBC está pensado principalmente como un SPI del controlador para ser consumido por las bibliotecas del cliente y no para ser utilizado directamente en el código de la aplicación.

R2DBC admite aplicaciones nativas en la nube que utilizan bases de datos relacionales como PostgreSQL, MySQL y otras. Los desarrolladores de aplicaciones son libres de elegir la base de datos adecuada para el trabajo sin estar limitados por las API.

Spring Data R2DBC, parte de la familia Spring Data, facilita la implementación de repositorios basados en R2DBC. Spring Data R2DBC aplica abstracciones de la familia de Spring y soporte de repositorio para R2DBC. Facilita la creación de aplicaciones basadas en Spring que utilizan tecnologías de acceso a datos relacionales en una stack de aplicaciones reactivas.

Spring Data R2DBC pretende ser conceptualmente fácil. Para lograr esto, NO ofrece almacenamiento en caché, carga diferida, escritura detrás o muchas otras características de los marcos ORM. Esto hace que Spring Data R2DBC sea un mapeador de objetos simple, limitado y con opiniones.

Spring Data R2DBC permite un enfoque funcional para interactuar con su base de datos proporcionando DatabaseClient como el punto de entrada para las aplicaciones.

Veamos un ejemplo con postgres : 

PostgresqlConnectionFactory connectionFactory = new PostgresqlConnectionFactory(PostgresqlConnectionConfiguration.builder()

.host(…)

.database(…)

.username(…)

.password(…).build());

DatabaseClient client = DatabaseClient.create(connectionFactory);

Mono<Integer> affectedRows = client.execute()

        .sql("UPDATE person SET name = 'Joe'")

        .fetch().rowsUpdated();

Flux<Person> all = client.execute()

        .sql("SELECT id, name FROM person")

        .as(Person.class)

        .fetch().all();

Otro enfoque para atacar el bloqueo de JDBC es Fibers. Fibers como una abstracción ligera que convertirá las API de bloqueo en no bloqueantes. Esto es posible mediante el cambio de pila tan pronto como una invocación ... Pero eso es otra Historia y va ha ser contada en otro post ... 

Dejo links: 

https://github.com/r2dbc

https://r2dbc.io

https://spring.io/projects/spring-data-r2dbc

Programación reactiva y bases de datos relacionales

Hay muchas respuestas sobre qué es la programación reactiva y cómo se compara con los sistemas reactivos. La Programación Reactiva se puede ver como un modelo de programación que facilita la escalabilidad y la estabilidad mediante la creación de tuberías funcionales sin bloqueo controladas por eventos que reaccionan a la disponibilidad y procesabilidad de los recursos. La ejecución diferida, la concurrencia y la asincronía son solo una consecuencia del modelo de programación subyacente.

Los beneficios completos de la programación reactiva entran en vigencia solo si toda el stack de tecnologías es reactiva y si todos los componentes participantes (código de aplicación, contenedor de tiempo de ejecución, integraciones) respetan la ejecución diferida, las API sin bloqueo y la naturaleza de flujo de flujo de datos, básicamente siguiendo los supuestos subyacentes .

Si bien es posible llevar componentes no reactivos a una aplicación que está escrita en un estilo funcional-reactivo, el resultado neto es que los beneficios reales esperados, disminuyen. En el peor de los casos, hay poca o ninguna diferencia en el comportamiento del tiempo de ejecución. Sin embargo, la programación reactiva ayuda a mejorar la legibilidad del código.

Si observamos el ecosistema reactivo, descubriremos varios frameworks, bibliotecas e integraciones. Cada uno de ellos tiene sus puntos fuertes específicos. Muchas áreas funcionales están bien cubiertas, ya sea con un enfoque genérico o dentro del contexto de un framework reactivo particular. 

Java utiliza JDBC como tecnología principal para integrarse con bases de datos relacionales. JDBC es de naturaleza bloqueante: no hay nada sensato que se pueda hacer para mitigar la naturaleza bloqueante de JDBC. La primera idea de cómo hacer que las llamadas no se bloqueen es descargar las llamadas JDBC a un ejecutor (generalmente grupo de subprocesos). Si bien este enfoque funciona, viene con varios inconvenientes que descuidan los beneficios de un modelo de programación reactiva.

Los grupos de subprocesos requieren, no es de extrañar, subprocesos para ejecutarse. Los tiempos de ejecución reactivos suelen utilizar un número limitado de subprocesos que coinciden con el número de núcleos de CPU. Los hilos adicionales introducen gastos generales y reducen el efecto de limitación de hilos. Además, las llamadas JDBC generalmente se acumulan en una cola, y una vez que los hilos están saturados de solicitudes, el grupo se bloqueará nuevamente. Entonces, JDBC ahora no es una opción.

Hay un par de controladores independientes, como el reactive-pg-client. Estos controladores vienen con una API específica del proveedor y no son realmente adecuados para una adopción más amplia. 

Como no hay una API estándar y la falta de disponibilidad de controladores, un equipo de Pivotal comenzó a investigar la idea de una API relacional reactiva que sería ideal para fines de programación reactiva. Se les ocurrió R2DBC que significa Conectividad de base de datos relacional reactiva. Pero eso es otra Historia y va ha ser contada en otro post ... 

Dejo link: 

https://r2dbc.io/

https://github.com/eclipse-vertx/vertx-sql-client

3 eBooks sobre Reactive Microservices con Java

Me llego un mail de lightbend que están regalando 3 ebooks sobre Reactive en Java. Nada, solo hay que bajarlo.

Dejo link: http://inform.lightbend.com/Xlk0yJC0l000Xz02f02aNN0

Reactive Programming with Java

Quiero recomendarles este libro que ha publicado de forma gratuita infoQ.

Sin más dejo el link:

https://www.infoq.com/minibooks/emag-reactive-programming-java

Reactivity con Elm

Escribir aplicaciones reactivas para un programador javascript puede ser complejo, pero que tal si utilizamos un lenguaje que sea construido bajo estos principios? Este es ELM.

Elm fue creada no para responder a la pregunta "¿cómo puede JavaScript ser mejor?", Sino para responder a la pregunta "¿como puedo mejorar experiencia general para desarrolladores de interfaces de usuario Web?"

En la búsqueda de este objetivo, Elm abraza varias ideas que JavaScript no:

Reactive programing como el único sistema para responder a la interacción
Una forma consistente de gestionar los efectos
Un compilador útil que elimina muchos errores antes
El resultado es un lenguaje que compila JavaScript, pero tiene una reputación radicalmente diferente. Es común escuchar que el código de Elm de producción de alguien nunca ha lanzado una excepción de tiempo de ejecución, ni siquiera el temido indefinido no es una función.

Dejo link: https://www.infoq.com/articles/language-reactivity-with-elm?utm_source=linkedin&utm_campaign=calendar&utm_medium=link
http://elm-lang.org/
https://evancz.gitbooks.io/an-introduction-to-elm/content/architecture/index.html

Vert.x construyendo aplicaciones reactivas en jvm

Vert.x es un framework java para la construcción de aplicaciones asíncronas, basadas en eventos y no bloqueantes. Esto significa que con Vert.x podemos obtener aplicaciones concurrentes con un mínimo de complejidad, y un hardware mínimo.

Para ser más explicativo se compara Vert.x con Node.js, por lo que se puede inferir que Vert.x fue pensado para ser concurrente, asíncrono y barato.

Otra ventaja de este framework es que es poliglota es decir podemos escribir aplicaciones en varios lenguajes, veamos un ejemplo:

Java:

public class Server extends AbstractVerticle {
  public void start() {
    vertx.createHttpServer().requestHandler(req -> {
      req.response()
        .putHeader("content-type", "text/plain")
        .end("Hello from Vert.x!");
    }).listen(8080);
  }
}

Ruby:

vertx.create_http_server().request_handler() { |req|
  req.response()
    .put_header("content-type", "text/plain")
    .end("Hello from Vert.x!")
}.listen(8080)

Javascript:

vertx.createHttpServer()
  .requestHandler(function (req) {
    req.response()
      .putHeader("content-type", "text/plain")
      .end("Hello from Vert.x!");
}).listen(8080);

Groovy:

vertx.createHttpServer().requestHandler({ req ->
  req.response()
    .putHeader("Content-Type", "text/html")
    .end("Hello from Vert.x!")
}).listen(8080)

Como vemos Vert.x proporciona una API en varios lenguajes que da soporte.

Dejo link:
http://vertx.io/
http://vertx.io/docs/

Elm

Elm es un lenguaje funcional basado en Functional Reactive Programming.

Con características bien funcionales como inmutabilidad, funciones de primer orden y otras ventajas de los lenguajes funcionales, se hace fácil escribir código fácil de mantener y corto.

Elm compila a HTML, CSS y JavaScript. No es fácil crear aplicaciones web pero con este framework podemos generar el maquetado de forma automática.



Dejo link: http://elm-lang.org/

Principles of Reactive Programming

Ayer comenzó, la segunda edición del curso "Principles of Reactive Programming"


A estudiar!!

Dejo link:
https://class.coursera.org/reactive-002

Reactive Programming

Hola a todos quería compartir con ustedes este video sobre Reactive:

Nuevamente se dicta Principles of Reactive Programming en Coursera.

Que bueno es dar buenas noticias!

Se dicta nuevamente el curso Principles of Reactive Programming en Coursera el 13 de Abril de 2015. Este curso es muy bueno y lo dictan unos monstruos...

Por favor por amor a su carrera o por curiosidad, tomen este curso, super recomendado.

Dejo link:
https://www.coursera.org/course/reactive

Que son las aplicaciones Reactivas?

Una pequeña infografia que describe muy bien los principios de las aplicaciones reactivas:

Ojo! se que no se ve, solo tienen que hacer click en la imagen.