google-site-verification: google61fe8ba583a51912.html
MACRO NEPAL
Functional Endpoints in WebFlux
awiskar acharya
November 4, 2025
3:17 pm

Functional endpoints in Spring WebFlux provide an alternative to annotation-based controllers using a functional programming style. They are particularly useful for reactive programming and offer more control over request handling.

1. Basic Setup

Dependencies (Maven)

<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId> </dependency> </dependencies>

Main Application Class

@SpringBootApplication public class WebFluxFunctionalApplication { public static void main(String[] args) { SpringApplication.run(WebFluxFunctionalApplication.class, args); } @Bean public RouterFunction<ServerResponse> routes(UserHandler userHandler) { return userHandler.routes(); } }

2. Core Components

HandlerFunction

@FunctionalInterface public interface HandlerFunction<T extends ServerResponse> { Mono<T> handle(ServerRequest request); }

RouterFunction

@FunctionalInterface public interface RouterFunction<T extends ServerResponse> { Mono<HandlerFunction<T>> route(ServerRequest request); }

3. Basic Functional Endpoint

Simple Handler

@Component public class HelloHandler { public RouterFunction<ServerResponse> routes() { return RouterFunctions.route() .GET("/hello", this::hello) .GET("/hello/{name}", this::helloName) .POST("/hello", this::createHello) .build(); } public Mono<ServerResponse> hello(ServerRequest request) { return ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .bodyValue(Map.of("message", "Hello World!")); } public Mono<ServerResponse> helloName(ServerRequest request) { String name = request.pathVariable("name"); return ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .bodyValue(Map.of("message", "Hello " + name + "!")); } public Mono<ServerResponse> createHello(ServerRequest request) { return request.bodyToMono(Map.class) .flatMap(body -> { String message = (String) body.get("message"); return ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .bodyValue(Map.of("received", message)); }); } }

4. CRUD Operations with Reactive Repository

Entity Class

@Data @AllArgsConstructor @NoArgsConstructor @Document(collection = "users") public class User { @Id private String id; private String name; private String email; private int age; private LocalDateTime createdAt; }

Reactive Repository

public interface UserRepository extends ReactiveMongoRepository<User, String> { Flux<User> findByName(String name); Mono<User> findByEmail(String email); }

User Handler with Full CRUD

@Component public class UserHandler { private final UserRepository userRepository; private final Validator validator; public UserHandler(UserRepository userRepository, Validator validator) { this.userRepository = userRepository; this.validator = validator; } public RouterFunction<ServerResponse> routes() { return RouterFunctions.route() .GET("/users", this::getAllUsers) .GET("/users/{id}", this::getUserById) .GET("/users/name/{name}", this::getUsersByName) .POST("/users", this::createUser) .PUT("/users/{id}", this::updateUser) .DELETE("/users/{id}", this::deleteUser) .filter(this::errorHandlingFilter) .build(); } // GET all users public Mono<ServerResponse> getAllUsers(ServerRequest request) { Flux<User> users = userRepository.findAll(); return ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .body(users, User.class); } // GET user by ID public Mono<ServerResponse> getUserById(ServerRequest request) { String id = request.pathVariable("id"); return userRepository.findById(id) .flatMap(user -> ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .bodyValue(user)) .switchIfEmpty(ServerResponse.notFound().build()); } // GET users by name public Mono<ServerResponse> getUsersByName(ServerRequest request) { String name = request.pathVariable("name"); Flux<User> users = userRepository.findByName(name); return ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .body(users, User.class); } // POST create user public Mono<ServerResponse> createUser(ServerRequest request) { return request.bodyToMono(User.class) .doOnNext(this::validate) .flatMap(user -> { user.setId(null); // Ensure new ID is generated user.setCreatedAt(LocalDateTime.now()); return userRepository.save(user); }) .flatMap(savedUser -> ServerResponse .created(URI.create("/users/" + savedUser.getId())) .contentType(MediaType.APPLICATION_JSON) .bodyValue(savedUser)) .onErrorResume(ValidationException.class, e -> ServerResponse.badRequest() .bodyValue(Map.of("error", e.getMessage()))); } // PUT update user public Mono<ServerResponse> updateUser(ServerRequest request) { String id = request.pathVariable("id"); return userRepository.findById(id) .zipWith(request.bodyToMono(User.class)) .flatMap(tuple -> { User existingUser = tuple.getT1(); User updatedUser = tuple.getT2(); existingUser.setName(updatedUser.getName()); existingUser.setEmail(updatedUser.getEmail()); existingUser.setAge(updatedUser.getAge()); return userRepository.save(existingUser); }) .flatMap(savedUser -> ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .bodyValue(savedUser)) .switchIfEmpty(ServerResponse.notFound().build()); } // DELETE user public Mono<ServerResponse> deleteUser(ServerRequest request) { String id = request.pathVariable("id"); return userRepository.existsById(id) .flatMap(exists -> { if (exists) { return userRepository.deleteById(id) .then(ServerResponse.noContent().build()); } else { return ServerResponse.notFound().build(); } }); } private void validate(User user) { Set<ConstraintViolation<User>> violations = validator.validate(user); if (!violations.isEmpty()) { String message = violations.stream() .map(ConstraintViolation::getMessage) .collect(Collectors.joining(", ")); throw new ValidationException(message); } } // Error handling filter private Mono<ServerResponse> errorHandlingFilter(ServerRequest request, HandlerFunction<ServerResponse> next) { return next.handle(request) .onErrorResume(ValidationException.class, e -> ServerResponse.badRequest() .bodyValue(Map.of("error", e.getMessage()))) .onErrorResume(DataAccessException.class, e -> ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR) .bodyValue(Map.of("error", "Database error"))) .onErrorResume(Exception.class, e -> ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR) .bodyValue(Map.of("error", "Internal server error"))); } }

5. Advanced Routing with Nested Routes

Nested Router Configuration

@Component public class ApiRouter { private final UserHandler userHandler; private final ProductHandler productHandler; public ApiRouter(UserHandler userHandler, ProductHandler productHandler) { this.userHandler = userHandler; this.productHandler = productHandler; } @Bean public RouterFunction<ServerResponse> apiRoutes() { return RouterFunctions.route() .path("/api", builder -> builder .nest(accept(MediaType.APPLICATION_JSON), nestedBuilder -> nestedBuilder .path("/v1", this::v1Routes) .path("/v2", this::v2Routes) ) ) .build(); } private RouterFunction<ServerResponse> v1Routes() { return RouterFunctions.route() .path("/users", () -> userHandler.routes()) .path("/products", () -> productHandler.routes()) .GET("/health", request -> ServerResponse.ok().bodyValue("API v1 is healthy")) .build(); } private RouterFunction<ServerResponse> v2Routes() { return RouterFunctions.route() .path("/users", () -> userHandler.routes()) .path("/products", () -> productHandler.routes()) .GET("/health", request -> ServerResponse.ok().bodyValue("API v2 is healthy")) .add(userHandler.advancedRoutes()) // Additional v2 specific routes .build(); } }

6. Request Predicates and Filters

Custom Request Predicates

@Component public class SecurityHandler { public RouterFunction<ServerResponse> securedRoutes() { return RouterFunctions.route() .GET("/admin/users", RequestPredicates.GET("/admin/users") .and(RequestPredicates.accept(MediaType.APPLICATION_JSON)) .and(this::isAdmin), this::getAdminUsers) .GET("/profile", RequestPredicates.GET("/profile").and(this::isAuthenticated), this::getUserProfile) .build(); } private boolean isAdmin(ServerRequest request) { // Check admin role from headers or tokens String role = request.headers().firstHeader("X-User-Role"); return "admin".equals(role); } private boolean isAuthenticated(ServerRequest request) { String authHeader = request.headers().firstHeader("Authorization"); return authHeader != null && authHeader.startsWith("Bearer "); } public Mono<ServerResponse> getAdminUsers(ServerRequest request) { // Return admin-only user data return ServerResponse.ok().bodyValue(Map.of("message", "Admin users data")); } public Mono<ServerResponse> getUserProfile(ServerRequest request) { String userId = extractUserId(request); return ServerResponse.ok().bodyValue(Map.of("userId", userId, "profile", "user data")); } private String extractUserId(ServerRequest request) { // Extract user ID from token return "user123"; } }

Advanced Filtering

@Component public class LoggingFilter { public HandlerFilterFunction<ServerResponse, ServerResponse> logging() { return (request, next) -> { long startTime = System.currentTimeMillis(); String path = request.path(); String method = request.methodName(); System.out.println("Started: " + method + " " + path); return next.handle(request) .doOnSuccess(response -> { long duration = System.currentTimeMillis() - startTime; System.out.println("Completed: " + method + " " + path + " - Status: " + response.statusCode() + " - Duration: " + duration + "ms"); }) .doOnError(error -> { long duration = System.currentTimeMillis() - startTime; System.out.println("Failed: " + method + " " + path + " - Error: " + error.getMessage() + " - Duration: " + duration + "ms"); }); }; } public HandlerFilterFunction<ServerResponse, ServerResponse> rateLimiting() { return (request, next) -> { String clientIp = getClientIp(request); // Implement rate limiting logic here if (isRateLimited(clientIp)) { return ServerResponse.status(HttpStatus.TOO_MANY_REQUESTS) .bodyValue(Map.of("error", "Rate limit exceeded")); } return next.handle(request); }; } private String getClientIp(ServerRequest request) { // Extract client IP from request return request.remoteAddress() .map(addr -> addr.getAddress().getHostAddress()) .orElse("unknown"); } private boolean isRateLimited(String clientIp) { // Implement rate limiting logic return false; } }

7. Streaming and SSE (Server-Sent Events)

Streaming Handler

@Component public class StreamHandler { public RouterFunction<ServerResponse> streamRoutes() { return RouterFunctions.route() .GET("/stream/numbers", this::streamNumbers) .GET("/stream/users", this::streamUsers) .GET("/events", this::serverSentEvents) .build(); } public Mono<ServerResponse> streamNumbers(ServerRequest request) { Flux<Long> numberStream = Flux.interval(Duration.ofSeconds(1)) .take(10); // Emit 10 numbers with 1-second interval return ServerResponse.ok() .contentType(MediaType.APPLICATION_NDJSON) .body(numberStream, Long.class); } public Mono<ServerResponse> streamUsers(ServerRequest request) { Flux<User> userStream = Flux.interval(Duration.ofSeconds(2)) .map(i -> new User( "id" + i, "User " + i, "user" + i + "@example.com", 20 + i.intValue(), LocalDateTime.now() )) .take(5); return ServerResponse.ok() .contentType(MediaType.APPLICATION_NDJSON) .body(userStream, User.class); } public Mono<ServerResponse> serverSentEvents(ServerRequest request) { Flux<ServerSentEvent<Object>> eventStream = Flux.interval(Duration.ofSeconds(1)) .map(sequence -> ServerSentEvent.builder() .id(String.valueOf(sequence)) .event("message") .data("Event " + sequence + " at " + LocalDateTime.now()) .build()) .take(10); return ServerResponse.ok() .contentType(MediaType.TEXT_EVENT_STREAM) .body(eventStream, ServerSentEvent.class); } }

8. File Upload and Download

File Handler

@Component public class FileHandler { private final Path uploadDir = Paths.get("uploads"); public FileHandler() { try { Files.createDirectories(uploadDir); } catch (IOException e) { throw new RuntimeException("Could not create upload directory", e); } } public RouterFunction<ServerResponse> fileRoutes() { return RouterFunctions.route() .POST("/upload", this::uploadFile) .GET("/download/{filename}", this::downloadFile) .GET("/files", this::listFiles) .build(); } public Mono<ServerResponse> uploadFile(ServerRequest request) { return request.multipartData() .flatMap(parts -> { FilePart filePart = (FilePart) parts.toSingleValueMap().get("file"); if (filePart == null) { return Mono.error(new IllegalArgumentException("File not provided")); } Path filePath = uploadDir.resolve(filePart.filename()); return filePart.transferTo(filePath) .then(ServerResponse.ok() .bodyValue(Map.of( "message", "File uploaded successfully", "filename", filePart.filename(), "size", filePart.headers().getContentLength() ))); }); } public Mono<ServerResponse> downloadFile(ServerRequest request) { String filename = request.pathVariable("filename"); Path filePath = uploadDir.resolve(filename); if (!Files.exists(filePath)) { return ServerResponse.notFound().build(); } Resource resource = new FileSystemResource(filePath); return ServerResponse.ok() .contentType(MediaType.APPLICATION_OCTET_STREAM) .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"" + filename + "\"") .bodyValue(resource); } public Mono<ServerResponse> listFiles(ServerRequest request) { try { Flux<String> files = Flux.fromStream(Files.list(uploadDir)) .map(path -> path.getFileName().toString()); return ServerResponse.ok() .contentType(MediaType.APPLICATION_JSON) .body(files, String.class); } catch (IOException e) { return ServerResponse.status(HttpStatus.INTERNAL_SERVER_ERROR) .bodyValue(Map.of("error", "Could not list files")); } } }

9. Testing Functional Endpoints

Test Dependencies

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-test</artifactId> <scope>test</scope> </dependency>

WebTestClient Tests

@SpringBootTest @AutoConfigureWebTestClient class UserHandlerTest { @Autowired private WebTestClient webTestClient; @MockBean private UserRepository userRepository; @Test void shouldGetAllUsers() { User user1 = new User("1", "John", "[email protected]", 25, LocalDateTime.now()); User user2 = new User("2", "Jane", "[email protected]", 30, LocalDateTime.now()); when(userRepository.findAll()).thenReturn(Flux.just(user1, user2)); webTestClient.get().uri("/users") .exchange() .expectStatus().isOk() .expectBodyList(User.class) .hasSize(2) .contains(user1, user2); } @Test void shouldCreateUser() { User newUser = new User(null, "John", "[email protected]", 25, null); User savedUser = new User("1", "John", "[email protected]", 25, LocalDateTime.now()); when(userRepository.save(any(User.class))).thenReturn(Mono.just(savedUser)); webTestClient.post().uri("/users") .contentType(MediaType.APPLICATION_JSON) .bodyValue(newUser) .exchange() .expectStatus().isCreated() .expectBody(User.class) .isEqualTo(savedUser); } @Test void shouldReturnNotFoundForNonExistentUser() { when(userRepository.findById("999")).thenReturn(Mono.empty()); webTestClient.get().uri("/users/999") .exchange() .expectStatus().isNotFound(); } }

10. Configuration and Best Practices

Application Configuration

# application.yml spring: data: mongodb: uri: mongodb://localhost:27017/webflux webflux: base-path: /api server: port: 8080

Best Practices

  1. Keep handlers focused and single-responsibility
  2. Use dependency injection for handlers
  3. Implement proper error handling
  4. Use reactive types (Mono/Flux) consistently
  5. Test with WebTestClient
  6. Use filters for cross-cutting concerns
  7. Implement proper validation
  8. Use nested routes for API versioning
  9. Implement proper logging and monitoring
  10. Use ServerSentEvent for real-time updates

Functional endpoints in WebFlux provide a powerful, flexible way to build reactive web applications with fine-grained control over request handling and routing.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper