Spring WebFlux 4: Comunicación Avanzada, Pruebas y Producción

La serie Spring WebFlux nos ha llevado a través de un viaje fascinante por el mundo de la programación reactiva, desde sus fundamentos y el poder de Project Reactor hasta la construcción de arquitecturas altamente concurrentes y la gestión de la comunicación con servicios externos y bases de datos. En esta cuarta parte, profundizaremos en aspectos más avanzados y críticos para el desarrollo y despliegue de aplicaciones WebFlux robustas y eficientes. Exploraremos desde la comunicación en tiempo real con Server-Sent Events y WebSockets, hasta la crucial gestión de la contrapresión, el contexto reactivo, las estrategias de testing y, por supuesto, la seguridad y las buenas prácticas en producción.

---

1. Server-Sent Events (SSE): Flujos de Eventos Unidireccionales

Los Server-Sent Events (SSE) son una tecnología web que permite a un servidor enviar actualizaciones automáticamente a un cliente a través de una conexión HTTP persistente y unidireccional. A diferencia de los WebSockets, que son bidireccionales y más complejos, los SSE están diseñados específicamente para escenarios donde el cliente solo necesita recibir datos del servidor. Piensa en ellos como un flujo continuo de noticias, actualizaciones de cotizaciones bursátiles o notificaciones en tiempo real.

¿Cómo funcionan los SSE?

El cliente establece una conexión HTTP normal con el servidor. Sin embargo, en lugar de cerrar la conexión después de enviar la respuesta inicial, el servidor la mantiene abierta y envía datos de forma continua. Cada “evento” se envía como un bloque de texto formateado de una manera específica, seguido de un salto de línea. El navegador o cliente (usando la API EventSource de JavaScript) interpreta estos bloques como eventos individuales.

SSE con Spring WebFlux

En Spring WebFlux, implementar SSE es sorprendentemente sencillo gracias a la naturaleza reactiva de Flux. Dado que un Flux puede emitir 0 a N elementos de forma asíncrona, es la elección natural para representar un flujo de eventos.

Para enviar eventos, simplemente necesitas devolver un Flux desde tu controlador. Spring WebFlux se encargará automáticamente de configurar los encabezados HTTP (Content-Type: text/event-stream) y formatear los datos para que el cliente los reciba como SSE.

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.time.LocalDateTime;

@RestController
public class SseController {

    @GetMapping(value = "/eventos", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> getEvents() {
        return Flux.interval(Duration.ofSeconds(1)) // Emite un elemento cada segundo
                   .map(sequence -> "Evento #" + sequence + " a las " + LocalDateTime.now());
    }

    @GetMapping(value = "/data-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<MyData> streamMyData() {
        return Flux.interval(Duration.ofSeconds(2))
                   .map(sequence -> new MyData("Item " + sequence, Math.random() * 100))
                   .take(5); // Limita el número de elementos
    }
}

En este ejemplo:

• getEvents() envía una cadena de texto cada segundo.
• streamMyData() envía objetos MyData (que se serializarán a JSON automáticamente) cada dos segundos, limitando la emisión a 5 elementos.

Del lado del cliente (JavaScript):

const eventSource = new EventSource('/eventos');

eventSource.onmessage = function(event) {
    console.log("Mensaje recibido:", event.data);
};

eventSource.onerror = function(error) {
    console.error("Error en el flujo de eventos:", error);
    eventSource.close();
};

// Si el servidor envía eventos con un 'event' type específico:
// eventSource.addEventListener('nombreDeEvento', function(event) {
//     console.log("Evento con nombre específico:", event.data);
// });

Los SSE son ideales para dashboards en tiempo real, feeds de actividad o cualquier escenario donde se necesiten actualizaciones push del servidor sin la complejidad de una conexión bidireccional completa.

---

2. Backpressure: Gestionando el Flujo de Datos

El concepto de backpressure (contrapresión) es fundamental en la programación reactiva y, en particular, en Project Reactor y Spring WebFlux. Se refiere a la capacidad de un suscriptor (consumidor) de señalar a un publicador (productor) qué tan rápido o cuántos elementos puede procesar. En un flujo reactivo, si el productor es mucho más rápido que el consumidor, los datos se acumularán en el buffer del consumidor, lo que puede llevar a problemas de memoria o a la caída del sistema. La contrapresión resuelve esto permitiendo que el consumidor “tire” de los datos solo cuando está listo para manejarlos.

¿Por qué es crucial la contrapresión?

Imagina un río (el publicador) que fluye muy rápido hacia un balde (el suscriptor) que solo puede contener una pequeña cantidad de agua a la vez. Sin contrapresión, el balde se desbordaría rápidamente. Con contrapresión, el balde puede indicarle al río que disminuya el caudal o que le envíe agua solo cuando haya espacio.

En el contexto de Spring WebFlux, la contrapresión es vital para la estabilidad y eficiencia del sistema. Evita que un servicio backend sobrecargue a un cliente más lento (como un navegador o una API externa con límites de tasa) o que una base de datos reactiva inunde el servicio con resultados que no puede procesar a tiempo.

Implementación en Reactor

Project Reactor implementa la contrapresión según las especificaciones de Reactive Streams. Esto significa que los operadores de Mono y Flux manejan la contrapresión de forma nativa. Cuando un Subscriber se suscribe a un Publisher, lo primero que hace es solicitar un número inicial de elementos. Luego, a medida que procesa esos elementos, puede solicitar más (request(n)).

import reactor.core.publisher.Flux;
import org.reactivestreams.Subscription;
import org.reactivestreams.Subscriber;

public class BackpressureExample {

    public static void main(String[] args) {
        Flux.range(1, 100) // Publicador que emite 100 números
            .subscribe(new Subscriber<Integer>() {
                private Subscription s;
                private int count = 0;

                @Override
                public void onSubscribe(Subscription s) {
                    this.s = s;
                    System.out.println("Suscrito. Solicitando 2 elementos.");
                    s.request(2); // Solicita inicialmente 2 elementos
                }

                @Override
                public void onNext(Integer integer) {
                    System.out.println("Procesando: " + integer);
                    count++;
                    if (count % 2 == 0) { // Después de procesar 2 elementos, solicita 2 más
                        System.out.println("Procesados 2. Solicitando 2 más.");
                        s.request(2);
                    }
                }

                @Override
                public void onError(Throwable t) {
                    System.err.println("Error: " + t);
                }

                @Override
                public void onComplete() {
                    System.out.println("Completado.");
                }
            });
    }
}

En este ejemplo simplificado, el Subscriber controla la velocidad de emisión al solicitar solo dos elementos a la vez. Este mecanismo es transparente en la mayoría de los casos cuando usas operadores de Reactor, pero es crucial entender que está ocurriendo “bajo el capó” para un comportamiento predecible y robusto.

---

3. Contexto Reactivo: Compartiendo Información

En la programación tradicional, ThreadLocal se utiliza comúnmente para compartir información a través de diferentes métodos en el mismo hilo de ejecución, como el contexto de seguridad o un ID de correlación para logging. Sin embargo, en un entorno reactivo y no bloqueante como Spring WebFlux, donde las operaciones pueden cambiar de hilo de forma asíncrona, ThreadLocal ya no es una opción viable porque la información se perdería entre los cambios de hilo.

Aquí es donde entra el Contexto Reactivo (Context) de Project Reactor. El Context es una característica que permite adjuntar datos a un flujo reactivo, haciéndolos disponibles para cualquier operador o suscriptor a lo largo de la cadena, independientemente de qué hilo esté ejecutando la operación.

¿Cómo funciona el Contexto Reactivo?

Cada flujo Mono o Flux tiene asociado un Context. Este Context es una estructura de datos inmutable (similar a un Map) que se propaga a lo largo de la cadena de operadores. Cuando un operador necesita acceder a información del contexto, puede hacerlo a través de métodos como contextWrite().

import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
import reactor.util.context.Context;

public class ReactiveContextExample {

    public static void main(String[] args) {
        String correlationId = "corr-123";

        Mono<String> dataMono = Mono.just("Hello")
                .doOnNext(s -> {
                    // Acceder al contexto para obtener el correlationId
                    Mono.deferContextual(ctx -> {
                        String id = ctx.get("correlationId");
                        System.out.println("doOnNext: Data = " + s + ", Correlation ID from Context = " + id);
                        return Mono.empty();
                    }).subscribe(); // Suscribirse para activar el deferContextual
                })
                .contextWrite(Context.of("correlationId", correlationId)); // Escribir en el contexto

        dataMono.subscribe(
                data -> System.out.println("Subscriber: Data = " + data),
                error -> System.err.println("Subscriber Error: " + error),
                () -> System.out.println("Subscriber: Completed")
        );

        System.out.println("\n--- Otro ejemplo con Flux y múltiples valores ---");

        Flux.just("Item A", "Item B")
                .contextWrite(Context.of("traceId", "trace-xyz")) // Escribir en el contexto
                .flatMap(item ->
                        Mono.deferContextual(ctx -> {
                            String traceId = ctx.get("traceId");
                            return Mono.just("Procesando " + item + " con Trace ID: " + traceId);
                        })
                )
                .subscribe(
                        result -> System.out.println("Subscriber: " + result),
                        error -> System.err.println("Subscriber Error: " + error)
                );
    }
}

Aplicaciones Comunes del Contexto Reactivo

• Propagación de IDs de Correlación/Traza: Esencial para el logging distribuido y la observabilidad. Puedes insertar un ID de correlación al inicio del flujo y que esté disponible en cada operador y en la capa de persistencia.
• Contexto de Seguridad: Información del usuario autenticado, roles, permisos.
• Parámetros de Configuración Dinámicos: Valores que pueden variar por solicitud pero que no son parte de la carga útil principal.
• Datos Transaccionales: Si bien Spring Data R2DBC maneja las transacciones reactivas, el contexto podría usarse para almacenar metadatos relacionados con la transacción.

El Context proporciona una forma segura y reactiva de pasar información a través de los límites de los hilos, manteniendo la integridad del flujo de datos.

---

4. Testing Reactivo: Garantizando la Robustez

Probar aplicaciones reactivas requiere un enfoque ligeramente diferente al de las aplicaciones síncronas debido a la naturaleza asíncrona y no bloqueante de los flujos. Spring WebFlux y Project Reactor ofrecen herramientas poderosas para facilitar este proceso, asegurando que tus flujos de datos se comporten como esperas.

TestUtils de Reactor: StepVerifier

La herramienta más importante para probar flujos Mono y Flux es StepVerifier de Project Reactor. Permite probar secuencias reactivas de manera determinista, verificando los valores emitidos, los errores y la finalización, e incluso simulando el tiempo para probar operadores basados en tiempo.

import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.time.Duration;

class ReactiveTestingExample {

    // Prueba de un Mono simple
    @Test
    void testMono() {
        Mono<String> mono = Mono.just("Hello Reactive!");

        StepVerifier.create(mono)
                .expectNext("Hello Reactive!") // Espera un valor específico
                .expectComplete()             // Espera que el flujo se complete
                .verify();                    // Inicia la verificación
    }

    // Prueba de un Flux con múltiples elementos
    @Test
    void testFlux() {
        Flux<Integer> flux = Flux.just(1, 2, 3);

        StepVerifier.create(flux)
                .expectNext(1)
                .expectNext(2)
                .expectNext(3)
                .expectComplete()
                .verify();
    }

    // Prueba de un Flux con un error
    @Test
    void testFluxWithError() {
        Flux<String> flux = Flux.just("data1", "data2")
                .concatWith(Mono.error(new RuntimeException("Oops!")));

        StepVerifier.create(flux)
                .expectNext("data1", "data2")
                .expectError(RuntimeException.class) // Espera un error de tipo RuntimeException
                .verify();
    }

    // Prueba de un Flux con retardo (simulando tiempo)
    @Test
    void testFluxWithDelay() {
        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1)).take(3);

        StepVerifier.withVirtualTime(() -> flux) // Usa tiempo virtual para acelerar la prueba
                .expectSubscription()
                .expectNoEvent(Duration.ofSeconds(1)) // No espera eventos por 1 segundo
                .expectNext(0L)
                .thenAwait(Duration.ofSeconds(1)) // Avanza el tiempo virtual 1 segundo
                .expectNext(1L)
                .thenAwait(Duration.ofSeconds(1))
                .expectNext(2L)
                .expectComplete()
                .verify();
    }
}

StepVerifier ofrece una API fluida y encadenable para definir las expectativas sobre el flujo. withVirtualTime() es particularmente útil para probar operadores basados en tiempo sin tener que esperar el tiempo real, acelerando significativamente las pruebas.

Testing de Controladores WebFlux

Para probar controladores WebFlux, puedes usar WebTestClient. Este cliente no bloqueante permite realizar solicitudes HTTP simuladas a tu aplicación WebFlux y verificar las respuestas reactivas. Es ideal para pruebas de integración o de slice.

import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.reactive.WebFluxTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.test.web.reactive.server.WebTestClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import static org.mockito.Mockito.when;

@WebFluxTest(MyReactiveController.class) // Especifica el controlador a probar
class MyReactiveControllerTest {

    @Autowired
    private WebTestClient webTestClient; // Cliente para realizar solicitudes HTTP

    @MockBean // Simula dependencias del controlador
    private MyReactiveService myReactiveService;

    @Test
    void testGetHello() {
        when(myReactiveService.getHelloMessage()).thenReturn(Mono.just("Hello from Service!"));

        webTestClient.get().uri("/hello")
                .exchange() // Realiza la solicitud
                .expectStatus().isOk() // Verifica el código de estado HTTP
                .expectBody(String.class).isEqualTo("Hello from Service!"); // Verifica el cuerpo de la respuesta
    }

    @Test
    void testGetAllItems() {
        when(myReactiveService.getAllItems()).thenReturn(Flux.just("Item1", "Item2"));

        webTestClient.get().uri("/items")
                .exchange()
                .expectStatus().isOk()
                .expectBodyList(String.class).containsExactly("Item1", "Item2"); // Verifica una lista de elementos
    }
}

En este ejemplo:

• @WebFluxTest configura un contexto de aplicación limitado para probar solo el controlador especificado.
• @MockBean permite simular las dependencias del controlador, lo que es crucial para aislar la lógica del controlador.
• WebTestClient simula las solicitudes HTTP y permite verificar la respuesta de manera reactiva.

Combinando StepVerifier para la lógica reactiva de negocio y WebTestClient para las interacciones HTTP, puedes construir un conjunto de pruebas robusto para tus aplicaciones Spring WebFlux.

---

5. Seguridad en Aplicaciones WebFlux (Spring Security Reactivo)

La seguridad es un pilar fundamental en cualquier aplicación, y las aplicaciones reactivas no son la excepción. Spring Security Reactivo proporciona una integración fluida con Spring WebFlux, ofreciendo un modelo de seguridad no bloqueante que se adapta perfectamente al paradigma reactivo. A diferencia del Spring Security tradicional, que se basa en ThreadLocal y filtros de Servlet, la versión reactiva opera con Mono y Flux para mantener la reactividad de principio a fin.

Componentes Clave de Spring Security Reactivo

1. SecurityWebFilterChain: Reemplaza al FilterChain de Servlets y define la cadena de filtros de seguridad reactivos.
2. ReactiveUserDetailsService: Para cargar detalles del usuario de forma reactiva.
3. ReactiveAuthenticationManager: Para autenticar usuarios de forma reactiva.
4. SecurityContextRepository: Para guardar y cargar el contexto de seguridad (ej. para sesiones o JWT).

Configuración Básica

Para habilitar Spring Security Reactivo, necesitas añadir la dependencia spring-boot-starter-security y configurar tu SecurityWebFilterChain.

// build.gradle
dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-webflux'
    implementation 'org.springframework.boot:spring-boot-starter-security'
    // ...
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.core.userdetails.MapReactiveUserDetailsService;
import org.springframework.security.core.userdetails.User;
import org.springframework.security.core.userdetails.UserDetails;
import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
import org.springframework.security.crypto.password.PasswordEncoder;
import org.springframework.security.web.server.SecurityWebFilterChain;

@Configuration
@EnableWebFluxSecurity
public class SecurityConfig {

    @Bean
    public SecurityWebFilterChain securityWebFilterChain(ServerHttpSecurity http) {
        return http
                .csrf(ServerHttpSecurity.CsrfSpec::disable) // Deshabilita CSRF para APIs sin estado
                .authorizeExchange(exchanges -> exchanges
                        .pathMatchers("/public/**").permitAll() // Rutas públicas accesibles sin autenticación
                        .pathMatchers("/admin/**").hasRole("ADMIN") // Rutas solo para ADMIN
                        .anyExchange().authenticated() // Todas las demás rutas requieren autenticación
                )
                .httpBasic(httpBasic -> httpBasic.init(http)) // Habilita autenticación HTTP Basic
                .formLogin(formLogin -> formLogin.disable()) // Deshabilita el formulario de login por defecto
                .build();
    }

    @Bean
    public MapReactiveUserDetailsService userDetailsService(PasswordEncoder passwordEncoder) {
        UserDetails user = User.withUsername("user")
                .password(passwordEncoder.encode("password"))
                .roles("USER")
                .build();
        UserDetails admin = User.withUsername("admin")
                .password(passwordEncoder.encode("adminpass"))
                .roles("ADMIN")
                .build();
        return new MapReactiveUserDetailsService(user, admin);
    }

    @Bean
    public PasswordEncoder passwordEncoder() {
        return new BCryptPasswordEncoder();
    }
}

En este ejemplo:

• Deshabilitamos CSRF (común para APIs RESTful sin estado).
• Definimos reglas de autorización para diferentes rutas (/public es accesible por todos, /admin solo por usuarios con rol ADMIN).
• Configuramos HTTP Basic para la autenticación simple.
• Se define un MapReactiveUserDetailsService para usuarios en memoria, aunque en un entorno real se usaría una base de datos reactiva.

Accediendo al Usuario Autenticado

En Spring WebFlux, puedes acceder al usuario autenticado usando Mono<Principal> o Mono<Authentication> en tus controladores o servicios.

import org.springframework.security.core.Authentication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;

import java.security.Principal;

@RestController
public class SecuredController {

    @GetMapping("/secure/user-info")
    public Mono<String> getUserInfo(Mono<Principal> principalMono) {
        return principalMono.map(principal -> "Hola, " + principal.getName() + "! Eres un usuario autenticado.");
    }

    @GetMapping("/admin/dashboard")
    public Mono<String> getAdminDashboard(Mono<Authentication> authenticationMono) {
        return authenticationMono.map(auth -> "Bienvenido al Dashboard de Admin, " + auth.getName() + "! Roles: " + auth.getAuthorities());
    }
}

Uso de JWT (JSON Web Tokens)

Para aplicaciones sin estado, el uso de JWT es una práctica común. Spring Security Reactivo facilita la implementación de autenticación basada en JWT. Generalmente, esto implica:

1. Un endpoint de login que recibe credenciales y devuelve un JWT.
2. Un filtro de seguridad que intercepta las solicitudes, valida el JWT en el encabezado Authorization y construye un Authentication reactivo.

Puedes crear tu propio ServerWebExchangeMatcher y ServerAuthenticationConverter para procesar el token y autenticar al usuario sin necesidad de sesiones.

Spring Security Reactivo se integra perfectamente con el modelo de programación reactiva, asegurando que tus mecanismos de seguridad no introduzcan bloqueos o cuellos de botella en tus aplicaciones de alto rendimiento.

---

6. WebSockets con WebFlux: Comunicación Bidireccional en Tiempo Real

Mientras que Server-Sent Events (SSE) son excelentes para la comunicación unidireccional del servidor al cliente, las aplicaciones que requieren comunicación bidireccional en tiempo real, como chats, juegos en línea o herramientas de colaboración, necesitan WebSockets. WebSockets proporcionan un canal de comunicación dúplex completo a través de una única conexión TCP. Spring WebFlux ofrece un soporte robusto y reactivo para WebSockets.

¿Cómo funcionan los WebSockets?

A diferencia de HTTP, que es de corta duración y sin estado, los WebSockets comienzan con un handshake HTTP. Una vez que este handshake es exitoso, la conexión se “actualiza” a un protocolo WebSocket, permaneciendo abierta indefinidamente. Esto permite que tanto el cliente como el servidor envíen mensajes de forma asíncrona en cualquier momento.

WebSockets con Spring WebFlux

Spring WebFlux proporciona una API funcional para manejar WebSockets, aprovechando Flux y Mono para la gestión de mensajes reactivos.

1. WebSocketHandler: Es la interfaz principal que implementas para manejar la lógica de la conexión WebSocket. El método handle recibe un WebSocketSession que te permite enviar y recibir mensajes.

2. WebSocketHandlerAdapter y SimpleUrlHandlerMapping: Estos beans son necesarios para mapear las URLs a tus WebSocketHandlers específicos.

Configuración de WebSocket

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.server.WebSocketService;
import org.springframework.web.reactive.socket.server.support.HandshakeWebSocketService;
import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
import org.springframework.web.reactive.socket.server.upgrade.ReactorNettyRequestUpgradeStrategy;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class WebSocketConfig {

    @Bean
    public SimpleUrlHandlerMapping webSocketHandlerMapping(WebSocketHandler echoHandler) {
        Map<String, WebSocketHandler> map = new HashMap<>();
        map.put("/echo", echoHandler); // Mapea /echo a nuestro handler
        map.put("/time-stream", new TimeStreamWebSocketHandler()); // Otro handler
        return new SimpleUrlHandlerMapping(map);
    }

    @Bean
    public WebSocketHandlerAdapter handlerAdapter(WebSocketService webSocketService) {
        return new WebSocketHandlerAdapter(webSocketService);
    }

    @Bean
    public WebSocketService webSocketService() {
        // Usa Reactor Netty por defecto, que es el servidor webflux por defecto
        return new HandshakeWebSocketService(new ReactorNettyRequestUpgradeStrategy());
    }

    @Bean
    public WebSocketHandler echoHandler() {
        return session -> session.send(
                session.receive() // Recibe mensajes del cliente
                        .doOnNext(message -> System.out.println("Received: " + message.getPayloadAsText()))
                        .map(message -> session.textMessage("ECHO: " + message.getPayloadAsText())) // Eco de vuelta
        ).and(session.receive()
                .doOnError(throwable -> System.err.println("Error en la conexión WebSocket: " + throwable.getMessage()))
                .then()); // Mantener la conexión abierta hasta que se complete o haya un error
    }
}

Creando un WebSocketHandler

Aquí tienes un ejemplo de un WebSocketHandler que envía la hora actual cada segundo:

// En un archivo separado o como inner class
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.LocalDateTime;

public class TimeStreamWebSocketHandler implements WebSocketHandler {

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // Envía un mensaje cada segundo al cliente
        Flux<WebSocketMessage> output = Flux.interval(Duration.ofSeconds(1))
                .map(value -> session.textMessage("Current Time: " + LocalDateTime.now()));

        // Mantén la conexión abierta para recibir mensajes (aunque este handler no los procese)
        // La conexión se cierra cuando el Mono<Void> retornado se completa
        return session.send(output)
                .and(session.receive() // Esto es importante para mantener la conexión abierta
                        .doOnNext(message -> System.out.println("Received from client on time stream: " + message.getPayloadAsText()))
                        .then()); // No hacemos nada con los mensajes recibidos aquí, solo los logueamos
    }
}

Cliente JavaScript para WebSockets

const ws = new WebSocket('ws://localhost:8080/echo'); // Para el handler de eco

ws.onopen = function(event) {
    console.log("Conectado al WebSocket!");
    ws.send("Hola desde el cliente!");
};

ws.onmessage = function(event) {
    console.log("Mensaje recibido del servidor:", event.data);
};

ws.onclose = function(event) {
    console.log("Conexión WebSocket cerrada:", event.code, event.reason);
};

ws.onerror = function(error) {
    console.error("Error WebSocket:", error);
};

// Para enviar más mensajes:
// ws.send("Otro mensaje...");

// Para el handler de tiempo:
// const wsTime = new WebSocket('ws://localhost:8080/time-stream');
// wsTime.onmessage = function(event) {
//     console.log("Tiempo recibido:", event.data);
// };

WebSockets con WebFlux te permiten construir aplicaciones de comunicación en tiempo real altamente eficientes, aprovechando la capacidad de Spring para manejar flujos de datos reactivos de forma nativa.

---

7. Buenas Prácticas en Producción para Aplicaciones WebFlux

Desarrollar una aplicación WebFlux es solo una parte del desafío; desplegarla y mantenerla en producción requiere atención a varias buenas prácticas para asegurar su rendimiento, estabilidad y observabilidad.

1. Monitoreo y Observabilidad

Las aplicaciones reactivas pueden ser más difíciles de depurar sin las herramientas adecuadas debido a la naturaleza asíncrona y la transición de hilos.

• Métricas (Micrometer/Prometheus): Spring Boot Actuator, combinado con Micrometer, facilita la exposición de métricas (JVM, WebFlux, Reactor, etc.) que pueden ser recolectadas por sistemas como Prometheus y visualizadas en Grafana. Monitorea la latencia, el rendimiento del Event Loop, el uso de memoria y el número de conexiones activas.
• Logging (Structured Logging): Utiliza un sistema de logging que soporte logging estructurado (ej. SLF4J con Logback configurado para JSON) para facilitar el análisis con herramientas como ELK Stack (Elasticsearch, Logstash, Kibana) o Grafana Loki.
• APM (Application Performance Monitoring): Herramientas como Dynatrace, New Relic o AppDynamics pueden proporcionar visibilidad profunda en el rendimiento de tu aplicación, incluyendo la trazabilidad de transacciones a través de hilos y servicios.
• Tracing (Brave/OpenTelemetry): Implementa Distributed Tracing (ej. con Spring Cloud Sleuth y Zipkin/Jaeger) para seguir el rastro de una solicitud a través de múltiples servicios, especialmente crucial en arquitecturas de microservicios reactivos.

2. Gestión de Recursos

• Connection Pooling: Asegúrate de que tus conexiones a bases de datos reactivas (R2DBC, MongoDB reactive drivers) o a otros servicios externos (WebClient) utilicen connection pooling para evitar la sobrecarga y el agotamiento de recursos.
• Timeouts: Configura timeouts apropiados en WebClient y en tus servidores para evitar que las solicitudes de larga duración o los servicios externos lentos bloqueen los recursos del Event Loop.
• Límites de Conexión: Establece límites de conexión adecuados en tus servidores (Netty, Undertow) para prevenir la sobrecarga.

3. Contrapresión Efectiva

Aunque Reactor maneja la contrapresión de forma nativa, es crucial entender cuándo y cómo se aplica, especialmente al integrar con sistemas que no son reactivos o que no la soportan. Asegúrate de que tus flujos de datos estén diseñados para manejar el backpressure correctamente para evitar la sobrecarga del consumidor.

4. Seguridad

• Principio de Mínimo Privilegio: Asegúrate de que tu aplicación solo tenga los permisos necesarios para realizar sus funciones.
• Secret Management: No guardes credenciales directamente en el código o en archivos de configuración. Utiliza soluciones de gestión de secretos como HashiCorp Vault, AWS Secrets Manager o Kubernetes Secrets.
• Actualizaciones y Parches: Mantén tus dependencias de Spring Boot, Spring Security y Reactor actualizadas para beneficiarte de las últimas correcciones de seguridad.
• HTTPS: Siempre utiliza HTTPS en producción para asegurar la comunicación cliente-servidor.

5. Configuración y Despliegue

• Externalización de la Configuración: Utiliza Spring Cloud Config Server, o simplemente application.properties/application.yml con perfiles, y variables de entorno para gestionar la configuración de forma externa al artefacto de despliegue.
• Contenedores (Docker/Kubernetes): Empaquetar tu aplicación en un contenedor Docker facilita el despliegue, la escalabilidad y la gestión de dependencias en entornos como Kubernetes.
• Liveness y Readiness Probes: En Kubernetes, configura Liveness y Readiness Probes para que el orquestador pueda saber cuándo tu aplicación está saludable y lista para recibir tráfico. Spring Boot Actuator proporciona endpoints /actuator/health que son perfectos para esto.
• Escalabilidad: Las aplicaciones WebFlux son inherentemente escalables horizontalmente. Asegúrate de que tu infraestructura de despliegue (Kubernetes, balanceadores de carga) pueda escalar tu aplicación de manera eficiente.

6. Pruebas de Carga y Rendimiento

Realiza pruebas de carga exhaustivas para simular escenarios de alto tráfico y verificar cómo se comporta tu aplicación WebFlux bajo presión. Esto te ayudará a identificar cuellos de botella y a optimizar la configuración.

7. Manejo de Errores Robustos

• ErrorWebExceptionHandler: Asegúrate de tener un ErrorWebExceptionHandler global bien configurado para manejar excepciones no capturadas y proporcionar respuestas de error consistentes y amigables para el cliente, sin exponer detalles internos.
• Circuit Breakers: Implementa patrones de Circuit Breaker (ej. con Resilience4j) al interactuar con servicios externos para evitar cascadas de fallos cuando un servicio dependiente no está disponible o es lento.

Al seguir estas buenas prácticas, puedes asegurar que tus aplicaciones Spring WebFlux no solo sean rápidas y eficientes en desarrollo, sino también robustas, seguras y fáciles de operar en producción.

---

Conclusión

En esta cuarta entrega de nuestra serie sobre Spring WebFlux, hemos explorado características avanzadas y cruciales que elevan el desarrollo de aplicaciones reactivas. Desde la implementación de Server-Sent Events (SSE) para flujos de datos unidireccionales hasta la robusta comunicación WebSocket para interacciones bidireccionales en tiempo real, hemos visto cómo Spring WebFlux simplifica la construcción de aplicaciones de tiempo real.

Hemos profundizado en la importancia de la contrapresión (backpressure), un mecanismo vital para garantizar la estabilidad del sistema al permitir que los consumidores controlen el flujo de datos. La gestión del contexto reactivo se ha revelado como una solución elegante para compartir información a través de los límites de los hilos en un entorno asíncrono, mientras que las herramientas de testing reactivo como StepVerifier y WebTestClient demuestran ser indispensables para asegurar la corrección de nuestros flujos. Finalmente, abordamos la integración de Spring Security Reactivo para asegurar nuestras aplicaciones de forma no bloqueante y delineamos un conjunto de buenas prácticas para la producción, fundamentales para el monitoreo, la estabilidad y la escalabilidad de nuestras aplicaciones WebFlux.

Esta serie ha cubierto los pilares esenciales para construir aplicaciones reactivas de alto rendimiento con Spring WebFlux. Con una base sólida en fundamentos, arquitectura, comunicación de datos, seguridad, pruebas y consideraciones de producción, estás bien preparado para enfrentar desafíos reales y llevar tus aplicaciones reactivas al siguiente nivel.

Como continuación natural de este camino, te recomendamos explorar algunas áreas complementarias que potenciarán aún más tus habilidades en entornos reactivos:

• R2DBC a profundidad: Explora la integración con bases de datos relacionales reactivas, optimización de consultas y rendimiento en entornos de alta demanda.
• Spring Cloud Gateway: Descubre cómo usar esta herramienta basada en WebFlux para implementar enrutamiento, seguridad y resiliencia en arquitecturas de microservicios.
• Programación Reactiva en el Frontend: Investiga cómo frameworks como React, Angular o Vue pueden conectarse eficientemente con backends WebFlux en escenarios de tiempo real.
• WebFlux y GraalVM Native Image: Evalúa las ventajas de empaquetar tus aplicaciones como imágenes nativas para mejorar el rendimiento y reducir el consumo de recursos.
• Patrones de resiliencia con WebFlux: Profundiza en técnicas como Circuit Breaker, Retry, Timeout y Rate Limiting mediante el uso de Resilience4j en entornos reactivos.

Explorar estos temas no solo ampliará tu dominio técnico, sino que también te permitirá diseñar soluciones más eficientes, resilientes y adaptadas a los retos actuales del desarrollo moderno. La programación reactiva, bien aplicada, abre la puerta a aplicaciones verdaderamente escalables y sensibles a la demanda del usuario.