Chapitre 1: Programmation Réactive avec Spring WebFlux

Objectifs du Chapitre

• Comprendre le paradigme réactif vs impératif
• Maîtriser les types Mono et Flux
• Créer des API REST réactives
• Utiliser R2DBC pour l'accès aux données réactif
• Comprendre les opérateurs de transformation

---

1. Introduction à la Programmation Réactive

1.1 Le Problème du Modèle Bloquant

Modèle Spring MVC (bloquant):

Thread-1: [Requête]--[Attente DB 100ms]--[Réponse]
Thread-2: [Requête]--[Attente DB 150ms]--[Réponse]
Thread-3: [Requête]--[Attente API 200ms]--[Réponse]
...
Thread-200: Pool épuisé! Nouvelles requêtes en attente...

Problème: 1 thread = 1 requête = ressources gaspillées pendant l'attente

1.2 Le Modèle Réactif

Modèle WebFlux (non-bloquant):

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

- Peu de threads (typiquement: nb CPU cores)
- Pas d'attente bloquante
- Callbacks quand les données sont prêtes
- Scalabilité accrue

1.3 Reactive Streams Specification

Les 4 interfaces de base:

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 et Flux

2.1 Mono: 0 ou 1 élément

// Création
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());

// Opérations
Mono<String> transformed = Mono.just("hello")
    .map(String::toUpperCase)              // HELLO
    .flatMap(s -> fetchFromDb(s))          // Appel async
    .defaultIfEmpty("default")             // Valeur par défaut
    .onErrorReturn("error fallback");      // En cas d'erreur

2.2 Flux: 0 à N éléments

// Création
Flux<Integer> empty = Flux.empty();
Flux<Integer> just = Flux.just(1, 2, 3);
Flux<Integer> range = Flux.range(1, 10);  // 1 à 10
Flux<Integer> fromIterable = Flux.fromIterable(list);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));  // Infini

// Opérations
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)                               // Limite à 3
    .collectList()                         // Mono<List<String>>
    .block();                              // Bloque (tests uniquement!)

2.3 Opérateurs Courants

Opérateur	Description	Exemple
map	Transformation synchrone	.map(s -> s.toUpperCase())
flatMap	Transformation async	.flatMap(id -> repository.findById(id))
filter	Filtrage	.filter(n -> n > 0)
take	Limite le nombre	.take(5)
skip	Ignore les premiers	.skip(2)
distinct	Éléments uniques	.distinct()
defaultIfEmpty	Valeur par défaut	.defaultIfEmpty("N/A")
switchIfEmpty	Publisher alternatif	.switchIfEmpty(Mono.just("default"))
onErrorReturn	Fallback en cas d'erreur	.onErrorReturn("error")
onErrorResume	Publisher alternatif	.onErrorResume(e -> Mono.empty())
retry	Réessayer	.retry(3)
timeout	Limite de temps	.timeout(Duration.ofSeconds(5))
zip	Combiner	Mono.zip(mono1, mono2)
merge	Fusionner en parallèle	Flux.merge(flux1, flux2)
concat	Fusionner séquentiellement	Flux.concat(flux1, flux2)

---

3. Spring WebFlux Configuration

3.1 Dépendances Maven

<dependencies>
    <!-- WebFlux (remplace spring-boot-starter-web) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    
    <!-- R2DBC pour accès données réactif -->
    <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>
    
    <!-- Ou H2 pour dev -->
    <dependency>
        <groupId>io.r2dbc</groupId>
        <artifactId>r2dbc-h2</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

3.2 Configuration

spring:
  r2dbc:
    url: r2dbc:h2:mem:///testdb
    username: sa
    password:
  
  # Initialisation du schéma
  sql:
    init:
      mode: always
      schema-locations: classpath:schema.sql

---

4. API REST Réactive

4.1 Entity

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 et 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 Repository Réactif

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 Service

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 Controller Réactif

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. Gestion des Erreurs

5.1 Exception Handler Réactif

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 Gestion dans les Opérateurs

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: Client HTTP Réactif

6.1 Configuration

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 Utilisation

@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. Bonnes Pratiques

8.1 Ne Jamais Bloquer

// Mauvais - Bloque le thread
String result = mono.block();

// Bon - Retourne le Mono
return mono;

8.2 Utiliser subscribeOn pour les Opérations Bloquantes

// Si vous devez appeler du code bloquant
Mono.fromCallable(() -> blockingOperation())
    .subscribeOn(Schedulers.boundedElastic())

8.3 Logging avec doOn

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

---

9. Résumé

Concept	Description
Mono	0 ou 1 élément asynchrone
Flux	0 à N éléments asynchrones
WebFlux	Framework web réactif Spring
R2DBC	Accès données réactif
WebClient	Client HTTP réactif
StepVerifier	Test des publishers

Quand Utiliser WebFlux

• Haute concurrence (milliers de connexions simultanées)
• Streaming de données
• Applications temps réel
• Intégration avec services externes nombreux

Quand Rester sur Spring MVC

• Applications CRUD simples
• Équipe non familière avec le réactif
• Bibliothèques tierces bloquantes
• Debugging simple nécessaire

---

Chapitre Suivant: Cache avec Redis

Retour au Module 14