Chapter 1: Reactive Programming with Spring WebFlux

Chapter Objectives

• Understand the reactive vs imperative paradigm
• Master the Mono and Flux types
• Create reactive REST APIs
• Use R2DBC for reactive data access
• Understand transformation operators

---

1. Introduction to Reactive Programming

1.1 The Problem with the Blocking Model

Spring MVC Model (blocking):

Thread-1: [Request]--[DB Wait 100ms]--[Response]
Thread-2: [Request]--[DB Wait 150ms]--[Response]
Thread-3: [Request]--[API Wait 200ms]--[Response]
...
Thread-200: Pool exhausted! New requests waiting...

Problem: 1 thread = 1 request = wasted resources during wait

1.2 The Reactive Model

WebFlux Model (non-blocking):

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

- Few threads (typically: number of CPU cores)
- No blocking wait
- Callbacks when data is ready
- Increased scalability

1.3 Reactive Streams Specification

The 4 basic interfaces:

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

2.1 Mono: 0 or 1 element

// Creation
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());

// Operations
Mono<String> transformed = Mono.just("hello")
    .map(String::toUpperCase)              // HELLO
    .flatMap(s -> fetchFromDb(s))          // Async call
    .defaultIfEmpty("default")             // Default value
    .onErrorReturn("error fallback");      // In case of error

2.2 Flux: 0 to N elements

// Creation
Flux<Integer> empty = Flux.empty();
Flux<Integer> just = Flux.just(1, 2, 3);
Flux<Integer> range = Flux.range(1, 10);  // 1 to 10
Flux<Integer> fromIterable = Flux.fromIterable(list);
Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));  // Infinite

// Operations
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)                               // Limit to 3
    .collectList()                         // Mono<List<String>>
    .block();                              // Block (tests only!)

2.3 Common Operators

Operator	Description	Example
map	Synchronous transformation	.map(s -> s.toUpperCase())
flatMap	Async transformation	.flatMap(id -> repository.findById(id))
filter	Filtering	.filter(n -> n > 0)
take	Limit the number	.take(5)
skip	Skip the first	.skip(2)
distinct	Unique elements	.distinct()
defaultIfEmpty	Default value	.defaultIfEmpty("N/A")
switchIfEmpty	Alternative publisher	.switchIfEmpty(Mono.just("default"))
onErrorReturn	Fallback in case of error	.onErrorReturn("error")
onErrorResume	Alternative publisher	.onErrorResume(e -> Mono.empty())
retry	Retry	.retry(3)
timeout	Time limit	.timeout(Duration.ofSeconds(5))
zip	Combine	Mono.zip(mono1, mono2)
merge	Merge in parallel	Flux.merge(flux1, flux2)
concat	Merge sequentially	Flux.concat(flux1, flux2)

---

3. Spring WebFlux Configuration

3.1 Maven Dependencies

<dependencies>
    <!-- WebFlux (replaces spring-boot-starter-web) -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    
    <!-- R2DBC for reactive data access -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-r2dbc</artifactId>
    </dependency>
    
    <!-- R2DBC PostgreSQL Driver -->
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>r2dbc-postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    
    <!-- Or H2 for 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:
  
  # Schema initialization
  sql:
    init:
      mode: always
      schema-locations: classpath:schema.sql

---

4. Reactive REST API

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 and 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 Reactive Repository

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 Reactive Controller

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. Error Handling

5.1 Reactive Exception Handler

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 Handling in Operators

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: Reactive HTTP Client

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 Usage

@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. Best Practices

8.1 Never Block

// Bad - Blocks the thread
String result = mono.block();

// Good - Returns the Mono
return mono;

8.2 Use subscribeOn for Blocking Operations

// If you must call blocking code
Mono.fromCallable(() -> blockingOperation())
    .subscribeOn(Schedulers.boundedElastic())

8.3 Logging with doOn

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

---

9. Summary

Concept	Description
Mono	0 or 1 asynchronous element
Flux	0 to N asynchronous elements
WebFlux	Spring reactive web framework
R2DBC	Reactive data access
WebClient	Reactive HTTP client
StepVerifier	Test publishers

When to Use WebFlux

• High concurrency (thousands of simultaneous connections)
• Data streaming
• Real-time applications
• Integration with numerous external services

When to Stick with Spring MVC

• Simple CRUD applications
• Team not familiar with reactive
• Blocking third-party libraries
• Simple debugging needed

---

Next Chapter: Cache with Redis

Back to Module 14