RSocket: adapta tus aplicaciones a la era reactiva.

Hace ya seis años que se publicó la versión 2.0 del Manifiesto Reactivo. Como bien se comenta en este documento, no hace tanto tiempo, una aplicación requería de decenas de servidores, varios segundos de respuesta, muchas horas de mantenimiento y bastantes gigabytes en datos. Hoy en día, la forma en que se construyen y despliegan los sistemas software está evolucionando a pasos agigantados, puesto que las necesidades han ido cambiando y ahora se requiere de aplicaciones que puedan ser desplegadas en cualquier lugar, desde dispositivos móviles hasta clusters en la nube, con tiempos de respuesta medidos en milisegundos, que operan 24x7 y manejan un millón de veces más en datos.

Llegados a este punto, dicho manifiesto introduce los sistemas reactivos, sistemas más flexibles, con bajo acoplamiento y altamente escalables y que se construyen adhiriéndose a 4 principios: capacidad de respuesta (responsive), resiliencia (resilient), elasticidad (elastic) y dirigidos por mensajes (message driven).

En este post vamos a hablar de RSocket, un nuevo protocolo de comunicaciones que se adapta como un guante a estos principios “reactivos”. Detallaremos sus características, las motivaciones por las que aparece y detalles de su implementación mediante un pequeño ejemplo práctico en el que nos mancharemos un poco las manos.

¿Qué es RSocket?

RSocket es un protocolo de comunicación reactivo, binario, asíncrono, uno a uno y sin estado. Está pensado, sobre todo, para las necesidades de la construcción de aplicaciones en la nube con microservicios. Además, es compatible con múltiples lenguajes de programación, como por ejemplo, Java, JavaScript o Python.

La comunicación en el protocolo Rsocket se descompone en frames. Cada frame consta de una cabecera (header), la cual contiene el identificador del stream, el tipo de frame y otra información específica del tipo de frame. Además del header, el frame contiene los metadatos y el payload. Existen varios tipos de frames que representan diferentes propósitos. Todos los tipos pueden consultarse en la documentación oficial.

Estos frames se envían como streams de bytes, permitiendo que este protocolo sea más eficiente que los típicos protocolos basados en texto. El protocolo no impone ninguna restricción en cuanto a la forma de serializar/deserializar la información, simplemente considera el frame como un contenedor de bits que puede ser convertido a cualquier cosa, haciendo posible el uso de JSON o cualquier otra solución como Protobuf.

En último lugar, el protocolo de Rsocket crea una serie de canales sobre la conexión física única (multiplexación). Esto permite enviar múltiples peticiones (requests) sobre la misma conexión, sin ser necesario abrir múltiples conexiones
con el consiguiente consumo de recursos.

Motivación

A finales del 2014 empezó a incrementarse el número de búsquedas sobre el término ‘microservicios’ y, en paralelo, se desarrollaron los primeros frameworks enfocados a la construcción de este tipo de arquitecturas. Hoy en día es un término del que no se para de hablar continuamente. Especialmente, debemos destacar el papel de la compañía estadounidense Netflix porque fue una de las empresas pioneras en migrar con éxito de una arquitectura monolítica a una de microservicios basada en la nube.

Actualmente, Netflix cuenta con más de 1000 microservicios y cada uno de ellos se encarga de una parte del modelo de negocio. RSocket tiene una estrecha relación con esta compañía: sus creadores tenían la intención de desarrollar una forma estándar para que las aplicaciones se comuniquen entre sí a través de la red y qué mejor lugar para hacerlo que ayudando a Netflix.

Os mostramos algunas de las motivaciones que sintieron los desarrolladores de este nuevo protocolo para su creación:

• Resolver el problema del modelo de interacción request - response del protocolo HTTP: el cliente envía una petición (request) y el servidor retorna una respuesta (response). Se trata de un modelo bastante rígido y resulta complicado implementar modelos de interacción donde, por ejemplo, un microservicio envíe una petición a otro componente sin tener en cuenta la respuesta de este o, bien, pueda recibir un stream de datos a lo largo del tiempo, según se encuentren disponibles.
• Resolver el problema del rendimiento asociado a la utilización del protocolo HTTP. Este problema viene derivado del modelo request - response, de HTTP 1.x, en el que cada petición única tiene su propia pila de memoria. HTTP 1.x es un protocolo basado en texto, con lo que el tamaño de los datos que deben ser transferidos es significativamente mayor que en el caso de los protocolos binarios (RSocket).
• Otro de los puntos fuertes que aporta RSocket es la reanudación de sesión (resumability), permitiendo un simple handshake para reanudar una sesión cliente/servidor a través de una nueva conexión de transporte. Esto es significativo en flujos de larga duración, en particular los que sirven suscripciones de clientes móviles en los que las desconexiones de la red pueden tener un impacto significativo en el costo y en el rendimiento si se deben restablecer todas las suscripciones.
• Otra característica que aporta RSocket es la implementación de un mecanismo de resistencia (backpressure mechanism). En una comunicación entre dos componentes, el receptor de los mensajes puede especificar cuántos datos desea consumir y no obtendrá más que eso hasta que notifique al emisor que está listo para procesar más.

En la especificación original de RSocket, se puede encontrar una información mucho más detallada sobre todos los aspectos que hemos ido comentado anteriormente.

Modelos de interacción

RSocket está basado en cuatro modelos que permiten una interacción simétrica a través de una sola comunicación:

• Fire and forget: es muy similar al modelo request-response, pero en este caso el cliente al enviar los datos al servidor no espera ninguna respuesta.
• Request-response: sería el modelo que más se ajusta al comportamiento del protocolo HTTP. El cliente espera la respuesta del servidor, pero lo hace de forma asíncrona (es decir, no bloqueante).
• Request-stream: es uno de los modelos más interesantes de utilizar junto al channel en aplicaciones cloud. Una única petición por parte del cliente recibe un flujo de respuestas del servidor. De esta manera, el cliente irá procesando los datos que le lleguen de forma automática según vayan estando disponibles.
• Channel: se trata del modelo más completo, ya que permite una comunicación bidireccional entre cliente y servidor a través de una única conexión. Es muy útil cuando el cliente necesita actualizar la suscripción para cambiar los criterios asociados a ella.

Hands-on

En este apartado vamos a ilustrar el modelo de interacción request - stream, por medio de un pequeño ejemplo práctico donde simularemos un famoso juego de lotería. Podéis encontrar el código de la demostración en el siguiente repositorio.

Para ejecutar la aplicación de manera local, no es necesario disponer de más software que, Docker y Docker Compose. Dentro del repositorio, existe un fichero docker-compose.yml, con el que levantar el sistema mediante el comando:docker-compose up -d (directamente desde el directorio donde has clonado el repositorio).

La aplicación se compone de dos servicios. Por un lado, el componente cliente (una aplicación desarrollada en Angular), que es el encargado de solicitar los números del sorteo al servidor y mostrarlos en la interfaz web. Por otro lado, tenemos el componente servidor (una aplicación desarrollada con Spring Boot) que se encarga de comunicarse con el cliente para la generación y transmisión de los números ganadores de la lotería.

En primer lugar, comenzaremos mostrando los aspectos más relevantes de la parte servidor. Se trata de una aplicación de Spring Boot, y para empezar a trabajar requerimos de las siguientes dependencias en el proyecto:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

El starter ‘spring-boot-starter-rsocket’ integra RSocket en Spring Boot y configura automáticamente cierta infraestructura de RSocket en tiempo de ejecución (autoconfiguración de servidor RSocket, autoconfiguración de cliente RSocket, autoconfiguración de estrategias RSocket, etc.). El starter ‘spring-boot-starter-webflux’ es la elección que nos permitirá construir la aplicación web de forma reactiva.

En conjunción con estas dependencias debemos configurar una serie de propiedades en el fichero application.yml (o application.properties) para que Spring Boot pueda realizar toda su magia.

Con las propiedades y valores definidos, Spring Boot registra de manera automática una serie de beans, como se muestra en las siguientes capturas de pantalla. Se definen las propiedades:

• spring.rsocket.server.port. Propiedad que especifica el puerto en el que escuchará el servidor de RSocket. Se usará el puerto 8081.
• spring.rsocket.server.mapping-path. Esta propiedad especifica el path bajo el que RSocket se encargará de manejar las peticiones. Funciona en conjunción con el protocolo ‘WebSocket’. En este ejemplo tendrá el valor de lottery-game.
• spring.rsocket.server.transport. Propiedad que especifica el protocolo de transporte RSocket. Para este ejemplo, utilizaremos WEBSOCKET. También se puede utilizar el valor TCP.

Sin entrar mucho en detalle, en esta imagen se puede ver cómo se registran una serie de beans pertenecientes a una clase de configuración (EmbeddedServerAutoConfiguration) en el momento que se encuentra definida la propiedad ‘spring.rsocket.server.port’. Uno de esos beans, rSocketServerFactory, se encarga de la creación de un objeto de tipo RSocketServerFactory que puede ser utilizado para crear un servidor de RSocket, respaldado por Netty.

Otro de los beans que se registran es rSocketServerBootstrap, creando un objeto de tipo RSocketServerBootstrap que permite el arranque de un servidor de RSocket y su iniciación con el contexto de aplicación de Spring.

Ya tenemos realizada la configuración del servidor. Como hemos visto previamente, Spring Boot se encargará de la mayor parte. Ahora, el siguiente punto será crear una clase Controlador (LotteryController) que esencialmente declarará endpoints del servicio, en este caso, endpoints RSocket. Dentro del controlador, declararemos el método que se encarga de realizar la operación de generar los números ganadores del sorteo.

El método generateLotteryNumbers recibirá por parámetro un objeto de tipo RequestData que representa el Payload de la petición y que indicará la cantidad de números a generar y si se tratan de números especiales o no. Por otro lado, el método será decorado con @MessageMapping(“generate.numbers”). Esta anotación indica que cualquier mensaje entrante cuyos metadatos contengan la routing-key ‘generate.numbers’ será procesado por este método. Este ejemplo es bastante sencillo, ya que solo disponemos de esta operación, pero si tuviéramos diferentes operaciones en el sistema esta anotación sirve para discernir entre diferentes métodos handler.

Con este último punto ya se han comentado los aspectos más relevantes de la parte servidor. En el repositorio. se puede consultar la pieza restante, el LotteryService, cuya responsabilidad es sortear los números en base a un rango. Este rango se define mediante properties y pueden ser sobreescritas en el arranque de la aplicación mediante variables de entorno.

En cuanto a lo que se refiere a la parte cliente de la aplicación se trata de un proyecto de Angular 11 creado con el propio angular-cli. Resaltar que este mismo ejemplo se podría haber mostrado en cualquier otro framework de JavaScript (React, Vue.js…) o incluso en JavaScript Vanilla.

De igual manera que se ha hecho en la parte servidor, se van a detallar los aspectos más significativos de la parte cliente referente a RSocket.

Para comenzar deberemos añadir las dependencias que se requieren en el proyecto. Después habrá que ejecutar el correspondiente comando npm i para que se instalen en el proyecto.

"dependencies": {
     …
    "rsocket-core": "0.0.19",
    "rsocket-flowable": "0.0.14",
    "rsocket-websocket-client": "0.0.19",
     …
  },
  "devDependencies": {
    …
    "@types/rsocket-core": "0.0.5",
    "@types/rsocket-websocket-client": "0.0.3",
   …
  }

Para gestionar la conexión con el servidor se ha creado una clase Client.ts que contiene los atributos y métodos necesarios para el ejemplo.

A continuación vamos a detallar los métodos más significativos de esta clase:

• constructor(): Para esta demostración el único atributo parametrizable que se ha usado en la creación del objeto es la URL en la cual se encuentra el lado servidor. Se han definido por defecto las propiedades de configuración de la conexión como son keepAlive, lifetime, dataMiymeType y metadataMimeType.

  constructor(address: string) {
    this.client = new RSocketClient({
      serializers: {
      data: JsonSerializer,
        metadata: IdentitySerializer,
      },
      setup: {
        keepAlive: 10000,
        lifetime: 20000,
        dataMimeType: 'application/json',
        metadataMimeType: 'message/x.rsocket.routing.v0',
      },
      transport: new RSocketWebSocketClient({ url: address }),
    });
  }

• connect(): Devuelve una promesa si la conexión al WebSocket ha ido correctamente.

 connect(): Promise<any> {
    return new Promise((resolve, reject) => {
      this.client.connect().subscribe({
        onComplete: (rs: object) => {
          this.rsocket = rs;
          this.rsocket.connectionStatus().subscribe((status: object) => {
            console.info(status);
          });

          resolve(this.rsocket);
        },
        onError: (error: Error) => {
          reject(error);
        },
        onSubscribe: (cancel: CancelCallback) => {
          this.cancel = cancel;
        },
      });
    });
  }

• requestStream(): Devuelve un flujo de respuestas en base a una única petición. Es necesario especificar tanto el data como el metadata. Para nuestro ejemplo, como veremos más adelante, será un objeto formado por la cantidad de números que queremos recibir y si son de tipo especial o no. Respecto al metadata se trata de la routing-key ‘generate.numbers’ que indica al servidor el método por el que deberá ser procesado.

 requestStream(clientData: ClientData): Flowable<any> {
    return this.socket.requestStream({
      data: clientData,
      metadata:
        String.fromCharCode('generate.numbers'.length) + 'generate.numbers',
    });
  }

• disconnect(): Método para cerrar la conexión que había establecida con el WebSocket.

  disconnect(): void {
    this.client.close();
  }

En cuanto al componente principal, el cual hemos llamado LotteryComponent, va a ser el encargado de gestionar el flujo y hacer uso del objeto Client.

Las funciones más representativas en cuanto al flujo e interacción con RSocket de esta componente son:

• connectClient(): Función para crear la instancia del objeto Client y realizar la conexión al servidor. Si se ha realizado correctamente comienza el flujo de nuestra aplicación.

  connectClient(): void {
    if (!this.connected) {
      this.client = new Client(this.address);
      this.client.connect().then(() => {
        this.connected = true;
        this.getNumbers(5, false, '1');
      });
    } else {
      this.client.disconnect();
      this.connected = false;
    }
  }

• getNumbers(): Invoca al método requestStream() del objeto Client para hacer la petición al servidor e iniciar el flujo. En este caso, le envía la información de la cantidad total de números y de si se tratan de números especiales o no. Además, se establece la suscripción asíncrona a la respuesta, de tal manera que el servidor según va contestando con números, el cliente los va procesando, en este caso los muestra en pantalla con la función printResults().

 getNumbers(totalNumbers: number, isSpecial: boolean, idContent: string): void {
    if (!this.streamInProgress) {
      const requestedMsg = totalNumbers;
      let processedMsg = 0;
      const clientData = new ClientData(totalNumbers, isSpecial);

      this.client.requestStream(clientData).subscribe({
        onSubscribe: (sub: object) => {
          this.requestStreamSubscription = sub;
          this.requestStreamSubscription.request(requestedMsg);
          this.streamInProgress = true;
        },
        onError: (error: Error) => {
          console.error(error);
        },
        onNext: (clientDto: ClientDto) => {
          this.printResults(clientDto.data, idContent, isSpecial);
          processedMsg++;

          if (processedMsg >= requestedMsg) {
            this.requestStreamSubscription.request(requestedMsg);
            processedMsg = 0;
          }
        },
        onComplete: () => {
          this.streamInProgress = false;
          if (!isSpecial) {
            this.getNumbers(2, true, '2');
          } else {
            this.client.disconnect();
            this.connected = false;
            this.goToHome();
          }
        },
      });
    } else {
      this.requestStreamSubscription.cancel();
    }
  }

Por último, comentar que, igual que en la parte servidor, el resto del código se puede visualizar en el repositorio donde se podrá observar diferentes funciones, sobre todo en la parte referente a la visualización de los números y otras propias del ciclo de vida de una aplicación Angular.

Conclusión

En este post hemos querido mostraros este nuevo protocolo, una alternativa atractiva para trabajar junto con las arquitecturas de software emergentes y en diferentes tipos de entorno como puede ser una arquitectura de microservicios basada en la nube. Viene a resolver muchos de los puntos débiles que existen en el protocolo que se usa de manera tradicional para las comunicaciones (flexibilidad, rendimiento, etc.), HTTP 1/x.

En definitiva, RSocket permite una comunicación efectiva y escalable entre los componentes de la aplicación que es uno de los aspectos cruciales en sistemas distribuidos, afectando directamente a la latencia que experimentan los usuarios y a la cantidad de recursos necesarios para construir y hacer funcionar el sistema.