Capítulo 1: Programação Reativa com Spring WebFlux

Objetivos do Capítulo

• Compreender o paradigma reativo vs imperativo
• Dominar os tipos Mono e Flux
• Criar APIs REST reativas
• Utilizar R2DBC para acesso reativo a dados
• Compreender os operadores de transformação

---

1. Introdução à Programação Reativa

1.1 O Problema do Modelo Bloqueante

Modelo Spring MVC (bloqueante):

Thread-1: [Requisição]--[Espera DB 100ms]--[Resposta]
Thread-2: [Requisição]--[Espera DB 150ms]--[Resposta]
Thread-3: [Requisição]--[Espera API 200ms]--[Resposta]
...
Thread-200: Pool esgotado! Novas requisições aguardando...

Problema: 1 thread = 1 requisição = recursos desperdiçados durante a espera

1.2 O Modelo Reativo

Modelo WebFlux (não-bloqueante):

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

- Poucas threads (tipicamente: nº de núcleos de CPU)
- Sem espera bloqueante
- Callbacks quando os dados estão prontos
- Escalabilidade aumentada

1.3 Reactive Streams Specification

As 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 e Flux

2.1 Mono: 0 ou 1 elemento

// Criação
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());

// Operações
Mono<String> transformed = Mono.just("hello")
    .map(String::toUpperCase)              // HELLO
    .flatMap(s -> fetchFromDb(s))          // Chamada async
    .defaultIfEmpty("default")             // Valor padrão
    .onErrorReturn("error fallback");      // Em caso de erro

2.2 Flux: 0 a N elementos

// Criação
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

// Operações
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();                              // Bloqueia (apenas para testes!)

2.3 Operadores Comuns

Operador	Descrição	Exemplo
map	Transformação síncrona	.map(s -> s.toUpperCase())
flatMap	Transformação assíncrona	.flatMap(id -> repository.findById(id))
filter	Filtragem	.filter(n -> n > 0)
take	Limita a quantidade	.take(5)
skip	Ignora os primeiros	.skip(2)
distinct	Elementos únicos	.distinct()
defaultIfEmpty	Valor padrão	.defaultIfEmpty("N/A")
switchIfEmpty	Publisher alternativo	.switchIfEmpty(Mono.just("default"))
onErrorReturn	Fallback em caso de erro	.onErrorReturn("error")
onErrorResume	Publisher alternativo	.onErrorResume(e -> Mono.empty())
retry	Tentar novamente	.retry(3)
timeout	Limite de tempo	.timeout(Duration.ofSeconds(5))
zip	Combinar	Mono.zip(mono1, mono2)
merge	Mesclar em paralelo	Flux.merge(flux1, flux2)
concat	Mesclar sequencialmente	Flux.concat(flux1, flux2)

---

3. Configuração do Spring WebFlux

3.1 Dependências Maven

<dependencies>
    <!-- WebFlux (substitui spring-boot-starter-web) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    
    <!-- R2DBC para acesso reativo a dados -->
    <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 para dev -->
    <dependency>
        <groupId>io.r2dbc</groupId>
        <artifactId>r2dbc-h2</artifactId>
        <scope>runtime</scope>
    </dependency>
</dependencies>

3.2 Configuração

spring:
  r2dbc:
    url: r2dbc:h2:mem:///testdb
    username: sa
    password:
  
  # Inicialização do schema
  sql:
    init:
      mode: always
      schema-locations: classpath:schema.sql

---

4. API REST Reativa

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 e 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 Repositório Reativo

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 Reativo

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. Tratamento de Erros

5.1 Exception Handler Reativo

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 Tratamento nos 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 Reativo

6.1 Configuração

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 Utilização

@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. Testando 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. Boas Práticas

8.1 Nunca Bloqueie

// Ruim - Bloqueia a thread
String result = mono.block();

// Bom - Retorna o Mono
return mono;

8.2 Use subscribeOn para Operações Bloqueantes

// Se você precisar chamar código bloqueante
Mono.fromCallable(() -> blockingOperation())
    .subscribeOn(Schedulers.boundedElastic())

8.3 Logging com doOn

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

---

9. Resumo

Conceito	Descrição
Mono	0 ou 1 elemento assíncrono
Flux	0 a N elementos assíncronos
WebFlux	Framework web reativo do Spring
R2DBC	Acesso reativo a dados
WebClient	Cliente HTTP reativo
StepVerifier	Teste de publishers

Quando Usar WebFlux

• Alta concorrência (milhares de conexões simultâneas)
• Streaming de dados
• Aplicações em tempo real
• Integração com muitos serviços externos

Quando Permanecer com Spring MVC

• Aplicações CRUD simples
• Equipe não familiarizada com reativo
• Bibliotecas de terceiros bloqueantes
• Debugging simples necessário

---

Próximo Capítulo: Cache com Redis

Voltar ao Módulo 14