Reactive WebFlux

WebFlux Controller

package com.example.presentation.rest;

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

    private final UserService userService;

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

    @GetMapping("/{id}")
    public Mono getUserById(@PathVariable Long id) {
        return userService.findById(id);
    }

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono createUser(@RequestBody @Valid UserRequest request) {
        return userService.create(request);
    }

    @PutMapping("/{id}")
    public Mono updateUser(
        @PathVariable Long id,
        @RequestBody @Valid UserRequest request
    ) {
        return userService.update(id, request);
    }

    @DeleteMapping("/{id}")
    @ResponseStatus(HttpStatus.NO_CONTENT)
    public Mono deleteUser(@PathVariable Long id) {
        return userService.delete(id);
    }
}

Reactive Service Layer

package com.example.application.service;

@Service
@RequiredArgsConstructor
public class UserService {

    private final UserRepository userRepository;
    private final UserMapper userMapper;

    public Flux findAll() {
        return userRepository.findAll()
            .map(userMapper::toResponse);
    }

    public Mono findById(Long id) {
        return userRepository.findById(id)
            .map(userMapper::toResponse)
            .switchIfEmpty(Mono.error(
                new EntityNotFoundException("User not found: " + id)
            ));
    }

    @Transactional
    public Mono create(UserRequest request) {
        return Mono.just(request)
            .map(userMapper::toEntity)
            .flatMap(userRepository::save)
            .map(userMapper::toResponse);
    }

    @Transactional
    public Mono update(Long id, UserRequest request) {
        return userRepository.findById(id)
            .switchIfEmpty(Mono.error(
                new EntityNotFoundException("User not found: " + id)
            ))
            .flatMap(existing -> {
                userMapper.updateEntity(request, existing);
                return userRepository.save(existing);
            })
            .map(userMapper::toResponse);
    }

    @Transactional
    public Mono delete(Long id) {
        return userRepository.findById(id)
            .switchIfEmpty(Mono.error(
                new EntityNotFoundException("User not found: " + id)
            ))
            .flatMap(userRepository::delete);
    }
}

R2DBC Repository

package com.example.domain.repository;

public interface UserRepository extends R2dbcRepository<User, Long> {

    Mono findByEmail(String email);

    Flux findByActiveTrue();

    @Query("""
        SELECT u.* FROM users u
        WHERE u.email LIKE CONCAT('%', :domain, '%')
        ORDER BY u.created_at DESC
        """)
    Flux findByEmailDomain(String domain);

    @Query("""
        SELECT COUNT(*) FROM users
        WHERE created_at > :since
        """)
    Mono countCreatedSince(Instant since);
}

R2DBC Entity

package com.example.domain.model;

@Table("users")
public record User(
    @Id Long id,
    String email,
    String username,
    Boolean active,
    @CreatedDate Instant createdAt,
    @LastModifiedDate Instant updatedAt
) {
    public User withId(Long id) {
        return new User(id, email, username, active, createdAt, updatedAt);
    }

    public User withEmail(String email) {
        return new User(id, email, username, active, createdAt, updatedAt);
    }
}

R2DBC Configuration

spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/demo
    username: demo
    password: demo
    pool:
      initial-size: 10
      max-size: 20
      max-idle-time: 30m

  data:
    r2dbc:
      repositories:
        enabled: true

WebClient for External APIs

package com.example.infrastructure.client;

@Component
@RequiredArgsConstructor
public class ExternalUserClient {

    private final WebClient webClient;

    public Mono getUser(Long id) {
        return webClient
            .get()
            .uri("/users/{id}", id)
            .retrieve()
            .bodyToMono(ExternalUserDto.class)
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
            .timeout(Duration.ofSeconds(5));
    }

    public Mono createUser(ExternalUserDto user) {
        return webClient
            .post()
            .uri("/users")
            .bodyValue(user)
            .retrieve()
            .bodyToMono(ExternalUserDto.class);
    }
}

@Configuration
class WebClientConfig {

    @Bean
    public WebClient webClient(WebClient.Builder builder) {
        return builder
            .baseUrl("https://api.example.com")
            .defaultHeader("User-Agent", "Demo Service")
            .build();
    }
}

Reactor Operators

// Transform data
Mono mono = Mono.just("hello")
    .map(String::toUpperCase)
    .map(s -> s + "!")
    .defaultIfEmpty("empty");

// Chain async operations
Mono result = userRepository.findById(id)
    .flatMap(user -> orderRepository.findByUserId(user.id())
        .collectList()
        .map(orders -> new UserResponse(user, orders))
    );

// Combine multiple sources
Mono combined = Mono.zip(
    userService.getUser(id),
    addressService.getAddress(id),
    (user, address) -> new UserDetails(user, address)
);

// Error handling
Mono safe = userRepository.findById(id)
    .onErrorResume(DatabaseException.class, e ->
        cacheRepository.findById(id)
    )
    .doOnError(e -> log.error("Failed to fetch user", e));

// Backpressure
Flux stream = dataRepository.findAll()
    .buffer(100)  // Process in batches
    .flatMap(batch -> processBatch(batch), 5);  // Max 5 concurrent

Testing Reactive Code

package com.example.application.service;

@ExtendWith(MockitoExtension.class)
class UserServiceTest {

    @Mock
    private UserRepository userRepository;

    @InjectMocks
    private UserService userService;

    @Test
    void shouldFindUserById() {
        User user = new User(1L, "test@example.com", "testuser", true, null, null);
        when(userRepository.findById(1L)).thenReturn(Mono.just(user));

        StepVerifier.create(userService.findById(1L))
            .expectNextMatches(response ->
                response.email().equals("test@example.com")
            )
            .verifyComplete();
    }

    @Test
    void shouldThrowWhenUserNotFound() {
        when(userRepository.findById(1L)).thenReturn(Mono.empty());

        StepVerifier.create(userService.findById(1L))
            .expectError(EntityNotFoundException.class)
            .verify();
    }
}

Quick Reference