Spring WebFlux

Full Reference: See advanced.md for SSE with Sinks, Testing with StepVerifier, and Context Propagation patterns.

Quick Start

@RestController @RequestMapping("/api/users") @RequiredArgsConstructor public class UserController {

private final UserService userService;

@GetMapping("/{id}")
public Mono<UserResponse> getUser(@PathVariable Long id) {
    return userService.findById(id);
}

@GetMapping
public Flux<UserResponse> getAllUsers() {
    return userService.findAll();
}

@PostMapping
@ResponseStatus(HttpStatus.CREATED)
public Mono<UserResponse> createUser(@RequestBody Mono<CreateUserRequest> request) {
    return request.flatMap(userService::create);
}

}

Mono & Flux Basics

Mono (0 or 1 element)

Mono<String> empty = Mono.empty(); Mono<String> just = Mono.just("Hello"); Mono<String> fromCallable = Mono.fromCallable(() -> expensiveOperation()); Mono<String> defer = Mono.defer(() -> Mono.just(dynamicValue())); Mono<String> fromOptional = Mono.justOrEmpty(optionalValue); Mono<String> fromFuture = Mono.fromFuture(completableFuture);

Flux (0 to N elements)

Flux<Integer> just = Flux.just(1, 2, 3); Flux<Integer> fromIterable = Flux.fromIterable(List.of(1, 2, 3)); Flux<Integer> range = Flux.range(1, 10); Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));

Flux<Integer> generate = Flux.generate( () -> 0, (state, sink) -> { sink.next(state); if (state == 10) sink.complete(); return state + 1; } );

Reactive Operators

Transformation

// map - transform each element users.map(user -> new UserResponse(user.getId(), user.getName()));

// flatMap - async transformation (parallel) users.flatMap(user -> orderRepository.findByUserId(user.getId()));

// flatMapSequential - maintains order users.flatMapSequential(user -> orderRepository.findByUserId(user.getId()));

// concatMap - sequential, one at a time users.concatMap(user -> orderRepository.findByUserId(user.getId()));

// switchMap - cancels previous when new arrives searchTerms.switchMap(term -> searchService.search(term));

Filtering

// filter users.filter(user -> user.getStatus() == Status.ACTIVE);

// filterWhen - async filter users.filterWhen(user -> permissionService.canAccess(user.getId()));

// distinct / distinctUntilChanged items.distinct(); values.distinctUntilChanged();

// take / skip users.skip((long) page * size).take(size);

Combining

// zip - combine by position Flux.zip(users, orders, UserWithOrders::new);

// merge - interleave from multiple sources Flux.merge(source1, source2);

// concat - sequential Flux.concat(first, second);

// zipWith on Mono userRepository.findById(userId) .zipWith(profileRepository.findByUserId(userId)) .map(tuple -> new UserWithProfile(tuple.getT1(), tuple.getT2()));

Error Handling

// onErrorReturn - default value on error userRepository.findById(id).onErrorReturn(new User("default"));

// onErrorResume - fallback Publisher primaryRepository.findById(id) .onErrorResume(e -> fallbackRepository.findById(id));

// onErrorResume with specific type userRepository.findById(id) .onErrorResume(NotFoundException.class, e -> Mono.empty()) .onErrorResume(TimeoutException.class, e -> cacheRepository.findById(id));

// onErrorMap - transform exception userRepository.findById(id) .onErrorMap(DataAccessException.class, e -> new ServiceException("Database error", e));

// retryWhen - advanced retry userRepository.findById(id) .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)) .filter(e -> e instanceof TransientException));

Side Effects

userRepository.findById(id) .doOnSubscribe(s -> log.info("Subscribed")) .doOnNext(user -> log.info("Found user: {}", user.getId())) .doOnError(e -> log.error("Error: {}", e.getMessage())) .doFinally(signalType -> log.info("Finally: {}", signalType));

WebClient

Configuration

@Configuration public class WebClientConfig {

@Bean
public WebClient webClient(WebClient.Builder builder) {
    HttpClient httpClient = HttpClient.create()
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
        .responseTimeout(Duration.ofSeconds(30))
        .doOnConnected(conn -> conn
            .addHandlerLast(new ReadTimeoutHandler(30, TimeUnit.SECONDS))
            .addHandlerLast(new WriteTimeoutHandler(10, TimeUnit.SECONDS)));

    return builder
        .baseUrl("https://api.example.com")
        .clientConnector(new ReactorClientHttpConnector(httpClient))
        .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
        .build();
}

}

Usage

@Service @RequiredArgsConstructor public class ExternalApiService {

private final WebClient webClient;

public Mono<UserDto> getUser(Long id) {
    return webClient.get()
        .uri("/users/{id}", id)
        .retrieve()
        .bodyToMono(UserDto.class);
}

public Mono<UserDto> getUserSafe(Long id) {
    return webClient.get()
        .uri("/users/{id}", id)
        .retrieve()
        .onStatus(HttpStatusCode::is4xxClientError, response ->
            response.bodyToMono(ErrorResponse.class)
                .flatMap(error -> Mono.error(new ClientException(error.getMessage()))))
        .onStatus(HttpStatusCode::is5xxServerError, response ->
            Mono.error(new ServerException("Server error")))
        .bodyToMono(UserDto.class);
}

// Parallel calls
public Mono<AggregatedData> getAggregatedData(Long userId) {
    return Mono.zip(
        getUser(userId),
        getOrders(userId).collectList(),
        getProfile(userId)
    ).map(tuple -> new AggregatedData(tuple.getT1(), tuple.getT2(), tuple.getT3()));
}

}

Functional Endpoints

@Configuration public class RouterConfig {

@Bean
public RouterFunction<ServerResponse> userRoutes(UserHandler handler) {
    return RouterFunctions.route()
        .path("/api/users", builder -> builder
            .GET("", handler::getAll)
            .GET("/{id}", handler::getById)
            .POST("", handler::create)
            .PUT("/{id}", handler::update)
            .DELETE("/{id}", handler::delete)
        )
        .build();
}

}

@Component @RequiredArgsConstructor public class UserHandler {

private final UserService userService;

public Mono<ServerResponse> getById(ServerRequest request) {
    Long id = Long.parseLong(request.pathVariable("id"));
    return userService.findById(id)
        .flatMap(user -> ServerResponse.ok()
            .contentType(MediaType.APPLICATION_JSON)
            .bodyValue(user))
        .switchIfEmpty(ServerResponse.notFound().build());
}

public Mono<ServerResponse> create(ServerRequest request) {
    return request.bodyToMono(CreateUserRequest.class)
        .flatMap(userService::create)
        .flatMap(user -> ServerResponse.created(
                URI.create("/api/users/" + user.getId()))
            .bodyValue(user));
}

}

Best Practices

Do Don't

Keep chain fully reactive Use .block() in handlers

Use appropriate operators (flatMap vs concatMap) Mix blocking and reactive

Handle errors with onError* operators Ignore errors

Use StepVerifier for testing Test with .block()

Propagate Context for MDC/security Use ThreadLocal

Production Checklist

• Timeout configured on WebClient

• Error handling complete

• Retry logic for transient errors

• Backpressure strategy defined

• Context propagation for logging

• Reactive metrics

• Test with StepVerifier

When NOT to Use This Skill

• Traditional blocking apps - Use spring-web skill

• Simple CRUD APIs - Use spring-rest skill

• CPU-bound workloads - Reactive doesn't help here

• Team unfamiliar with reactive - Learning curve is steep

Anti-Patterns

Anti-Pattern Problem Solution

Nothing executes Publisher not subscribed Ensure subscribe/return from controller

Blocking call .block() in reactive chain Avoid block, use operators

Context lost MDC not propagated Use Context propagation

Memory leak Infinite Flux without backpressure Use backpressure operators

Cold vs Hot confusion Publisher recreated every subscribe Use .share() or .cache()

Quick Troubleshooting

Problem Diagnostic Fix

Mono/Flux never completes Check for missing subscribe Return from controller

Context not available Check propagation Use Hooks.enableAutomaticContextPropagation()

Backpressure overflow Check buffer size Use onBackpressure* operators

Test times out Check StepVerifier usage Use virtual time for delays

Memory keeps growing Check for leaks Use .limitRate() or .take()

Reference Documentation

• Spring WebFlux

• Project Reactor

Related Skills

• spring-r2dbc

• For reactive database access

• spring-web

• For comparison with MVC

• spring-websocket

• For reactive WebSocket