Project Reactor + WebClient + Reactive MongoDB

package com.veloz.reactor.config;

import io.netty.channel.ChannelOption;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.netty.http.client.HttpClient;
import reactor.netty.resources.ConnectionProvider;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

@Configuration
public class WebClientConfig {

    @Bean
    public WebClient webClient(WebClient.Builder builder) {
        // Connection pool
        ConnectionProvider provider = ConnectionProvider.builder("custom")
            .maxConnections(500)
            .maxIdleTime(Duration.ofSeconds(20))
            .maxLifeTime(Duration.ofSeconds(60))
            .pendingAcquireTimeout(Duration.ofSeconds(60))
            .evictInBackground(Duration.ofSeconds(120))
            .build();

        // HTTP client with timeouts
        HttpClient httpClient = HttpClient.create(provider)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
            .responseTimeout(Duration.ofSeconds(10))
            .doOnConnected(conn -> conn
                .addHandlerLast(new ReadTimeoutHandler(10, TimeUnit.SECONDS))
                .addHandlerLast(new WriteTimeoutHandler(10, TimeUnit.SECONDS)));

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

    private ExchangeFilterFunction logRequest() {
        return ExchangeFilterFunction.ofRequestProcessor(request -> {
            log.debug("Request: {} {}", request.method(), request.url());
            return Mono.just(request);
        });
    }

    private ExchangeFilterFunction logResponse() {
        return ExchangeFilterFunction.ofResponseProcessor(response -> {
            log.debug("Response: {}", response.statusCode());
            return Mono.just(response);
        });
    }
}

import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

// GET — single object
Mono<User> user = webClient.get()
    .uri("/users/{id}", userId)
    .retrieve()
    .bodyToMono(User.class);

// GET — list (Flux)
Flux<User> users = webClient.get()
    .uri(uriBuilder -> uriBuilder
        .path("/users")
        .queryParam("status", "active")
        .queryParam("page", 0)
        .queryParam("size", 20)
        .build())
    .retrieve()
    .bodyToFlux(User.class);

// POST
Mono<User> created = webClient.post()
    .uri("/users")
    .bodyValue(new CreateUserRequest("John", "[email protected]"))
    .retrieve()
    .bodyToMono(User.class);

// PUT
Mono<User> updated = webClient.put()
    .uri("/users/{id}", userId)
    .bodyValue(updateRequest)
    .retrieve()
    .bodyToMono(User.class);

// DELETE
Mono<Void> deleted = webClient.delete()
    .uri("/users/{id}", userId)
    .retrieve()
    .bodyToMono(Void.class);

Mono<User> user = webClient.get()
    .uri("/users/{id}", userId)
    .retrieve()
    .onStatus(status -> status.value() == 404,
        response -> Mono.error(new UserNotFoundException(userId)))
    .onStatus(status -> status.is4xxClientError(),
        response -> response.bodyToMono(String.class)
            .flatMap(body -> Mono.error(new ClientException(body))))
    .onStatus(status -> status.is5xxServerError(),
        response -> Mono.error(new ServiceException("Server error")))
    .bodyToMono(User.class);

// retrieve() — simple, auto-handles errors (recommended)
Mono<User> user = webClient.get()
    .uri("/users/{id}", userId)
    .retrieve()
    .bodyToMono(User.class);

// exchangeToMono() — full control over response (status, headers, body)
Mono<User> user2 = webClient.get()
    .uri("/users/{id}", userId)
    .exchangeToMono(response -> {
        if (response.statusCode().is2xxSuccessful()) {
            return response.bodyToMono(User.class);
        } else if (response.statusCode().value() == 404) {
            return Mono.empty();
        } else {
            return response.createError();
        }
    });

// Stream large responses as Flux
Flux<DataChunk> stream = webClient.get()
    .uri("/api/stream")
    .accept(MediaType.APPLICATION_NDJSON)
    .retrieve()
    .bodyToFlux(DataChunk.class);

// File download
Mono<Void> download = webClient.get()
    .uri("/files/{name}", filename)
    .retrieve()
    .bodyToMono(DataBuffer.class)
    .flatMap(dataBuffer -> DataBufferUtils
        .write(Mono.just(dataBuffer), Path.of("/tmp/" + filename))
        .then());

// File upload
Mono<String> upload = webClient.post()
    .uri("/files/upload")
    .contentType(MediaType.MULTIPART_FORM_DATA)
    .body(BodyInserters.fromMultipartData("file",
        new FileSystemResource("/tmp/upload.pdf")))
    .retrieve()
    .bodyToMono(String.class);

package com.veloz.reactor.client;

import com.veloz.reactor.dto.User;
import com.veloz.reactor.dto.CreateUserRequest;
import com.veloz.reactor.exception.UserNotFoundException;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;

@Component
public class UserServiceClient {

    private final WebClient webClient;

    public UserServiceClient(WebClient.Builder builder) {
        this.webClient = builder
            .baseUrl("https://user-service.internal")
            .build();
    }

    public Mono<User> getUser(String id) {
        return webClient.get()
            .uri("/api/v1/users/{id}", id)
            .retrieve()
            .onStatus(s -> s.value() == 404,
                r -> Mono.error(new UserNotFoundException(id)))
            .bodyToMono(User.class)
            .timeout(Duration.ofSeconds(5))
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                .filter(e -> !(e instanceof UserNotFoundException)));
    }

    public Flux<User> getAllUsers() {
        return webClient.get()
            .uri("/api/v1/users")
            .retrieve()
            .bodyToFlux(User.class)
            .timeout(Duration.ofSeconds(10));
    }

    public Mono<User> createUser(CreateUserRequest request) {
        return webClient.post()
            .uri("/api/v1/users")
            .bodyValue(request)
            .retrieve()
            .bodyToMono(User.class);
    }

    public Mono<Void> deleteUser(String id) {
        return webClient.delete()
            .uri("/api/v1/users/{id}", id)
            .retrieve()
            .bodyToMono(Void.class);
    }
}

package com.veloz.reactor.service;

import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;

@Service
public class DashboardService {

    private final WebClient userClient;
    private final WebClient orderClient;
    private final WebClient inventoryClient;

    public DashboardService(WebClient.Builder builder) {
        this.userClient = builder.baseUrl("https://user-service").build();
        this.orderClient = builder.baseUrl("https://order-service").build();
        this.inventoryClient = builder.baseUrl("https://inventory-service").build();
    }

    public Mono<DashboardDto> getDashboard(String userId) {
        Mono<UserProfile> profile = userClient.get()
            .uri("/users/{id}/profile", userId)
            .retrieve().bodyToMono(UserProfile.class);

        Mono<List<Order>> orders = orderClient.get()
            .uri("/orders?userId={id}", userId)
            .retrieve().bodyToFlux(Order.class)
            .collectList();

        Mono<InventorySummary> inventory = inventoryClient.get()
            .uri("/inventory/summary")
            .retrieve().bodyToMono(InventorySummary.class);

        // All 3 calls execute IN PARALLEL, combine when all complete
        return Mono.zip(profile, orders, inventory)
            .map(tuple -> new DashboardDto(
                tuple.getT1(),  // UserProfile
                tuple.getT2(),  // List<Order>
                tuple.getT3()   // InventorySummary
            ));
    }
}