Capítulo 1: Programación Reactiva con Spring WebFlux

Objetivos del Capítulo

• Comprender el paradigma reactivo vs imperativo
• Dominar los tipos Mono y Flux
• Crear APIs REST reactivas
• Utilizar R2DBC para el acceso a datos reactivo
• Comprender los operadores de transformación

---

1. Introducción a la Programación Reactiva

1.1 El Problema del Modelo Bloqueante

Modelo Spring MVC (bloqueante):

Thread-1: [Solicitud]--[Espera BD 100ms]--[Respuesta]
Thread-2: [Solicitud]--[Espera BD 150ms]--[Respuesta]
Thread-3: [Solicitud]--[Espera API 200ms]--[Respuesta]
...
Thread-200: ¡Pool agotado! Nuevas solicitudes en espera...

Problema: 1 thread = 1 solicitud = recursos desperdiciados durante la espera

1.2 El Modelo Reactivo

Modelo WebFlux (no bloqueante):

Event Loop: [Req1]→[Req2]→[Req3]→...→[Callback Req1]→[Callback Req2]

- Pocos threads (típicamente: núcleos de CPU)
- Sin espera bloqueante
- Callbacks cuando los datos están listos
- Escalabilidad aumentada

1.3 Especificación Reactive Streams

Las 4 interfaces básicas:

public interface Publisher<T> {
    void subscribe(Subscriber<? super T> s);
}

public interface Subscriber<T> {
    void onSubscribe(Subscription s);
    void onNext(T t);
    void onError(Throwable t);
    void onComplete();
}

public interface Subscription {
    void request(long n);  // Backpressure
    void cancel();
}

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

---

2. Project Reactor: Mono y Flux

2.1 Mono: 0 o 1 elemento

// Creación
Mono<String> empty = Mono.empty();
Mono<String> just = Mono.just("Hello");
Mono<String> error = Mono.error(new RuntimeException("Oops"));
Mono<String> fromSupplier = Mono.fromSupplier(() -> computeValue());
Mono<String> fromCallable = Mono.fromCallable(() -> blockingCall());

// Operaciones
Mono<String> transformed = Mono.just("hello")
    .map(String::toUpperCase)              // HELLO
    .flatMap(s -> fetchFromDb(s))          // Llamada async
    .defaultIfEmpty("default")             // Valor por defecto
    .onErrorReturn("error fallback");      // En caso de error

2.2 Flux: 0 a N elementos

// Creación
Flux<Integer> empty = Flux.empty();
Flux<Integer> just = Flux.just(1, 2, 3);
Flux<Integer> range = Flux.range(1, 10);  // 1 a 10
Flux<Integer> fromIterable = Flux.fromIterable(list);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));  // Infinito

// Operaciones
Flux<String> transformed = Flux.just(1, 2, 3, 4, 5)
    .filter(n -> n % 2 == 0)               // 2, 4
    .map(n -> "Number: " + n)              // "Number: 2", "Number: 4"
    .take(3)                               // Limita a 3
    .collectList()                         // Mono<List<String>>
    .block();                              // Bloquea (¡solo pruebas!)

2.3 Operadores Comunes

Operador	Descripción	Ejemplo
map	Transformación síncrona	.map(s -> s.toUpperCase())
flatMap	Transformación async	.flatMap(id -> repository.findById(id))
filter	Filtrado	.filter(n -> n > 0)
take	Limita la cantidad	.take(5)
skip	Ignora los primeros	.skip(2)
distinct	Elementos únicos	.distinct()
defaultIfEmpty	Valor por defecto	.defaultIfEmpty("N/A")
switchIfEmpty	Publisher alternativo	.switchIfEmpty(Mono.just("default"))
onErrorReturn	Fallback en caso de error	.onErrorReturn("error")
onErrorResume	Publisher alternativo	.onErrorResume(e -> Mono.empty())
retry	Reintentar	.retry(3)
timeout	Límite de tiempo	.timeout(Duration.ofSeconds(5))
zip	Combinar	Mono.zip(mono1, mono2)
merge	Fusionar en paralelo	Flux.merge(flux1, flux2)
concat	Fusionar secuencialmente	Flux.concat(flux1, flux2)

---

3. Configuración de Spring WebFlux

3.1 Dependencias Maven

<dependencies>
    <!-- WebFlux (reemplaza spring-boot-starter-web) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    
    <!-- R2DBC para acceso a datos reactivo -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-r2dbc</artifactId>
    </dependency>
    
    <!-- Driver R2DBC PostgreSQL -->
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>r2dbc-postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    
    <!-- O H2 para desarrollo -->
    <dependency>
        <groupId>io.r2dbc</groupId>
        <artifactId>r2dbc-h2</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

3.2 Configuración

spring:
  r2dbc:
    url: r2dbc:h2:mem:///testdb
    username: sa
    password:
  
  # Inicialización del esquema
  sql:
    init:
      mode: always
      schema-locations: classpath:schema.sql

---

4. API REST Reactiva

4.1 Entidad

package com.example.reactive.model;

import org.springframework.data.annotation.Id;
import org.springframework.data.relational.core.mapping.Table;

@Table("products")
public class Product {
    
    @Id
    private Long id;
    private String name;
    private Double price;
    private Integer stock;
    
    public Product() {}
    
    public Product(String name, Double price, Integer stock) {
        this.name = name;
        this.price = price;
        this.stock = stock;
    }
    
    // Getters y Setters
    public Long getId() { return id; }
    public void setId(Long id) { this.id = id; }
    
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    
    public Double getPrice() { return price; }
    public void setPrice(Double price) { this.price = price; }
    
    public Integer getStock() { return stock; }
    public void setStock(Integer stock) { this.stock = stock; }
}

4.2 Repositorio Reactivo

package com.example.reactive.repository;

import com.example.reactive.model.Product;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public interface ProductRepository extends ReactiveCrudRepository<Product, Long> {
    
    Flux<Product> findByNameContaining(String name);
    
    Flux<Product> findByPriceLessThan(Double price);
    
    @Query("SELECT * FROM products WHERE stock > 0 ORDER BY price ASC")
    Flux<Product> findAvailableProducts();
    
    @Query("UPDATE products SET stock = stock - :quantity WHERE id = :id AND stock >= :quantity")
    Mono<Integer> decrementStock(Long id, Integer quantity);
}

4.3 Servicio

package com.example.reactive.service;

import com.example.reactive.model.Product;
import com.example.reactive.repository.ProductRepository;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
public class ProductService {
    
    private final ProductRepository productRepository;
    
    public ProductService(ProductRepository productRepository) {
        this.productRepository = productRepository;
    }
    
    public Flux<Product> getAllProducts() {
        return productRepository.findAll();
    }
    
    public Mono<Product> getProductById(Long id) {
        return productRepository.findById(id);
    }
    
    public Mono<Product> createProduct(Product product) {
        return productRepository.save(product);
    }
    
    public Mono<Product> updateProduct(Long id, Product product) {
        return productRepository.findById(id)
            .flatMap(existing -> {
                existing.setName(product.getName());
                existing.setPrice(product.getPrice());
                existing.setStock(product.getStock());
                return productRepository.save(existing);
            });
    }
    
    public Mono<Void> deleteProduct(Long id) {
        return productRepository.deleteById(id);
    }
    
    public Flux<Product> searchByName(String name) {
        return productRepository.findByNameContaining(name);
    }
    
    public Mono<Boolean> purchaseProduct(Long id, Integer quantity) {
        return productRepository.decrementStock(id, quantity)
            .map(rowsUpdated -> rowsUpdated > 0);
    }
}

4.4 Controlador Reactivo

package com.example.reactive.controller;

import com.example.reactive.model.Product;
import com.example.reactive.service.ProductService;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/api/products")
public class ProductController {
    
    private final ProductService productService;
    
    public ProductController(ProductService productService) {
        this.productService = productService;
    }
    
    @GetMapping
    public Flux<Product> getAllProducts() {
        return productService.getAllProducts();
    }
    
    @GetMapping("/{id}")
    public Mono<ResponseEntity<Product>> getProductById(@PathVariable Long id) {
        return productService.getProductById(id)
            .map(ResponseEntity::ok)
            .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    
    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<Product> createProduct(@RequestBody Product product) {
        return productService.createProduct(product);
    }
    
    @PutMapping("/{id}")
    public Mono<ResponseEntity<Product>> updateProduct(
            @PathVariable Long id, 
            @RequestBody Product product) {
        return productService.updateProduct(id, product)
            .map(ResponseEntity::ok)
            .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    
    @DeleteMapping("/{id}")
    @ResponseStatus(HttpStatus.NO_CONTENT)
    public Mono<Void> deleteProduct(@PathVariable Long id) {
        return productService.deleteProduct(id);
    }
    
    @GetMapping("/search")
    public Flux<Product> searchProducts(@RequestParam String name) {
        return productService.searchByName(name);
    }
    
    @PostMapping("/{id}/purchase")
    public Mono<ResponseEntity<String>> purchaseProduct(
            @PathVariable Long id,
            @RequestParam Integer quantity) {
        return productService.purchaseProduct(id, quantity)
            .map(success -> success 
                ? ResponseEntity.ok("Purchase successful")
                : ResponseEntity.badRequest().body("Insufficient stock"));
    }
}

---

5. Gestión de Errores

5.1 Exception Handler Reactivo

package com.example.reactive.exception;

import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.util.Map;

@RestControllerAdvice
public class GlobalExceptionHandler {
    
    @ExceptionHandler(ProductNotFoundException.class)
    public Mono<ResponseEntity<Map<String, Object>>> handleNotFound(
            ProductNotFoundException ex) {
        return Mono.just(ResponseEntity
            .status(HttpStatus.NOT_FOUND)
            .body(Map.of(
                "timestamp", LocalDateTime.now().toString(),
                "status", 404,
                "error", "Not Found",
                "message", ex.getMessage()
            )));
    }
    
    @ExceptionHandler(Exception.class)
    public Mono<ResponseEntity<Map<String, Object>>> handleGeneric(Exception ex) {
        return Mono.just(ResponseEntity
            .status(HttpStatus.INTERNAL_SERVER_ERROR)
            .body(Map.of(
                "timestamp", LocalDateTime.now().toString(),
                "status", 500,
                "error", "Internal Server Error",
                "message", ex.getMessage()
            )));
    }
}

5.2 Gestión en los Operadores

public Mono<Product> getProductOrDefault(Long id) {
    return productRepository.findById(id)
        .switchIfEmpty(Mono.error(new ProductNotFoundException(id)))
        .onErrorResume(e -> {
            log.error("Error fetching product {}: {}", id, e.getMessage());
            return Mono.empty();
        });
}

---

6. WebClient: Cliente HTTP Reactivo

6.1 Configuración

package com.example.reactive.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.client.WebClient;

@Configuration
public class WebClientConfig {
    
    @Bean
    public WebClient webClient(WebClient.Builder builder) {
        return builder
            .baseUrl("http://external-api.com")
            .defaultHeader("Accept", "application/json")
            .build();
    }
}

6.2 Uso

@Service
public class ExternalApiService {
    
    private final WebClient webClient;
    
    public ExternalApiService(WebClient webClient) {
        this.webClient = webClient;
    }
    
    public Mono<ExternalData> fetchData(String id) {
        return webClient.get()
            .uri("/data/{id}", id)
            .retrieve()
            .onStatus(HttpStatusCode::is4xxClientError, 
                response -> Mono.error(new NotFoundException("Data not found")))
            .onStatus(HttpStatusCode::is5xxServerError,
                response -> Mono.error(new ServiceException("External service error")))
            .bodyToMono(ExternalData.class)
            .timeout(Duration.ofSeconds(5))
            .retry(3);
    }
    
    public Flux<ExternalData> fetchAllData() {
        return webClient.get()
            .uri("/data")
            .retrieve()
            .bodyToFlux(ExternalData.class);
    }
    
    public Mono<ExternalData> createData(ExternalData data) {
        return webClient.post()
            .uri("/data")
            .bodyValue(data)
            .retrieve()
            .bodyToMono(ExternalData.class);
    }
}

---

7. Testing WebFlux

7.1 StepVerifier

import reactor.test.StepVerifier;

@Test
void testGetAllProducts() {
    Flux<Product> products = productService.getAllProducts();
    
    StepVerifier.create(products)
        .expectNextCount(3)
        .verifyComplete();
}

@Test
void testGetProductById() {
    Mono<Product> product = productService.getProductById(1L);
    
    StepVerifier.create(product)
        .assertNext(p -> {
            assertThat(p.getName()).isEqualTo("Laptop");
            assertThat(p.getPrice()).isEqualTo(999.99);
        })
        .verifyComplete();
}

@Test
void testProductNotFound() {
    Mono<Product> product = productService.getProductById(999L);
    
    StepVerifier.create(product)
        .verifyComplete();  // Empty Mono
}

7.2 WebTestClient

@WebFluxTest(ProductController.class)
class ProductControllerTest {
    
    @Autowired
    private WebTestClient webTestClient;
    
    @MockBean
    private ProductService productService;
    
    @Test
    void testGetAllProducts() {
        when(productService.getAllProducts())
            .thenReturn(Flux.just(
                new Product("Laptop", 999.99, 10),
                new Product("Phone", 699.99, 20)
            ));
        
        webTestClient.get()
            .uri("/api/products")
            .exchange()
            .expectStatus().isOk()
            .expectBodyList(Product.class)
            .hasSize(2);
    }
    
    @Test
    void testGetProductById() {
        Product product = new Product("Laptop", 999.99, 10);
        product.setId(1L);
        
        when(productService.getProductById(1L))
            .thenReturn(Mono.just(product));
        
        webTestClient.get()
            .uri("/api/products/1")
            .exchange()
            .expectStatus().isOk()
            .expectBody()
            .jsonPath("$.name").isEqualTo("Laptop")
            .jsonPath("$.price").isEqualTo(999.99);
    }
    
    @Test
    void testCreateProduct() {
        Product product = new Product("Tablet", 499.99, 15);
        Product saved = new Product("Tablet", 499.99, 15);
        saved.setId(1L);
        
        when(productService.createProduct(any()))
            .thenReturn(Mono.just(saved));
        
        webTestClient.post()
            .uri("/api/products")
            .bodyValue(product)
            .exchange()
            .expectStatus().isCreated()
            .expectBody()
            .jsonPath("$.id").isEqualTo(1);
    }
}

---

8. Buenas Prácticas

8.1 Nunca Bloquear

// Malo - Bloquea el thread
String result = mono.block();

// Bueno - Retorna el Mono
return mono;

8.2 Usar subscribeOn para Operaciones Bloqueantes

// Si debe llamar código bloqueante
Mono.fromCallable(() -> blockingOperation())
    .subscribeOn(Schedulers.boundedElastic())

8.3 Logging con doOn

flux.doOnNext(item -> log.debug("Processing: {}", item))
    .doOnError(e -> log.error("Error: {}", e.getMessage()))
    .doOnComplete(() -> log.info("Processing complete"))

---

9. Resumen

Concepto	Descripción
Mono	0 o 1 elemento asíncrono
Flux	0 a N elementos asíncronos
WebFlux	Framework web reactivo de Spring
R2DBC	Acceso a datos reactivo
WebClient	Cliente HTTP reactivo
StepVerifier	Test de publishers

Cuándo Usar WebFlux

• Alta concurrencia (miles de conexiones simultáneas)
• Streaming de datos
• Aplicaciones en tiempo real
• Integración con numerosos servicios externos

Cuándo Mantenerse en Spring MVC

• Aplicaciones CRUD simples
• Equipo no familiarizado con lo reactivo
• Bibliotecas de terceros bloqueantes
• Necesidad de debugging sencillo

---

Capítulo Siguiente: Caché con Redis

Volver al Módulo 14